260
class TestAuthHeader(tests.TestCase):
262
def parse_header(self, header, auth_handler_class=None):
263
if auth_handler_class is None:
264
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
265
self.auth_handler = auth_handler_class()
266
return self.auth_handler._parse_auth_header(header)
268
def test_empty_header(self):
269
scheme, remainder = self.parse_header('')
270
self.assertEqual('', scheme)
271
self.assertIs(None, remainder)
273
def test_negotiate_header(self):
274
scheme, remainder = self.parse_header('Negotiate')
275
self.assertEqual('negotiate', scheme)
276
self.assertIs(None, remainder)
278
def test_basic_header(self):
279
scheme, remainder = self.parse_header(
280
'Basic realm="Thou should not pass"')
281
self.assertEqual('basic', scheme)
282
self.assertEqual('realm="Thou should not pass"', remainder)
284
def test_basic_extract_realm(self):
285
scheme, remainder = self.parse_header(
286
'Basic realm="Thou should not pass"',
287
_urllib2_wrappers.BasicAuthHandler)
288
match, realm = self.auth_handler.extract_realm(remainder)
289
self.assertTrue(match is not None)
290
self.assertEqual('Thou should not pass', realm)
292
def test_digest_header(self):
293
scheme, remainder = self.parse_header(
294
'Digest realm="Thou should not pass"')
295
self.assertEqual('digest', scheme)
296
self.assertEqual('realm="Thou should not pass"', remainder)
299
class TestHTTPServer(tests.TestCase):
300
"""Test the HTTP servers implementations."""
302
def test_invalid_protocol(self):
303
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
305
protocol_version = 'HTTP/0.1'
307
server = http_server.HttpServer(BogusRequestHandler)
309
self.assertRaises(httplib.UnknownProtocol, server.start_server)
312
self.fail('HTTP Server creation did not raise UnknownProtocol')
314
def test_force_invalid_protocol(self):
315
server = http_server.HttpServer(protocol_version='HTTP/0.1')
317
self.assertRaises(httplib.UnknownProtocol, server.start_server)
320
self.fail('HTTP Server creation did not raise UnknownProtocol')
322
def test_server_start_and_stop(self):
323
server = http_server.HttpServer()
324
server.start_server()
326
self.assertTrue(server._http_running)
329
self.assertFalse(server._http_running)
331
def test_create_http_server_one_zero(self):
332
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
334
protocol_version = 'HTTP/1.0'
336
server = http_server.HttpServer(RequestHandlerOneZero)
337
self.start_server(server)
338
self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
340
def test_create_http_server_one_one(self):
341
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
343
protocol_version = 'HTTP/1.1'
345
server = http_server.HttpServer(RequestHandlerOneOne)
346
self.start_server(server)
347
self.assertIsInstance(server._httpd,
348
http_server.TestingThreadingHTTPServer)
350
def test_create_http_server_force_one_one(self):
351
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
353
protocol_version = 'HTTP/1.0'
355
server = http_server.HttpServer(RequestHandlerOneZero,
356
protocol_version='HTTP/1.1')
357
self.start_server(server)
358
self.assertIsInstance(server._httpd,
359
http_server.TestingThreadingHTTPServer)
361
def test_create_http_server_force_one_zero(self):
362
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
364
protocol_version = 'HTTP/1.1'
366
server = http_server.HttpServer(RequestHandlerOneOne,
367
protocol_version='HTTP/1.0')
368
self.start_server(server)
369
self.assertIsInstance(server._httpd,
370
http_server.TestingHTTPServer)
373
class TestWithTransport_pycurl(object):
374
"""Test case to inherit from if pycurl is present"""
376
def _get_pycurl_maybe(self):
377
self.requireFeature(features.pycurl)
378
return PyCurlTransport
380
_transport = property(_get_pycurl_maybe)
383
class TestHttpUrls(tests.TestCase):
385
# TODO: This should be moved to authorization tests once they
129
class TestHttpUrls(TestCase):
131
# FIXME: Some of these tests should be done for both
388
134
def test_url_parsing(self):
389
135
f = FakeManager()
390
url = http.extract_auth('http://example.com', f)
391
self.assertEqual('http://example.com', url)
392
self.assertEqual(0, len(f.credentials))
393
url = http.extract_auth(
394
'http://user:pass@example.com/bzr/bzr.dev', f)
395
self.assertEqual('http://example.com/bzr/bzr.dev', url)
396
self.assertEqual(1, len(f.credentials))
397
self.assertEqual([None, 'example.com', 'user', 'pass'],
401
class TestHttpTransportUrls(tests.TestCase):
402
"""Test the http urls."""
136
url = extract_auth('http://example.com', f)
137
self.assertEquals('http://example.com', url)
138
self.assertEquals(0, len(f.credentials))
139
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
140
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
141
self.assertEquals(1, len(f.credentials))
142
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
404
145
def test_abs_url(self):
405
146
"""Construction of absolute http URLs"""
406
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
147
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
407
148
eq = self.assertEqualDiff
408
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
409
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
410
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
150
'http://bazaar-vcs.org/bzr/bzr.dev')
151
eq(t.abspath('foo/bar'),
152
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
153
eq(t.abspath('.bzr'),
154
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
411
155
eq(t.abspath('.bzr/1//2/./3'),
412
156
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
414
158
def test_invalid_http_urls(self):
415
159
"""Trap invalid construction of urls"""
416
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
417
self.assertRaises(errors.InvalidURL,
419
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
160
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
161
self.assertRaises(ValueError,
164
t = HttpTransport_urllib('http://http://bazaar-vcs.org/bzr/bzr.dev/')
165
self.assertRaises(errors.InvalidURL, t.has, 'foo/bar')
421
167
def test_http_root_urls(self):
422
168
"""Construction of URLs from server root"""
423
t = self._transport('http://bzr.ozlabs.org/')
169
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
424
170
eq = self.assertEqualDiff
425
171
eq(t.abspath('.bzr/tree-version'),
426
172
'http://bzr.ozlabs.org/.bzr/tree-version')
428
174
def test_http_impl_urls(self):
429
175
"""There are servers which ask for particular clients to connect"""
430
server = self._server()
431
server.start_server()
176
server = HttpServer_PyCurl()
433
179
url = server.get_url()
434
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
180
self.assertTrue(url.startswith('http+pycurl://'))
439
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
441
# TODO: This should really be moved into another pycurl
442
# specific test. When https tests will be implemented, take
443
# this one into account.
444
def test_pycurl_without_https_support(self):
445
"""Test that pycurl without SSL do not fail with a traceback.
447
For the purpose of the test, we force pycurl to ignore
448
https by supplying a fake version_info that do not
451
self.requireFeature(features.pycurl)
452
# Import the module locally now that we now it's available.
453
pycurl = features.pycurl.module
455
self.overrideAttr(pycurl, 'version_info',
456
# Fake the pycurl version_info This was taken from
457
# a windows pycurl without SSL (thanks to bialix)
466
('ftp', 'gopher', 'telnet',
467
'dict', 'ldap', 'http', 'file'),
471
self.assertRaises(errors.DependencyNotPresent, self._transport,
472
'https://launchpad.net')
475
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
476
"""Test the http connections."""
185
class TestHttpConnections(object):
186
"""Test the http connections.
188
This MUST be used by daughter classes that also inherit from
189
TestCaseWithWebserver.
191
We can't inherit directly from TestCaseWithWebserver or the
192
test framework will try to create an instance which cannot
193
run, its implementation being incomplete.
479
http_utils.TestCaseWithWebserver.setUp(self)
480
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
197
TestCaseWithWebserver.setUp(self)
198
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
481
199
transport=self.get_transport())
483
201
def test_http_has(self):
823
595
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
824
596
t.readv, 'a', [(12,2)])
826
def test_readv_multiple_get_requests(self):
827
server = self.get_readonly_server()
828
t = self._transport(server.get_url())
829
# force transport to issue multiple requests
830
t._max_readv_combine = 1
831
t._max_get_ranges = 1
832
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
833
self.assertEqual(l[0], (0, '0'))
834
self.assertEqual(l[1], (1, '1'))
835
self.assertEqual(l[2], (3, '34'))
836
self.assertEqual(l[3], (9, '9'))
837
# The server should have issued 4 requests
838
self.assertEqual(4, server.GET_request_nb)
840
def test_readv_get_max_size(self):
841
server = self.get_readonly_server()
842
t = self._transport(server.get_url())
843
# force transport to issue multiple requests by limiting the number of
844
# bytes by request. Note that this apply to coalesced offsets only, a
845
# single range will keep its size even if bigger than the limit.
847
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
848
self.assertEqual(l[0], (0, '0'))
849
self.assertEqual(l[1], (1, '1'))
850
self.assertEqual(l[2], (2, '2345'))
851
self.assertEqual(l[3], (6, '6789'))
852
# The server should have issued 3 requests
853
self.assertEqual(3, server.GET_request_nb)
855
def test_complete_readv_leave_pipe_clean(self):
856
server = self.get_readonly_server()
857
t = self._transport(server.get_url())
858
# force transport to issue multiple requests
860
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
861
# The server should have issued 3 requests
862
self.assertEqual(3, server.GET_request_nb)
863
self.assertEqual('0123456789', t.get_bytes('a'))
864
self.assertEqual(4, server.GET_request_nb)
866
def test_incomplete_readv_leave_pipe_clean(self):
867
server = self.get_readonly_server()
868
t = self._transport(server.get_url())
869
# force transport to issue multiple requests
871
# Don't collapse readv results into a list so that we leave unread
872
# bytes on the socket
873
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
874
self.assertEqual((0, '0'), ireadv.next())
875
# The server should have issued one request so far
876
self.assertEqual(1, server.GET_request_nb)
877
self.assertEqual('0123456789', t.get_bytes('a'))
878
# get_bytes issued an additional request, the readv pending ones are
880
self.assertEqual(2, server.GET_request_nb)
883
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
884
"""Always reply to range request as if they were single.
886
Don't be explicit about it, just to annoy the clients.
889
def get_multiple_ranges(self, file, file_size, ranges):
890
"""Answer as if it was a single range request and ignores the rest"""
891
(start, end) = ranges[0]
892
return self.get_single_range(file, file_size, start, end)
895
599
class TestSingleRangeRequestServer(TestRangeRequestServer):
896
600
"""Test readv against a server which accept only single range requests"""
898
_req_handler_class = SingleRangeRequestHandler
901
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
902
"""Only reply to simple range requests, errors out on multiple"""
904
def get_multiple_ranges(self, file, file_size, ranges):
905
"""Refuses the multiple ranges request"""
908
self.send_error(416, "Requested range not satisfiable")
910
(start, end) = ranges[0]
911
return self.get_single_range(file, file_size, start, end)
914
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
915
"""Test readv against a server which only accept single range requests"""
917
_req_handler_class = SingleOnlyRangeRequestHandler
920
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
921
"""Ignore range requests without notice"""
924
# Update the statistics
925
self.server.test_case_server.GET_request_nb += 1
926
# Just bypass the range handling done by TestingHTTPRequestHandler
927
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
602
def create_transport_readonly_server(self):
603
return HttpServer(SingleRangeRequestHandler)
606
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
607
TestCaseWithWebserver):
608
"""Tests single range requests accepting server for urllib implementation"""
610
_transport = HttpTransport_urllib
613
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
614
TestSingleRangeRequestServer,
615
TestCaseWithWebserver):
616
"""Tests single range requests accepting server for pycurl implementation"""
930
619
class TestNoRangeRequestServer(TestRangeRequestServer):
931
620
"""Test readv against a server which do not accept range requests"""
933
_req_handler_class = NoRangeRequestHandler
936
class MultipleRangeWithoutContentLengthRequestHandler(
937
http_server.TestingHTTPRequestHandler):
938
"""Reply to multiple range requests without content length header."""
940
def get_multiple_ranges(self, file, file_size, ranges):
941
self.send_response(206)
942
self.send_header('Accept-Ranges', 'bytes')
943
boundary = "%d" % random.randint(0,0x7FFFFFFF)
944
self.send_header("Content-Type",
945
"multipart/byteranges; boundary=%s" % boundary)
947
for (start, end) in ranges:
948
self.wfile.write("--%s\r\n" % boundary)
949
self.send_header("Content-type", 'application/octet-stream')
950
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
954
self.send_range_content(file, start, end - start + 1)
956
self.wfile.write("--%s\r\n" % boundary)
959
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
961
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
964
class TruncatedMultipleRangeRequestHandler(
965
http_server.TestingHTTPRequestHandler):
966
"""Reply to multiple range requests truncating the last ones.
968
This server generates responses whose Content-Length describes all the
969
ranges, but fail to include the last ones leading to client short reads.
970
This has been observed randomly with lighttpd (bug #179368).
973
_truncated_ranges = 2
975
def get_multiple_ranges(self, file, file_size, ranges):
976
self.send_response(206)
977
self.send_header('Accept-Ranges', 'bytes')
979
self.send_header('Content-Type',
980
'multipart/byteranges; boundary=%s' % boundary)
981
boundary_line = '--%s\r\n' % boundary
982
# Calculate the Content-Length
984
for (start, end) in ranges:
985
content_length += len(boundary_line)
986
content_length += self._header_line_length(
987
'Content-type', 'application/octet-stream')
988
content_length += self._header_line_length(
989
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
990
content_length += len('\r\n') # end headers
991
content_length += end - start # + 1
992
content_length += len(boundary_line)
993
self.send_header('Content-length', content_length)
996
# Send the multipart body
998
for (start, end) in ranges:
999
self.wfile.write(boundary_line)
1000
self.send_header('Content-type', 'application/octet-stream')
1001
self.send_header('Content-Range', 'bytes %d-%d/%d'
1002
% (start, end, file_size))
1004
if cur + self._truncated_ranges >= len(ranges):
1005
# Abruptly ends the response and close the connection
1006
self.close_connection = 1
1008
self.send_range_content(file, start, end - start + 1)
1011
self.wfile.write(boundary_line)
1014
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
1016
_req_handler_class = TruncatedMultipleRangeRequestHandler
1019
super(TestTruncatedMultipleRangeServer, self).setUp()
1020
self.build_tree_contents([('a', '0123456789')],)
1022
def test_readv_with_short_reads(self):
1023
server = self.get_readonly_server()
1024
t = self._transport(server.get_url())
1025
# Force separate ranges for each offset
1026
t._bytes_to_read_before_seek = 0
1027
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1028
self.assertEqual((0, '0'), ireadv.next())
1029
self.assertEqual((2, '2'), ireadv.next())
1030
if not self._testing_pycurl():
1031
# Only one request have been issued so far (except for pycurl that
1032
# try to read the whole response at once)
1033
self.assertEqual(1, server.GET_request_nb)
1034
self.assertEqual((4, '45'), ireadv.next())
1035
self.assertEqual((9, '9'), ireadv.next())
1036
# Both implementations issue 3 requests but:
1037
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1039
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
1040
self.assertEqual(3, server.GET_request_nb)
1041
# Finally the client have tried a single range request and stays in
1043
self.assertEqual('single', t._range_hint)
1045
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1046
"""Errors out when range specifiers exceed the limit"""
1048
def get_multiple_ranges(self, file, file_size, ranges):
1049
"""Refuses the multiple ranges request"""
1050
tcs = self.server.test_case_server
1051
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1053
# Emulate apache behavior
1054
self.send_error(400, "Bad Request")
1056
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1057
self, file, file_size, ranges)
1060
class LimitedRangeHTTPServer(http_server.HttpServer):
1061
"""An HttpServer erroring out on requests with too much range specifiers"""
1063
def __init__(self, request_handler=LimitedRangeRequestHandler,
1064
protocol_version=None,
1066
http_server.HttpServer.__init__(self, request_handler,
1067
protocol_version=protocol_version)
1068
self.range_limit = range_limit
1071
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1072
"""Tests readv requests against a server erroring out on too much ranges."""
1074
# Requests with more range specifiers will error out
1077
622
def create_transport_readonly_server(self):
1078
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1079
protocol_version=self._protocol_version)
1081
def get_transport(self):
1082
return self._transport(self.get_readonly_server().get_url())
1085
http_utils.TestCaseWithWebserver.setUp(self)
1086
# We need to manipulate ranges that correspond to real chunks in the
1087
# response, so we build a content appropriately.
1088
filler = ''.join(['abcdefghij' for x in range(102)])
1089
content = ''.join(['%04d' % v + filler for v in range(16)])
1090
self.build_tree_contents([('a', content)],)
1092
def test_few_ranges(self):
1093
t = self.get_transport()
1094
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1095
self.assertEqual(l[0], (0, '0000'))
1096
self.assertEqual(l[1], (1024, '0001'))
1097
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1099
def test_more_ranges(self):
1100
t = self.get_transport()
1101
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1102
self.assertEqual(l[0], (0, '0000'))
1103
self.assertEqual(l[1], (1024, '0001'))
1104
self.assertEqual(l[2], (4096, '0004'))
1105
self.assertEqual(l[3], (8192, '0008'))
1106
# The server will refuse to serve the first request (too much ranges),
1107
# a second request will succeed.
1108
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1111
class TestHttpProxyWhiteBox(tests.TestCase):
1112
"""Whitebox test proxy http authorization.
1114
Only the urllib implementation is tested here.
1118
tests.TestCase.setUp(self)
1123
tests.TestCase.tearDown(self)
1125
def _install_env(self, env):
1126
for name, value in env.iteritems():
1127
self._old_env[name] = osutils.set_or_unset_env(name, value)
1129
def _restore_env(self):
1130
for name, value in self._old_env.iteritems():
1131
osutils.set_or_unset_env(name, value)
1133
def _proxied_request(self):
1134
handler = _urllib2_wrappers.ProxyHandler()
1135
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
1136
handler.set_proxy(request, 'http')
1139
def test_empty_user(self):
1140
self._install_env({'http_proxy': 'http://bar.com'})
1141
request = self._proxied_request()
1142
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1144
def test_invalid_proxy(self):
1145
"""A proxy env variable without scheme"""
1146
self._install_env({'http_proxy': 'host:1234'})
1147
self.assertRaises(errors.InvalidURL, self._proxied_request)
1150
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
623
return HttpServer(NoRangeRequestHandler)
626
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
627
TestCaseWithWebserver):
628
"""Tests range requests refusing server for urllib implementation"""
630
_transport = HttpTransport_urllib
633
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
634
TestNoRangeRequestServer,
635
TestCaseWithWebserver):
636
"""Tests range requests refusing server for pycurl implementation"""
639
class TestProxyHttpServer(object):
1151
640
"""Tests proxy server.
642
This MUST be used by daughter classes that also inherit from
643
TestCaseWithTwoWebservers.
645
We can't inherit directly from TestCaseWithTwoWebservers or
646
the test framework will try to create an instance which
647
cannot run, its implementation being incomplete.
1153
649
Be aware that we do not setup a real proxy here. Instead, we
1154
650
check that the *connection* goes through the proxy by serving
1155
651
different content (the faked proxy server append '-proxied'
1251
731
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
1252
732
'NO_PROXY': self.no_proxy_host})
1254
def test_http_proxy_without_scheme(self):
1255
if self._testing_pycurl():
1256
# pycurl *ignores* invalid proxy env variables. If that ever change
1257
# in the future, this test will fail indicating that pycurl do not
1258
# ignore anymore such variables.
1259
self.not_proxied_in_env({'http_proxy': self.proxy_address})
1261
self.assertRaises(errors.InvalidURL,
1262
self.proxied_in_env,
1263
{'http_proxy': self.proxy_address})
1266
class TestRanges(http_utils.TestCaseWithWebserver):
1267
"""Test the Range header in GET methods."""
1270
http_utils.TestCaseWithWebserver.setUp(self)
735
class TestProxyHttpServer_urllib(TestProxyHttpServer,
736
TestCaseWithTwoWebservers):
737
"""Tests proxy server for urllib implementation"""
739
_transport = HttpTransport_urllib
742
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
744
TestCaseWithTwoWebservers):
745
"""Tests proxy server for pycurl implementation"""
748
TestProxyHttpServer.setUp(self)
749
# Oh my ! pycurl does not check for the port as part of
750
# no_proxy :-( So we just test the host part
751
self.no_proxy_host = 'localhost'
753
def test_HTTP_PROXY(self):
754
# pycurl do not check HTTP_PROXY for security reasons
755
# (for use in a CGI context that we do not care
756
# about. Should we ?)
759
def test_HTTP_PROXY_with_NO_PROXY(self):
763
class TestRanges(object):
764
"""Test the Range header in GET methods..
766
This MUST be used by daughter classes that also inherit from
767
TestCaseWithWebserver.
769
We can't inherit directly from TestCaseWithWebserver or the
770
test framework will try to create an instance which cannot
771
run, its implementation being incomplete.
775
TestCaseWithWebserver.setUp(self)
1271
776
self.build_tree_contents([('a', '0123456789')],)
1272
777
server = self.get_readonly_server()
1273
778
self.transport = self._transport(server.get_url())
1275
def create_transport_readonly_server(self):
1276
return http_server.HttpServer(protocol_version=self._protocol_version)
1278
def _file_contents(self, relpath, ranges):
1279
offsets = [ (start, end - start + 1) for start, end in ranges]
1280
coalesce = self.transport._coalesce_offsets
1281
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1282
code, data = self.transport._get(relpath, coalesced)
1283
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1284
for start, end in ranges:
1286
yield data.read(end - start + 1)
780
def _file_contents(self, relpath, ranges, tail_amount=0):
781
code, data = self.transport._get(relpath, ranges)
782
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
783
for start, end in ranges:
785
yield data.read(end - start + 1)
1288
787
def _file_tail(self, relpath, tail_amount):
1289
code, data = self.transport._get(relpath, [], tail_amount)
1290
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1291
data.seek(-tail_amount, 2)
1292
return data.read(tail_amount)
788
code, data = self.transport._get(relpath, [], tail_amount)
789
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
790
data.seek(-tail_amount + 1, 2)
791
return data.read(tail_amount)
1294
793
def test_range_header(self):
1296
795
map(self.assertEqual,['0', '234'],
1297
796
list(self._file_contents('a', [(0,0), (2,4)])),)
1299
def test_range_header_tail(self):
1300
798
self.assertEqual('789', self._file_tail('a', 3))
1302
def test_syntactically_invalid_range_header(self):
1303
self.assertListRaises(errors.InvalidHttpRange,
1304
self._file_contents, 'a', [(4, 3)])
1306
def test_semantically_invalid_range_header(self):
1307
self.assertListRaises(errors.InvalidHttpRange,
1308
self._file_contents, 'a', [(42, 128)])
1311
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1312
"""Test redirection between http servers."""
1314
def create_transport_secondary_server(self):
1315
"""Create the secondary server redirecting to the primary server"""
1316
new = self.get_readonly_server()
1318
redirecting = http_utils.HTTPServerRedirecting(
1319
protocol_version=self._protocol_version)
1320
redirecting.redirect_to(new.host, new.port)
1324
super(TestHTTPRedirections, self).setUp()
1325
self.build_tree_contents([('a', '0123456789'),
1327
'# Bazaar revision bundle v0.9\n#\n')
1329
# The requests to the old server will be redirected to the new server
1330
self.old_transport = self._transport(self.old_server.get_url())
1332
def test_redirected(self):
1333
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1334
t = self._transport(self.new_server.get_url())
1335
self.assertEqual('0123456789', t.get('a').read())
1337
def test_read_redirected_bundle_from_url(self):
1338
from bzrlib.bundle import read_bundle_from_url
1339
url = self.old_transport.abspath('bundle')
1340
bundle = self.applyDeprecated(deprecated_in((1, 12, 0)),
1341
read_bundle_from_url, url)
1342
# If read_bundle_from_url was successful we get an empty bundle
1343
self.assertEqual([], bundle.revisions)
1346
class RedirectedRequest(_urllib2_wrappers.Request):
1347
"""Request following redirections. """
1349
init_orig = _urllib2_wrappers.Request.__init__
1351
def __init__(self, method, url, *args, **kwargs):
1355
# Since the tests using this class will replace
1356
# _urllib2_wrappers.Request, we can't just call the base class __init__
1358
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1359
self.follow_redirections = True
1362
def install_redirected_request(test):
1363
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1366
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1367
"""Test redirections.
1369
http implementations do not redirect silently anymore (they
1370
do not redirect at all in fact). The mechanism is still in
1371
place at the _urllib2_wrappers.Request level and these tests
1374
For the pycurl implementation
1375
the redirection have been deleted as we may deprecate pycurl
1376
and I have no place to keep a working implementation.
1381
if (features.pycurl.available()
1382
and self._transport == PyCurlTransport):
1383
raise tests.TestNotApplicable(
1384
"pycurl doesn't redirect silently annymore")
1385
super(TestHTTPSilentRedirections, self).setUp()
1386
install_redirected_request(self)
1387
self.build_tree_contents([('a','a'),
1389
('1/a', 'redirected once'),
1391
('2/a', 'redirected twice'),
1393
('3/a', 'redirected thrice'),
1395
('4/a', 'redirected 4 times'),
1397
('5/a', 'redirected 5 times'),
1400
self.old_transport = self._transport(self.old_server.get_url())
1402
def create_transport_secondary_server(self):
1403
"""Create the secondary server, redirections are defined in the tests"""
1404
return http_utils.HTTPServerRedirecting(
1405
protocol_version=self._protocol_version)
1407
def test_one_redirection(self):
1408
t = self.old_transport
1410
req = RedirectedRequest('GET', t.abspath('a'))
1411
new_prefix = 'http://%s:%s' % (self.new_server.host,
1412
self.new_server.port)
1413
self.old_server.redirections = \
1414
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1415
self.assertEqual('redirected once',t._perform(req).read())
1417
def test_five_redirections(self):
1418
t = self.old_transport
1420
req = RedirectedRequest('GET', t.abspath('a'))
1421
old_prefix = 'http://%s:%s' % (self.old_server.host,
1422
self.old_server.port)
1423
new_prefix = 'http://%s:%s' % (self.new_server.host,
1424
self.new_server.port)
1425
self.old_server.redirections = [
1426
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1427
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1428
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1429
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1430
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1432
self.assertEqual('redirected 5 times',t._perform(req).read())
1435
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1436
"""Test transport.do_catching_redirections."""
1439
super(TestDoCatchRedirections, self).setUp()
1440
self.build_tree_contents([('a', '0123456789'),],)
1442
self.old_transport = self._transport(self.old_server.get_url())
1444
def get_a(self, transport):
1445
return transport.get('a')
1447
def test_no_redirection(self):
1448
t = self._transport(self.new_server.get_url())
1450
# We use None for redirected so that we fail if redirected
1451
self.assertEqual('0123456789',
1452
transport.do_catching_redirections(
1453
self.get_a, t, None).read())
1455
def test_one_redirection(self):
1456
self.redirections = 0
1458
def redirected(transport, exception, redirection_notice):
1459
self.redirections += 1
1460
dir, file = urlutils.split(exception.target)
1461
return self._transport(dir)
1463
self.assertEqual('0123456789',
1464
transport.do_catching_redirections(
1465
self.get_a, self.old_transport, redirected).read())
1466
self.assertEqual(1, self.redirections)
1468
def test_redirection_loop(self):
1470
def redirected(transport, exception, redirection_notice):
1471
# By using the redirected url as a base dir for the
1472
# *old* transport, we create a loop: a => a/a =>
1474
return self.old_transport.clone(exception.target)
1476
self.assertRaises(errors.TooManyRedirections,
1477
transport.do_catching_redirections,
1478
self.get_a, self.old_transport, redirected)
1481
class TestAuth(http_utils.TestCaseWithWebserver):
1482
"""Test authentication scheme"""
1484
_auth_header = 'Authorization'
1485
_password_prompt_prefix = ''
1486
_username_prompt_prefix = ''
1491
super(TestAuth, self).setUp()
1492
self.server = self.get_readonly_server()
1493
self.build_tree_contents([('a', 'contents of a\n'),
1494
('b', 'contents of b\n'),])
1496
def create_transport_readonly_server(self):
1497
return self._auth_server(protocol_version=self._protocol_version)
1499
def _testing_pycurl(self):
1500
# TODO: This is duplicated for lots of the classes in this file
1501
return (features.pycurl.available()
1502
and self._transport == PyCurlTransport)
1504
def get_user_url(self, user, password):
1505
"""Build an url embedding user and password"""
1506
url = '%s://' % self.server._url_protocol
1507
if user is not None:
1509
if password is not None:
1510
url += ':' + password
1512
url += '%s:%s/' % (self.server.host, self.server.port)
1515
def get_user_transport(self, user, password):
1516
return self._transport(self.get_user_url(user, password))
1518
def test_no_user(self):
1519
self.server.add_user('joe', 'foo')
1520
t = self.get_user_transport(None, None)
1521
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1522
# Only one 'Authentication Required' error should occur
1523
self.assertEqual(1, self.server.auth_required_errors)
1525
def test_empty_pass(self):
1526
self.server.add_user('joe', '')
1527
t = self.get_user_transport('joe', '')
1528
self.assertEqual('contents of a\n', t.get('a').read())
1529
# Only one 'Authentication Required' error should occur
1530
self.assertEqual(1, self.server.auth_required_errors)
1532
def test_user_pass(self):
1533
self.server.add_user('joe', 'foo')
1534
t = self.get_user_transport('joe', 'foo')
1535
self.assertEqual('contents of a\n', t.get('a').read())
1536
# Only one 'Authentication Required' error should occur
1537
self.assertEqual(1, self.server.auth_required_errors)
1539
def test_unknown_user(self):
1540
self.server.add_user('joe', 'foo')
1541
t = self.get_user_transport('bill', 'foo')
1542
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1543
# Two 'Authentication Required' errors should occur (the
1544
# initial 'who are you' and 'I don't know you, who are
1546
self.assertEqual(2, self.server.auth_required_errors)
1548
def test_wrong_pass(self):
1549
self.server.add_user('joe', 'foo')
1550
t = self.get_user_transport('joe', 'bar')
1551
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1552
# Two 'Authentication Required' errors should occur (the
1553
# initial 'who are you' and 'this is not you, who are you')
1554
self.assertEqual(2, self.server.auth_required_errors)
1556
def test_prompt_for_username(self):
1557
if self._testing_pycurl():
1558
raise tests.TestNotApplicable(
1559
'pycurl cannot prompt, it handles auth by embedding'
1560
' user:pass in urls only')
1562
self.server.add_user('joe', 'foo')
1563
t = self.get_user_transport(None, None)
1564
stdout = tests.StringIOWrapper()
1565
stderr = tests.StringIOWrapper()
1566
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1567
stdout=stdout, stderr=stderr)
1568
self.assertEqual('contents of a\n',t.get('a').read())
1569
# stdin should be empty
1570
self.assertEqual('', ui.ui_factory.stdin.readline())
1572
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1573
self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1574
self.assertEqual('', stdout.getvalue())
1575
self._check_password_prompt(t._unqualified_scheme, 'joe',
1578
def test_prompt_for_password(self):
1579
if self._testing_pycurl():
1580
raise tests.TestNotApplicable(
1581
'pycurl cannot prompt, it handles auth by embedding'
1582
' user:pass in urls only')
1584
self.server.add_user('joe', 'foo')
1585
t = self.get_user_transport('joe', None)
1586
stdout = tests.StringIOWrapper()
1587
stderr = tests.StringIOWrapper()
1588
ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1589
stdout=stdout, stderr=stderr)
1590
self.assertEqual('contents of a\n', t.get('a').read())
1591
# stdin should be empty
1592
self.assertEqual('', ui.ui_factory.stdin.readline())
1593
self._check_password_prompt(t._unqualified_scheme, 'joe',
1595
self.assertEqual('', stdout.getvalue())
1596
# And we shouldn't prompt again for a different request
1597
# against the same transport.
1598
self.assertEqual('contents of b\n',t.get('b').read())
1600
# And neither against a clone
1601
self.assertEqual('contents of b\n',t2.get('b').read())
1602
# Only one 'Authentication Required' error should occur
1603
self.assertEqual(1, self.server.auth_required_errors)
1605
def _check_password_prompt(self, scheme, user, actual_prompt):
1606
expected_prompt = (self._password_prompt_prefix
1607
+ ("%s %s@%s:%d, Realm: '%s' password: "
1609
user, self.server.host, self.server.port,
1610
self.server.auth_realm)))
1611
self.assertEqual(expected_prompt, actual_prompt)
1613
def _expected_username_prompt(self, scheme):
1614
return (self._username_prompt_prefix
1615
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1616
self.server.host, self.server.port,
1617
self.server.auth_realm))
1619
def test_no_prompt_for_password_when_using_auth_config(self):
1620
if self._testing_pycurl():
1621
raise tests.TestNotApplicable(
1622
'pycurl does not support authentication.conf'
1623
' since it cannot prompt')
1627
stdin_content = 'bar\n' # Not the right password
1628
self.server.add_user(user, password)
1629
t = self.get_user_transport(user, None)
1630
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1631
stderr=tests.StringIOWrapper())
1632
# Create a minimal config file with the right password
1633
conf = config.AuthenticationConfig()
1634
conf._get_config().update(
1635
{'httptest': {'scheme': 'http', 'port': self.server.port,
1636
'user': user, 'password': password}})
1638
# Issue a request to the server to connect
1639
self.assertEqual('contents of a\n',t.get('a').read())
1640
# stdin should have been left untouched
1641
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1642
# Only one 'Authentication Required' error should occur
1643
self.assertEqual(1, self.server.auth_required_errors)
1645
def test_user_from_auth_conf(self):
1646
if self._testing_pycurl():
1647
raise tests.TestNotApplicable(
1648
'pycurl does not support authentication.conf')
1651
self.server.add_user(user, password)
1652
# Create a minimal config file with the right password
1653
conf = config.AuthenticationConfig()
1654
conf._get_config().update(
1655
{'httptest': {'scheme': 'http', 'port': self.server.port,
1656
'user': user, 'password': password}})
1658
t = self.get_user_transport(None, None)
1659
# Issue a request to the server to connect
1660
self.assertEqual('contents of a\n', t.get('a').read())
1661
# Only one 'Authentication Required' error should occur
1662
self.assertEqual(1, self.server.auth_required_errors)
1664
def test_changing_nonce(self):
1665
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1666
http_utils.ProxyDigestAuthServer):
1667
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1668
if self._testing_pycurl():
1669
raise tests.KnownFailure(
1670
'pycurl does not handle a nonce change')
1671
self.server.add_user('joe', 'foo')
1672
t = self.get_user_transport('joe', 'foo')
1673
self.assertEqual('contents of a\n', t.get('a').read())
1674
self.assertEqual('contents of b\n', t.get('b').read())
1675
# Only one 'Authentication Required' error should have
1677
self.assertEqual(1, self.server.auth_required_errors)
1678
# The server invalidates the current nonce
1679
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1680
self.assertEqual('contents of a\n', t.get('a').read())
1681
# Two 'Authentication Required' errors should occur (the
1682
# initial 'who are you' and a second 'who are you' with the new nonce)
1683
self.assertEqual(2, self.server.auth_required_errors)
1687
class TestProxyAuth(TestAuth):
1688
"""Test proxy authentication schemes."""
1690
_auth_header = 'Proxy-authorization'
1691
_password_prompt_prefix = 'Proxy '
1692
_username_prompt_prefix = 'Proxy '
1695
super(TestProxyAuth, self).setUp()
1697
self.addCleanup(self._restore_env)
1698
# Override the contents to avoid false positives
1699
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1700
('b', 'not proxied contents of b\n'),
1701
('a-proxied', 'contents of a\n'),
1702
('b-proxied', 'contents of b\n'),
1705
def get_user_transport(self, user, password):
1706
self._install_env({'all_proxy': self.get_user_url(user, password)})
1707
return self._transport(self.server.get_url())
1709
def _install_env(self, env):
1710
for name, value in env.iteritems():
1711
self._old_env[name] = osutils.set_or_unset_env(name, value)
1713
def _restore_env(self):
1714
for name, value in self._old_env.iteritems():
1715
osutils.set_or_unset_env(name, value)
1717
def test_empty_pass(self):
1718
if self._testing_pycurl():
1720
if pycurl.version_info()[1] < '7.16.0':
1721
raise tests.KnownFailure(
1722
'pycurl < 7.16.0 does not handle empty proxy passwords')
1723
super(TestProxyAuth, self).test_empty_pass()
1726
class SampleSocket(object):
1727
"""A socket-like object for use in testing the HTTP request handler."""
1729
def __init__(self, socket_read_content):
1730
"""Constructs a sample socket.
1732
:param socket_read_content: a byte sequence
1734
# Use plain python StringIO so we can monkey-patch the close method to
1735
# not discard the contents.
1736
from StringIO import StringIO
1737
self.readfile = StringIO(socket_read_content)
1738
self.writefile = StringIO()
1739
self.writefile.close = lambda: None
1741
def makefile(self, mode='r', bufsize=None):
1743
return self.readfile
1745
return self.writefile
1748
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1751
super(SmartHTTPTunnellingTest, self).setUp()
1752
# We use the VFS layer as part of HTTP tunnelling tests.
1753
self._captureVar('BZR_NO_SMART_VFS', None)
1754
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1756
def create_transport_readonly_server(self):
1757
return http_utils.HTTPServerWithSmarts(
1758
protocol_version=self._protocol_version)
1760
def test_open_bzrdir(self):
1761
branch = self.make_branch('relpath')
1762
http_server = self.get_readonly_server()
1763
url = http_server.get_url() + 'relpath'
1764
bd = bzrdir.BzrDir.open(url)
1765
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1767
def test_bulk_data(self):
1768
# We should be able to send and receive bulk data in a single message.
1769
# The 'readv' command in the smart protocol both sends and receives
1770
# bulk data, so we use that.
1771
self.build_tree(['data-file'])
1772
http_server = self.get_readonly_server()
1773
http_transport = self._transport(http_server.get_url())
1774
medium = http_transport.get_smart_medium()
1775
# Since we provide the medium, the url below will be mostly ignored
1776
# during the test, as long as the path is '/'.
1777
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1780
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1782
def test_http_send_smart_request(self):
1784
post_body = 'hello\n'
1785
expected_reply_body = 'ok\x012\n'
1787
http_server = self.get_readonly_server()
1788
http_transport = self._transport(http_server.get_url())
1789
medium = http_transport.get_smart_medium()
1790
response = medium.send_http_smart_request(post_body)
1791
reply_body = response.read()
1792
self.assertEqual(expected_reply_body, reply_body)
1794
def test_smart_http_server_post_request_handler(self):
1795
httpd = self.get_readonly_server()._get_httpd()
1797
socket = SampleSocket(
1798
'POST /.bzr/smart %s \r\n' % self._protocol_version
1799
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1801
+ 'Content-Length: 6\r\n'
1804
# Beware: the ('localhost', 80) below is the
1805
# client_address parameter, but we don't have one because
1806
# we have defined a socket which is not bound to an
1807
# address. The test framework never uses this client
1808
# address, so far...
1809
request_handler = http_utils.SmartRequestHandler(socket,
1812
response = socket.writefile.getvalue()
1813
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1814
# This includes the end of the HTTP headers, and all the body.
1815
expected_end_of_response = '\r\n\r\nok\x012\n'
1816
self.assertEndsWith(response, expected_end_of_response)
1819
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1820
"""No smart server here request handler."""
1823
self.send_error(403, "Forbidden")
1826
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1827
"""Test smart client behaviour against an http server without smarts."""
1829
_req_handler_class = ForbiddenRequestHandler
1831
def test_probe_smart_server(self):
1832
"""Test error handling against server refusing smart requests."""
1833
server = self.get_readonly_server()
1834
t = self._transport(server.get_url())
1835
# No need to build a valid smart request here, the server will not even
1836
# try to interpret it.
1837
self.assertRaises(errors.SmartProtocolError,
1838
t.get_smart_medium().send_http_smart_request,
1841
class Test_redirected_to(tests.TestCase):
1843
def test_redirected_to_subdir(self):
1844
t = self._transport('http://www.example.com/foo')
1845
r = t._redirected_to('http://www.example.com/foo',
1846
'http://www.example.com/foo/subdir')
1847
self.assertIsInstance(r, type(t))
1848
# Both transports share the some connection
1849
self.assertEqual(t._get_connection(), r._get_connection())
1851
def test_redirected_to_self_with_slash(self):
1852
t = self._transport('http://www.example.com/foo')
1853
r = t._redirected_to('http://www.example.com/foo',
1854
'http://www.example.com/foo/')
1855
self.assertIsInstance(r, type(t))
1856
# Both transports share the some connection (one can argue that we
1857
# should return the exact same transport here, but that seems
1859
self.assertEqual(t._get_connection(), r._get_connection())
1861
def test_redirected_to_host(self):
1862
t = self._transport('http://www.example.com/foo')
1863
r = t._redirected_to('http://www.example.com/foo',
1864
'http://foo.example.com/foo/subdir')
1865
self.assertIsInstance(r, type(t))
1867
def test_redirected_to_same_host_sibling_protocol(self):
1868
t = self._transport('http://www.example.com/foo')
1869
r = t._redirected_to('http://www.example.com/foo',
1870
'https://www.example.com/foo')
1871
self.assertIsInstance(r, type(t))
1873
def test_redirected_to_same_host_different_protocol(self):
1874
t = self._transport('http://www.example.com/foo')
1875
r = t._redirected_to('http://www.example.com/foo',
1876
'ftp://www.example.com/foo')
1877
self.assertNotEquals(type(r), type(t))
1879
def test_redirected_to_different_host_same_user(self):
1880
t = self._transport('http://joe@www.example.com/foo')
1881
r = t._redirected_to('http://www.example.com/foo',
1882
'https://foo.example.com/foo')
1883
self.assertIsInstance(r, type(t))
1884
self.assertEqual(t._user, r._user)
1887
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1888
"""Request handler for a unique and pre-defined request.
1890
The only thing we care about here is how many bytes travel on the wire. But
1891
since we want to measure it for a real http client, we have to send it
1894
We expect to receive a *single* request nothing more (and we won't even
1895
check what request it is, we just measure the bytes read until an empty
1899
def handle_one_request(self):
1900
tcs = self.server.test_case_server
1901
requestline = self.rfile.readline()
1902
headers = self.MessageClass(self.rfile, 0)
1903
# We just read: the request, the headers, an empty line indicating the
1904
# end of the headers.
1905
bytes_read = len(requestline)
1906
for line in headers.headers:
1907
bytes_read += len(line)
1908
bytes_read += len('\r\n')
1909
if requestline.startswith('POST'):
1910
# The body should be a single line (or we don't know where it ends
1911
# and we don't want to issue a blocking read)
1912
body = self.rfile.readline()
1913
bytes_read += len(body)
1914
tcs.bytes_read = bytes_read
1916
# We set the bytes written *before* issuing the write, the client is
1917
# supposed to consume every produced byte *before* checking that value.
1919
# Doing the oppposite may lead to test failure: we may be interrupted
1920
# after the write but before updating the value. The client can then
1921
# continue and read the value *before* we can update it. And yes,
1922
# this has been observed -- vila 20090129
1923
tcs.bytes_written = len(tcs.canned_response)
1924
self.wfile.write(tcs.canned_response)
1927
class ActivityServerMixin(object):
1929
def __init__(self, protocol_version):
1930
super(ActivityServerMixin, self).__init__(
1931
request_handler=PredefinedRequestHandler,
1932
protocol_version=protocol_version)
1933
# Bytes read and written by the server
1935
self.bytes_written = 0
1936
self.canned_response = None
1939
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1943
if tests.HTTPSServerFeature.available():
1944
from bzrlib.tests import https_server
1945
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1949
class TestActivityMixin(object):
1950
"""Test socket activity reporting.
1952
We use a special purpose server to control the bytes sent and received and
1953
be able to predict the activity on the client socket.
1957
tests.TestCase.setUp(self)
1958
self.server = self._activity_server(self._protocol_version)
1959
self.server.start_server()
1960
self.activities = {}
1961
def report_activity(t, bytes, direction):
1962
count = self.activities.get(direction, 0)
1964
self.activities[direction] = count
1966
# We override at class level because constructors may propagate the
1967
# bound method and render instance overriding ineffective (an
1968
# alternative would be to define a specific ui factory instead...)
1969
self.orig_report_activity = self._transport._report_activity
1970
self._transport._report_activity = report_activity
1973
self._transport._report_activity = self.orig_report_activity
1974
self.server.stop_server()
1975
tests.TestCase.tearDown(self)
1977
def get_transport(self):
1978
return self._transport(self.server.get_url())
1980
def assertActivitiesMatch(self):
1981
self.assertEqual(self.server.bytes_read,
1982
self.activities.get('write', 0), 'written bytes')
1983
self.assertEqual(self.server.bytes_written,
1984
self.activities.get('read', 0), 'read bytes')
1987
self.server.canned_response = '''HTTP/1.1 200 OK\r
1988
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
1989
Server: Apache/2.0.54 (Fedora)\r
1990
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
1991
ETag: "56691-23-38e9ae00"\r
1992
Accept-Ranges: bytes\r
1993
Content-Length: 35\r
1995
Content-Type: text/plain; charset=UTF-8\r
1997
Bazaar-NG meta directory, format 1
1999
t = self.get_transport()
2000
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2001
t.get('foo/bar').read())
2002
self.assertActivitiesMatch()
2005
self.server.canned_response = '''HTTP/1.1 200 OK\r
2006
Server: SimpleHTTP/0.6 Python/2.5.2\r
2007
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2008
Content-type: application/octet-stream\r
2009
Content-Length: 20\r
2010
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2013
t = self.get_transport()
2014
self.assertTrue(t.has('foo/bar'))
2015
self.assertActivitiesMatch()
2017
def test_readv(self):
2018
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2019
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2020
Server: Apache/2.0.54 (Fedora)\r
2021
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2022
ETag: "238a3c-16ec2-805c5540"\r
2023
Accept-Ranges: bytes\r
2024
Content-Length: 1534\r
2026
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2029
--418470f848b63279b\r
2030
Content-type: text/plain; charset=UTF-8\r
2031
Content-range: bytes 0-254/93890\r
2033
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2034
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2035
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2036
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2037
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2039
--418470f848b63279b\r
2040
Content-type: text/plain; charset=UTF-8\r
2041
Content-range: bytes 1000-2049/93890\r
2044
mbp@sourcefrog.net-20050311063625-07858525021f270b
2045
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2046
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2047
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2048
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2049
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2050
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2051
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2052
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2053
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2054
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2055
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2056
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2057
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2058
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2059
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2060
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2061
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2062
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2063
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2065
--418470f848b63279b--\r
2067
t = self.get_transport()
2068
# Remember that the request is ignored and that the ranges below
2069
# doesn't have to match the canned response.
2070
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2071
self.assertEqual(2, len(l))
2072
self.assertActivitiesMatch()
2074
def test_post(self):
2075
self.server.canned_response = '''HTTP/1.1 200 OK\r
2076
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2077
Server: Apache/2.0.54 (Fedora)\r
2078
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2079
ETag: "56691-23-38e9ae00"\r
2080
Accept-Ranges: bytes\r
2081
Content-Length: 35\r
2083
Content-Type: text/plain; charset=UTF-8\r
2085
lalala whatever as long as itsssss
2087
t = self.get_transport()
2088
# We must send a single line of body bytes, see
2089
# PredefinedRequestHandler.handle_one_request
2090
code, f = t._post('abc def end-of-body\n')
2091
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2092
self.assertActivitiesMatch()
2095
class TestActivity(tests.TestCase, TestActivityMixin):
2098
tests.TestCase.setUp(self)
2099
self.server = self._activity_server(self._protocol_version)
2100
self.server.start_server()
2101
self.activities = {}
2102
def report_activity(t, bytes, direction):
2103
count = self.activities.get(direction, 0)
2105
self.activities[direction] = count
2107
# We override at class level because constructors may propagate the
2108
# bound method and render instance overriding ineffective (an
2109
# alternative would be to define a specific ui factory instead...)
2110
self.orig_report_activity = self._transport._report_activity
2111
self._transport._report_activity = report_activity
2114
self._transport._report_activity = self.orig_report_activity
2115
self.server.stop_server()
2116
tests.TestCase.tearDown(self)
2119
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2122
tests.TestCase.setUp(self)
2123
# Unlike TestActivity, we are really testing ReportingFileSocket and
2124
# ReportingSocket, so we don't need all the parametrization. Since
2125
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2126
# test them through their use by the transport than directly (that's a
2127
# bit less clean but far more simpler and effective).
2128
self.server = ActivityHTTPServer('HTTP/1.1')
2129
self._transport=_urllib.HttpTransport_urllib
2131
self.server.start_server()
2133
# We override at class level because constructors may propagate the
2134
# bound method and render instance overriding ineffective (an
2135
# alternative would be to define a specific ui factory instead...)
2136
self.orig_report_activity = self._transport._report_activity
2137
self._transport._report_activity = None
2140
self._transport._report_activity = self.orig_report_activity
2141
self.server.stop_server()
2142
tests.TestCase.tearDown(self)
2144
def assertActivitiesMatch(self):
2145
# Nothing to check here
2149
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2150
"""Test authentication on the redirected http server."""
2152
_auth_header = 'Authorization'
2153
_password_prompt_prefix = ''
2154
_username_prompt_prefix = ''
2155
_auth_server = http_utils.HTTPBasicAuthServer
2156
_transport = _urllib.HttpTransport_urllib
2158
def create_transport_readonly_server(self):
2159
return self._auth_server()
2161
def create_transport_secondary_server(self):
2162
"""Create the secondary server redirecting to the primary server"""
2163
new = self.get_readonly_server()
2165
redirecting = http_utils.HTTPServerRedirecting()
2166
redirecting.redirect_to(new.host, new.port)
2170
super(TestAuthOnRedirected, self).setUp()
2171
self.build_tree_contents([('a','a'),
2173
('1/a', 'redirected once'),
2175
new_prefix = 'http://%s:%s' % (self.new_server.host,
2176
self.new_server.port)
2177
self.old_server.redirections = [
2178
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2179
self.old_transport = self._transport(self.old_server.get_url())
2180
self.new_server.add_user('joe', 'foo')
2182
def get_a(self, transport):
2183
return transport.get('a')
2185
def test_auth_on_redirected_via_do_catching_redirections(self):
2186
self.redirections = 0
2188
def redirected(transport, exception, redirection_notice):
2189
self.redirections += 1
2190
dir, file = urlutils.split(exception.target)
2191
return self._transport(dir)
2193
stdout = tests.StringIOWrapper()
2194
stderr = tests.StringIOWrapper()
2195
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2196
stdout=stdout, stderr=stderr)
2197
self.assertEqual('redirected once',
2198
transport.do_catching_redirections(
2199
self.get_a, self.old_transport, redirected).read())
2200
self.assertEqual(1, self.redirections)
2201
# stdin should be empty
2202
self.assertEqual('', ui.ui_factory.stdin.readline())
2203
# stdout should be empty, stderr will contains the prompts
2204
self.assertEqual('', stdout.getvalue())
2206
def test_auth_on_redirected_via_following_redirections(self):
2207
self.new_server.add_user('joe', 'foo')
2208
stdout = tests.StringIOWrapper()
2209
stderr = tests.StringIOWrapper()
2210
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2211
stdout=stdout, stderr=stderr)
2212
t = self.old_transport
2213
req = RedirectedRequest('GET', t.abspath('a'))
2214
new_prefix = 'http://%s:%s' % (self.new_server.host,
2215
self.new_server.port)
2216
self.old_server.redirections = [
2217
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2218
self.assertEqual('redirected once',t._perform(req).read())
2219
# stdin should be empty
2220
self.assertEqual('', ui.ui_factory.stdin.readline())
2221
# stdout should be empty, stderr will contains the prompts
2222
self.assertEqual('', stdout.getvalue())
799
# Syntactically invalid range
800
self.assertRaises(errors.InvalidRange,
801
self.transport._get, 'a', [(4, 3)])
802
# Semantically invalid range
803
self.assertRaises(errors.InvalidRange,
804
self.transport._get, 'a', [(42, 128)])
807
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
808
"""Test the Range header in GET methods for urllib implementation"""
810
_transport = HttpTransport_urllib
813
class TestRanges_pycurl(TestWithTransport_pycurl,
815
TestCaseWithWebserver):
816
"""Test the Range header in GET methods for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
