228
class TestHTTPServer(tests.TestCase):
229
"""Test the HTTP servers implementations."""
231
def test_invalid_protocol(self):
232
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
234
protocol_version = 'HTTP/0.1'
236
server = http_server.HttpServer(BogusRequestHandler)
238
self.assertRaises(httplib.UnknownProtocol,server.setUp)
241
self.fail('HTTP Server creation did not raise UnknownProtocol')
243
def test_force_invalid_protocol(self):
244
server = http_server.HttpServer(protocol_version='HTTP/0.1')
246
self.assertRaises(httplib.UnknownProtocol,server.setUp)
249
self.fail('HTTP Server creation did not raise UnknownProtocol')
251
def test_server_start_and_stop(self):
252
server = http_server.HttpServer()
254
self.assertTrue(server._http_running)
256
self.assertFalse(server._http_running)
258
def test_create_http_server_one_zero(self):
259
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
261
protocol_version = 'HTTP/1.0'
263
server = http_server.HttpServer(RequestHandlerOneZero)
265
self.addCleanup(server.tearDown)
266
self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
268
def test_create_http_server_one_one(self):
269
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
271
protocol_version = 'HTTP/1.1'
273
server = http_server.HttpServer(RequestHandlerOneOne)
275
self.addCleanup(server.tearDown)
276
self.assertIsInstance(server._httpd,
277
http_server.TestingThreadingHTTPServer)
279
def test_create_http_server_force_one_one(self):
280
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
282
protocol_version = 'HTTP/1.0'
284
server = http_server.HttpServer(RequestHandlerOneZero,
285
protocol_version='HTTP/1.1')
287
self.addCleanup(server.tearDown)
288
self.assertIsInstance(server._httpd,
289
http_server.TestingThreadingHTTPServer)
291
def test_create_http_server_force_one_zero(self):
292
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
294
protocol_version = 'HTTP/1.1'
296
server = http_server.HttpServer(RequestHandlerOneOne,
297
protocol_version='HTTP/1.0')
299
self.addCleanup(server.tearDown)
300
self.assertIsInstance(server._httpd,
301
http_server.TestingHTTPServer)
304
class TestWithTransport_pycurl(object):
305
"""Test case to inherit from if pycurl is present"""
307
def _get_pycurl_maybe(self):
309
from bzrlib.transport.http._pycurl import PyCurlTransport
310
return PyCurlTransport
311
except errors.DependencyNotPresent:
312
raise tests.TestSkipped('pycurl not present')
314
_transport = property(_get_pycurl_maybe)
317
class TestHttpUrls(tests.TestCase):
319
# 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
322
134
def test_url_parsing(self):
323
135
f = FakeManager()
324
url = http.extract_auth('http://example.com', f)
136
url = extract_auth('http://example.com', f)
325
137
self.assertEquals('http://example.com', url)
326
138
self.assertEquals(0, len(f.credentials))
327
url = http.extract_auth(
328
'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
139
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
329
140
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
330
141
self.assertEquals(1, len(f.credentials))
331
142
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
332
143
f.credentials[0])
335
class TestHttpTransportUrls(tests.TestCase):
336
"""Test the http urls."""
338
145
def test_abs_url(self):
339
146
"""Construction of absolute http URLs"""
340
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
147
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
341
148
eq = self.assertEqualDiff
342
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
343
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
344
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')
345
155
eq(t.abspath('.bzr/1//2/./3'),
346
156
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
348
158
def test_invalid_http_urls(self):
349
159
"""Trap invalid construction of urls"""
350
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
351
self.assertRaises(errors.InvalidURL,
353
'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')
355
167
def test_http_root_urls(self):
356
168
"""Construction of URLs from server root"""
357
t = self._transport('http://bzr.ozlabs.org/')
169
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
358
170
eq = self.assertEqualDiff
359
171
eq(t.abspath('.bzr/tree-version'),
360
172
'http://bzr.ozlabs.org/.bzr/tree-version')
362
174
def test_http_impl_urls(self):
363
175
"""There are servers which ask for particular clients to connect"""
364
server = self._server()
176
server = HttpServer_PyCurl()
367
179
url = server.get_url()
368
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
180
self.assertTrue(url.startswith('http+pycurl://'))
370
182
server.tearDown()
373
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
375
# TODO: This should really be moved into another pycurl
376
# specific test. When https tests will be implemented, take
377
# this one into account.
378
def test_pycurl_without_https_support(self):
379
"""Test that pycurl without SSL do not fail with a traceback.
381
For the purpose of the test, we force pycurl to ignore
382
https by supplying a fake version_info that do not
388
raise tests.TestSkipped('pycurl not present')
390
version_info_orig = pycurl.version_info
392
# Now that we have pycurl imported, we can fake its version_info
393
# This was taken from a windows pycurl without SSL
395
pycurl.version_info = lambda : (2,
403
('ftp', 'gopher', 'telnet',
404
'dict', 'ldap', 'http', 'file'),
408
self.assertRaises(errors.DependencyNotPresent, self._transport,
409
'https://launchpad.net')
411
# Restore the right function
412
pycurl.version_info = version_info_orig
415
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
416
"""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.
419
http_utils.TestCaseWithWebserver.setUp(self)
420
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
197
TestCaseWithWebserver.setUp(self)
198
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
421
199
transport=self.get_transport())
423
201
def test_http_has(self):
767
595
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
768
596
t.readv, 'a', [(12,2)])
770
def test_readv_multiple_get_requests(self):
771
server = self.get_readonly_server()
772
t = self._transport(server.get_url())
773
# force transport to issue multiple requests
774
t._max_readv_combine = 1
775
t._max_get_ranges = 1
776
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
777
self.assertEqual(l[0], (0, '0'))
778
self.assertEqual(l[1], (1, '1'))
779
self.assertEqual(l[2], (3, '34'))
780
self.assertEqual(l[3], (9, '9'))
781
# The server should have issued 4 requests
782
self.assertEqual(4, server.GET_request_nb)
784
def test_readv_get_max_size(self):
785
server = self.get_readonly_server()
786
t = self._transport(server.get_url())
787
# force transport to issue multiple requests by limiting the number of
788
# bytes by request. Note that this apply to coalesced offsets only, a
789
# single range will keep its size even if bigger than the limit.
791
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
792
self.assertEqual(l[0], (0, '0'))
793
self.assertEqual(l[1], (1, '1'))
794
self.assertEqual(l[2], (2, '2345'))
795
self.assertEqual(l[3], (6, '6789'))
796
# The server should have issued 3 requests
797
self.assertEqual(3, server.GET_request_nb)
799
def test_complete_readv_leave_pipe_clean(self):
800
server = self.get_readonly_server()
801
t = self._transport(server.get_url())
802
# force transport to issue multiple requests
804
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
805
# The server should have issued 3 requests
806
self.assertEqual(3, server.GET_request_nb)
807
self.assertEqual('0123456789', t.get_bytes('a'))
808
self.assertEqual(4, server.GET_request_nb)
810
def test_incomplete_readv_leave_pipe_clean(self):
811
server = self.get_readonly_server()
812
t = self._transport(server.get_url())
813
# force transport to issue multiple requests
815
# Don't collapse readv results into a list so that we leave unread
816
# bytes on the socket
817
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
818
self.assertEqual((0, '0'), ireadv.next())
819
# The server should have issued one request so far
820
self.assertEqual(1, server.GET_request_nb)
821
self.assertEqual('0123456789', t.get_bytes('a'))
822
# get_bytes issued an additional request, the readv pending ones are
824
self.assertEqual(2, server.GET_request_nb)
827
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
828
"""Always reply to range request as if they were single.
830
Don't be explicit about it, just to annoy the clients.
833
def get_multiple_ranges(self, file, file_size, ranges):
834
"""Answer as if it was a single range request and ignores the rest"""
835
(start, end) = ranges[0]
836
return self.get_single_range(file, file_size, start, end)
839
599
class TestSingleRangeRequestServer(TestRangeRequestServer):
840
600
"""Test readv against a server which accept only single range requests"""
842
_req_handler_class = SingleRangeRequestHandler
845
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
846
"""Only reply to simple range requests, errors out on multiple"""
848
def get_multiple_ranges(self, file, file_size, ranges):
849
"""Refuses the multiple ranges request"""
852
self.send_error(416, "Requested range not satisfiable")
854
(start, end) = ranges[0]
855
return self.get_single_range(file, file_size, start, end)
858
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
859
"""Test readv against a server which only accept single range requests"""
861
_req_handler_class = SingleOnlyRangeRequestHandler
864
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
865
"""Ignore range requests without notice"""
868
# Update the statistics
869
self.server.test_case_server.GET_request_nb += 1
870
# Just bypass the range handling done by TestingHTTPRequestHandler
871
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"""
874
619
class TestNoRangeRequestServer(TestRangeRequestServer):
875
620
"""Test readv against a server which do not accept range requests"""
877
_req_handler_class = NoRangeRequestHandler
880
class MultipleRangeWithoutContentLengthRequestHandler(
881
http_server.TestingHTTPRequestHandler):
882
"""Reply to multiple range requests without content length header."""
884
def get_multiple_ranges(self, file, file_size, ranges):
885
self.send_response(206)
886
self.send_header('Accept-Ranges', 'bytes')
887
boundary = "%d" % random.randint(0,0x7FFFFFFF)
888
self.send_header("Content-Type",
889
"multipart/byteranges; boundary=%s" % boundary)
891
for (start, end) in ranges:
892
self.wfile.write("--%s\r\n" % boundary)
893
self.send_header("Content-type", 'application/octet-stream')
894
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
898
self.send_range_content(file, start, end - start + 1)
900
self.wfile.write("--%s\r\n" % boundary)
903
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
905
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
908
class TruncatedMultipleRangeRequestHandler(
909
http_server.TestingHTTPRequestHandler):
910
"""Reply to multiple range requests truncating the last ones.
912
This server generates responses whose Content-Length describes all the
913
ranges, but fail to include the last ones leading to client short reads.
914
This has been observed randomly with lighttpd (bug #179368).
917
_truncated_ranges = 2
919
def get_multiple_ranges(self, file, file_size, ranges):
920
self.send_response(206)
921
self.send_header('Accept-Ranges', 'bytes')
923
self.send_header('Content-Type',
924
'multipart/byteranges; boundary=%s' % boundary)
925
boundary_line = '--%s\r\n' % boundary
926
# Calculate the Content-Length
928
for (start, end) in ranges:
929
content_length += len(boundary_line)
930
content_length += self._header_line_length(
931
'Content-type', 'application/octet-stream')
932
content_length += self._header_line_length(
933
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
934
content_length += len('\r\n') # end headers
935
content_length += end - start # + 1
936
content_length += len(boundary_line)
937
self.send_header('Content-length', content_length)
940
# Send the multipart body
942
for (start, end) in ranges:
943
self.wfile.write(boundary_line)
944
self.send_header('Content-type', 'application/octet-stream')
945
self.send_header('Content-Range', 'bytes %d-%d/%d'
946
% (start, end, file_size))
948
if cur + self._truncated_ranges >= len(ranges):
949
# Abruptly ends the response and close the connection
950
self.close_connection = 1
952
self.send_range_content(file, start, end - start + 1)
955
self.wfile.write(boundary_line)
958
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
960
_req_handler_class = TruncatedMultipleRangeRequestHandler
963
super(TestTruncatedMultipleRangeServer, self).setUp()
964
self.build_tree_contents([('a', '0123456789')],)
966
def test_readv_with_short_reads(self):
967
server = self.get_readonly_server()
968
t = self._transport(server.get_url())
969
# Force separate ranges for each offset
970
t._bytes_to_read_before_seek = 0
971
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
972
self.assertEqual((0, '0'), ireadv.next())
973
self.assertEqual((2, '2'), ireadv.next())
974
if not self._testing_pycurl():
975
# Only one request have been issued so far (except for pycurl that
976
# try to read the whole response at once)
977
self.assertEqual(1, server.GET_request_nb)
978
self.assertEqual((4, '45'), ireadv.next())
979
self.assertEqual((9, '9'), ireadv.next())
980
# Both implementations issue 3 requests but:
981
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
983
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
984
self.assertEqual(3, server.GET_request_nb)
985
# Finally the client have tried a single range request and stays in
987
self.assertEqual('single', t._range_hint)
989
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
990
"""Errors out when range specifiers exceed the limit"""
992
def get_multiple_ranges(self, file, file_size, ranges):
993
"""Refuses the multiple ranges request"""
994
tcs = self.server.test_case_server
995
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
997
# Emulate apache behavior
998
self.send_error(400, "Bad Request")
1000
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1001
self, file, file_size, ranges)
1004
class LimitedRangeHTTPServer(http_server.HttpServer):
1005
"""An HttpServer erroring out on requests with too much range specifiers"""
1007
def __init__(self, request_handler=LimitedRangeRequestHandler,
1008
protocol_version=None,
1010
http_server.HttpServer.__init__(self, request_handler,
1011
protocol_version=protocol_version)
1012
self.range_limit = range_limit
1015
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1016
"""Tests readv requests against a server erroring out on too much ranges."""
1018
# Requests with more range specifiers will error out
1021
622
def create_transport_readonly_server(self):
1022
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1023
protocol_version=self._protocol_version)
1025
def get_transport(self):
1026
return self._transport(self.get_readonly_server().get_url())
1029
http_utils.TestCaseWithWebserver.setUp(self)
1030
# We need to manipulate ranges that correspond to real chunks in the
1031
# response, so we build a content appropriately.
1032
filler = ''.join(['abcdefghij' for x in range(102)])
1033
content = ''.join(['%04d' % v + filler for v in range(16)])
1034
self.build_tree_contents([('a', content)],)
1036
def test_few_ranges(self):
1037
t = self.get_transport()
1038
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1039
self.assertEqual(l[0], (0, '0000'))
1040
self.assertEqual(l[1], (1024, '0001'))
1041
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1043
def test_more_ranges(self):
1044
t = self.get_transport()
1045
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1046
self.assertEqual(l[0], (0, '0000'))
1047
self.assertEqual(l[1], (1024, '0001'))
1048
self.assertEqual(l[2], (4096, '0004'))
1049
self.assertEqual(l[3], (8192, '0008'))
1050
# The server will refuse to serve the first request (too much ranges),
1051
# a second request will succeed.
1052
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1055
class TestHttpProxyWhiteBox(tests.TestCase):
1056
"""Whitebox test proxy http authorization.
1058
Only the urllib implementation is tested here.
1062
tests.TestCase.setUp(self)
1067
tests.TestCase.tearDown(self)
1069
def _install_env(self, env):
1070
for name, value in env.iteritems():
1071
self._old_env[name] = osutils.set_or_unset_env(name, value)
1073
def _restore_env(self):
1074
for name, value in self._old_env.iteritems():
1075
osutils.set_or_unset_env(name, value)
1077
def _proxied_request(self):
1078
handler = _urllib2_wrappers.ProxyHandler()
1079
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
1080
handler.set_proxy(request, 'http')
1083
def test_empty_user(self):
1084
self._install_env({'http_proxy': 'http://bar.com'})
1085
request = self._proxied_request()
1086
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1088
def test_invalid_proxy(self):
1089
"""A proxy env variable without scheme"""
1090
self._install_env({'http_proxy': 'host:1234'})
1091
self.assertRaises(errors.InvalidURL, self._proxied_request)
1094
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):
1095
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.
1097
649
Be aware that we do not setup a real proxy here. Instead, we
1098
650
check that the *connection* goes through the proxy by serving
1099
651
different content (the faked proxy server append '-proxied'
1193
731
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
1194
732
'NO_PROXY': self.no_proxy_host})
1196
def test_http_proxy_without_scheme(self):
1197
if self._testing_pycurl():
1198
# pycurl *ignores* invalid proxy env variables. If that ever change
1199
# in the future, this test will fail indicating that pycurl do not
1200
# ignore anymore such variables.
1201
self.not_proxied_in_env({'http_proxy': self.proxy_address})
1203
self.assertRaises(errors.InvalidURL,
1204
self.proxied_in_env,
1205
{'http_proxy': self.proxy_address})
1208
class TestRanges(http_utils.TestCaseWithWebserver):
1209
"""Test the Range header in GET methods."""
1212
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)
1213
776
self.build_tree_contents([('a', '0123456789')],)
1214
777
server = self.get_readonly_server()
1215
778
self.transport = self._transport(server.get_url())
1217
def create_transport_readonly_server(self):
1218
return http_server.HttpServer(protocol_version=self._protocol_version)
1220
def _file_contents(self, relpath, ranges):
1221
offsets = [ (start, end - start + 1) for start, end in ranges]
1222
coalesce = self.transport._coalesce_offsets
1223
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1224
code, data = self.transport._get(relpath, coalesced)
1225
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1226
for start, end in ranges:
1228
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)
1230
787
def _file_tail(self, relpath, tail_amount):
1231
code, data = self.transport._get(relpath, [], tail_amount)
1232
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1233
data.seek(-tail_amount, 2)
1234
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)
1236
793
def test_range_header(self):
1238
795
map(self.assertEqual,['0', '234'],
1239
796
list(self._file_contents('a', [(0,0), (2,4)])),)
1241
def test_range_header_tail(self):
1242
798
self.assertEqual('789', self._file_tail('a', 3))
1244
def test_syntactically_invalid_range_header(self):
1245
self.assertListRaises(errors.InvalidHttpRange,
1246
self._file_contents, 'a', [(4, 3)])
1248
def test_semantically_invalid_range_header(self):
1249
self.assertListRaises(errors.InvalidHttpRange,
1250
self._file_contents, 'a', [(42, 128)])
1253
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1254
"""Test redirection between http servers."""
1256
def create_transport_secondary_server(self):
1257
"""Create the secondary server redirecting to the primary server"""
1258
new = self.get_readonly_server()
1260
redirecting = http_utils.HTTPServerRedirecting(
1261
protocol_version=self._protocol_version)
1262
redirecting.redirect_to(new.host, new.port)
1266
super(TestHTTPRedirections, self).setUp()
1267
self.build_tree_contents([('a', '0123456789'),
1269
'# Bazaar revision bundle v0.9\n#\n')
1272
self.old_transport = self._transport(self.old_server.get_url())
1274
def test_redirected(self):
1275
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1276
t = self._transport(self.new_server.get_url())
1277
self.assertEqual('0123456789', t.get('a').read())
1279
def test_read_redirected_bundle_from_url(self):
1280
from bzrlib.bundle import read_bundle_from_url
1281
url = self.old_transport.abspath('bundle')
1282
bundle = read_bundle_from_url(url)
1283
# If read_bundle_from_url was successful we get an empty bundle
1284
self.assertEqual([], bundle.revisions)
1287
class RedirectedRequest(_urllib2_wrappers.Request):
1288
"""Request following redirections. """
1290
init_orig = _urllib2_wrappers.Request.__init__
1292
def __init__(self, method, url, *args, **kwargs):
1296
# Since the tests using this class will replace
1297
# _urllib2_wrappers.Request, we can't just call the base class __init__
1299
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1300
self.follow_redirections = True
1303
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1304
"""Test redirections.
1306
http implementations do not redirect silently anymore (they
1307
do not redirect at all in fact). The mechanism is still in
1308
place at the _urllib2_wrappers.Request level and these tests
1311
For the pycurl implementation
1312
the redirection have been deleted as we may deprecate pycurl
1313
and I have no place to keep a working implementation.
1318
if pycurl_present and self._transport == PyCurlTransport:
1319
raise tests.TestNotApplicable(
1320
"pycurl doesn't redirect silently annymore")
1321
super(TestHTTPSilentRedirections, self).setUp()
1322
self.setup_redirected_request()
1323
self.addCleanup(self.cleanup_redirected_request)
1324
self.build_tree_contents([('a','a'),
1326
('1/a', 'redirected once'),
1328
('2/a', 'redirected twice'),
1330
('3/a', 'redirected thrice'),
1332
('4/a', 'redirected 4 times'),
1334
('5/a', 'redirected 5 times'),
1337
self.old_transport = self._transport(self.old_server.get_url())
1339
def setup_redirected_request(self):
1340
self.original_class = _urllib2_wrappers.Request
1341
_urllib2_wrappers.Request = RedirectedRequest
1343
def cleanup_redirected_request(self):
1344
_urllib2_wrappers.Request = self.original_class
1346
def create_transport_secondary_server(self):
1347
"""Create the secondary server, redirections are defined in the tests"""
1348
return http_utils.HTTPServerRedirecting(
1349
protocol_version=self._protocol_version)
1351
def test_one_redirection(self):
1352
t = self.old_transport
1354
req = RedirectedRequest('GET', t.abspath('a'))
1355
req.follow_redirections = True
1356
new_prefix = 'http://%s:%s' % (self.new_server.host,
1357
self.new_server.port)
1358
self.old_server.redirections = \
1359
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1360
self.assertEquals('redirected once',t._perform(req).read())
1362
def test_five_redirections(self):
1363
t = self.old_transport
1365
req = RedirectedRequest('GET', t.abspath('a'))
1366
req.follow_redirections = True
1367
old_prefix = 'http://%s:%s' % (self.old_server.host,
1368
self.old_server.port)
1369
new_prefix = 'http://%s:%s' % (self.new_server.host,
1370
self.new_server.port)
1371
self.old_server.redirections = [
1372
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1373
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1374
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1375
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1376
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1378
self.assertEquals('redirected 5 times',t._perform(req).read())
1381
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1382
"""Test transport.do_catching_redirections."""
1385
super(TestDoCatchRedirections, self).setUp()
1386
self.build_tree_contents([('a', '0123456789'),],)
1388
self.old_transport = self._transport(self.old_server.get_url())
1390
def get_a(self, transport):
1391
return transport.get('a')
1393
def test_no_redirection(self):
1394
t = self._transport(self.new_server.get_url())
1396
# We use None for redirected so that we fail if redirected
1397
self.assertEquals('0123456789',
1398
transport.do_catching_redirections(
1399
self.get_a, t, None).read())
1401
def test_one_redirection(self):
1402
self.redirections = 0
1404
def redirected(transport, exception, redirection_notice):
1405
self.redirections += 1
1406
dir, file = urlutils.split(exception.target)
1407
return self._transport(dir)
1409
self.assertEquals('0123456789',
1410
transport.do_catching_redirections(
1411
self.get_a, self.old_transport, redirected).read())
1412
self.assertEquals(1, self.redirections)
1414
def test_redirection_loop(self):
1416
def redirected(transport, exception, redirection_notice):
1417
# By using the redirected url as a base dir for the
1418
# *old* transport, we create a loop: a => a/a =>
1420
return self.old_transport.clone(exception.target)
1422
self.assertRaises(errors.TooManyRedirections,
1423
transport.do_catching_redirections,
1424
self.get_a, self.old_transport, redirected)
1427
class TestAuth(http_utils.TestCaseWithWebserver):
1428
"""Test authentication scheme"""
1430
_auth_header = 'Authorization'
1431
_password_prompt_prefix = ''
1434
super(TestAuth, self).setUp()
1435
self.server = self.get_readonly_server()
1436
self.build_tree_contents([('a', 'contents of a\n'),
1437
('b', 'contents of b\n'),])
1439
def create_transport_readonly_server(self):
1440
if self._auth_scheme == 'basic':
1441
server = http_utils.HTTPBasicAuthServer(
1442
protocol_version=self._protocol_version)
1444
if self._auth_scheme != 'digest':
1445
raise AssertionError('Unknown auth scheme: %r'
1446
% self._auth_scheme)
1447
server = http_utils.HTTPDigestAuthServer(
1448
protocol_version=self._protocol_version)
1451
def _testing_pycurl(self):
1452
return pycurl_present and self._transport == PyCurlTransport
1454
def get_user_url(self, user=None, password=None):
1455
"""Build an url embedding user and password"""
1456
url = '%s://' % self.server._url_protocol
1457
if user is not None:
1459
if password is not None:
1460
url += ':' + password
1462
url += '%s:%s/' % (self.server.host, self.server.port)
1465
def get_user_transport(self, user=None, password=None):
1466
return self._transport(self.get_user_url(user, password))
1468
def test_no_user(self):
1469
self.server.add_user('joe', 'foo')
1470
t = self.get_user_transport()
1471
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1472
# Only one 'Authentication Required' error should occur
1473
self.assertEqual(1, self.server.auth_required_errors)
1475
def test_empty_pass(self):
1476
self.server.add_user('joe', '')
1477
t = self.get_user_transport('joe', '')
1478
self.assertEqual('contents of a\n', t.get('a').read())
1479
# Only one 'Authentication Required' error should occur
1480
self.assertEqual(1, self.server.auth_required_errors)
1482
def test_user_pass(self):
1483
self.server.add_user('joe', 'foo')
1484
t = self.get_user_transport('joe', 'foo')
1485
self.assertEqual('contents of a\n', t.get('a').read())
1486
# Only one 'Authentication Required' error should occur
1487
self.assertEqual(1, self.server.auth_required_errors)
1489
def test_unknown_user(self):
1490
self.server.add_user('joe', 'foo')
1491
t = self.get_user_transport('bill', 'foo')
1492
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1493
# Two 'Authentication Required' errors should occur (the
1494
# initial 'who are you' and 'I don't know you, who are
1496
self.assertEqual(2, self.server.auth_required_errors)
1498
def test_wrong_pass(self):
1499
self.server.add_user('joe', 'foo')
1500
t = self.get_user_transport('joe', 'bar')
1501
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1502
# Two 'Authentication Required' errors should occur (the
1503
# initial 'who are you' and 'this is not you, who are you')
1504
self.assertEqual(2, self.server.auth_required_errors)
1506
def test_prompt_for_password(self):
1507
if self._testing_pycurl():
1508
raise tests.TestNotApplicable(
1509
'pycurl cannot prompt, it handles auth by embedding'
1510
' user:pass in urls only')
1512
self.server.add_user('joe', 'foo')
1513
t = self.get_user_transport('joe', None)
1514
stdout = tests.StringIOWrapper()
1515
ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1516
self.assertEqual('contents of a\n',t.get('a').read())
1517
# stdin should be empty
1518
self.assertEqual('', ui.ui_factory.stdin.readline())
1519
self._check_password_prompt(t._unqualified_scheme, 'joe',
1521
# And we shouldn't prompt again for a different request
1522
# against the same transport.
1523
self.assertEqual('contents of b\n',t.get('b').read())
1525
# And neither against a clone
1526
self.assertEqual('contents of b\n',t2.get('b').read())
1527
# Only one 'Authentication Required' error should occur
1528
self.assertEqual(1, self.server.auth_required_errors)
1530
def _check_password_prompt(self, scheme, user, actual_prompt):
1531
expected_prompt = (self._password_prompt_prefix
1532
+ ("%s %s@%s:%d, Realm: '%s' password: "
1534
user, self.server.host, self.server.port,
1535
self.server.auth_realm)))
1536
self.assertEquals(expected_prompt, actual_prompt)
1538
def test_no_prompt_for_password_when_using_auth_config(self):
1539
if self._testing_pycurl():
1540
raise tests.TestNotApplicable(
1541
'pycurl does not support authentication.conf'
1542
' since it cannot prompt')
1546
stdin_content = 'bar\n' # Not the right password
1547
self.server.add_user(user, password)
1548
t = self.get_user_transport(user, None)
1549
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1550
stdout=tests.StringIOWrapper())
1551
# Create a minimal config file with the right password
1552
conf = config.AuthenticationConfig()
1553
conf._get_config().update(
1554
{'httptest': {'scheme': 'http', 'port': self.server.port,
1555
'user': user, 'password': password}})
1557
# Issue a request to the server to connect
1558
self.assertEqual('contents of a\n',t.get('a').read())
1559
# stdin should have been left untouched
1560
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1561
# Only one 'Authentication Required' error should occur
1562
self.assertEqual(1, self.server.auth_required_errors)
1564
def test_changing_nonce(self):
1565
if self._auth_scheme != 'digest':
1566
raise tests.TestNotApplicable('HTTP auth digest only test')
1567
if self._testing_pycurl():
1568
raise tests.KnownFailure(
1569
'pycurl does not handle a nonce change')
1570
self.server.add_user('joe', 'foo')
1571
t = self.get_user_transport('joe', 'foo')
1572
self.assertEqual('contents of a\n', t.get('a').read())
1573
self.assertEqual('contents of b\n', t.get('b').read())
1574
# Only one 'Authentication Required' error should have
1576
self.assertEqual(1, self.server.auth_required_errors)
1577
# The server invalidates the current nonce
1578
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1579
self.assertEqual('contents of a\n', t.get('a').read())
1580
# Two 'Authentication Required' errors should occur (the
1581
# initial 'who are you' and a second 'who are you' with the new nonce)
1582
self.assertEqual(2, self.server.auth_required_errors)
1586
class TestProxyAuth(TestAuth):
1587
"""Test proxy authentication schemes."""
1589
_auth_header = 'Proxy-authorization'
1590
_password_prompt_prefix='Proxy '
1593
super(TestProxyAuth, self).setUp()
1595
self.addCleanup(self._restore_env)
1596
# Override the contents to avoid false positives
1597
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1598
('b', 'not proxied contents of b\n'),
1599
('a-proxied', 'contents of a\n'),
1600
('b-proxied', 'contents of b\n'),
1603
def create_transport_readonly_server(self):
1604
if self._auth_scheme == 'basic':
1605
server = http_utils.ProxyBasicAuthServer(
1606
protocol_version=self._protocol_version)
1608
if self._auth_scheme != 'digest':
1609
raise AssertionError('Unknown auth scheme: %r'
1610
% self._auth_scheme)
1611
server = http_utils.ProxyDigestAuthServer(
1612
protocol_version=self._protocol_version)
1615
def get_user_transport(self, user=None, password=None):
1616
self._install_env({'all_proxy': self.get_user_url(user, password)})
1617
return self._transport(self.server.get_url())
1619
def _install_env(self, env):
1620
for name, value in env.iteritems():
1621
self._old_env[name] = osutils.set_or_unset_env(name, value)
1623
def _restore_env(self):
1624
for name, value in self._old_env.iteritems():
1625
osutils.set_or_unset_env(name, value)
1627
def test_empty_pass(self):
1628
if self._testing_pycurl():
1630
if pycurl.version_info()[1] < '7.16.0':
1631
raise tests.KnownFailure(
1632
'pycurl < 7.16.0 does not handle empty proxy passwords')
1633
super(TestProxyAuth, self).test_empty_pass()
1636
class SampleSocket(object):
1637
"""A socket-like object for use in testing the HTTP request handler."""
1639
def __init__(self, socket_read_content):
1640
"""Constructs a sample socket.
1642
:param socket_read_content: a byte sequence
1644
# Use plain python StringIO so we can monkey-patch the close method to
1645
# not discard the contents.
1646
from StringIO import StringIO
1647
self.readfile = StringIO(socket_read_content)
1648
self.writefile = StringIO()
1649
self.writefile.close = lambda: None
1651
def makefile(self, mode='r', bufsize=None):
1653
return self.readfile
1655
return self.writefile
1658
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1661
super(SmartHTTPTunnellingTest, self).setUp()
1662
# We use the VFS layer as part of HTTP tunnelling tests.
1663
self._captureVar('BZR_NO_SMART_VFS', None)
1664
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1666
def create_transport_readonly_server(self):
1667
return http_utils.HTTPServerWithSmarts(
1668
protocol_version=self._protocol_version)
1670
def test_bulk_data(self):
1671
# We should be able to send and receive bulk data in a single message.
1672
# The 'readv' command in the smart protocol both sends and receives
1673
# bulk data, so we use that.
1674
self.build_tree(['data-file'])
1675
http_server = self.get_readonly_server()
1676
http_transport = self._transport(http_server.get_url())
1677
medium = http_transport.get_smart_medium()
1678
# Since we provide the medium, the url below will be mostly ignored
1679
# during the test, as long as the path is '/'.
1680
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1683
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1685
def test_http_send_smart_request(self):
1687
post_body = 'hello\n'
1688
expected_reply_body = 'ok\x012\n'
1690
http_server = self.get_readonly_server()
1691
http_transport = self._transport(http_server.get_url())
1692
medium = http_transport.get_smart_medium()
1693
response = medium.send_http_smart_request(post_body)
1694
reply_body = response.read()
1695
self.assertEqual(expected_reply_body, reply_body)
1697
def test_smart_http_server_post_request_handler(self):
1698
httpd = self.get_readonly_server()._get_httpd()
1700
socket = SampleSocket(
1701
'POST /.bzr/smart %s \r\n' % self._protocol_version
1702
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1704
+ 'Content-Length: 6\r\n'
1707
# Beware: the ('localhost', 80) below is the
1708
# client_address parameter, but we don't have one because
1709
# we have defined a socket which is not bound to an
1710
# address. The test framework never uses this client
1711
# address, so far...
1712
request_handler = http_utils.SmartRequestHandler(socket,
1715
response = socket.writefile.getvalue()
1716
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1717
# This includes the end of the HTTP headers, and all the body.
1718
expected_end_of_response = '\r\n\r\nok\x012\n'
1719
self.assertEndsWith(response, expected_end_of_response)
1722
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1723
"""No smart server here request handler."""
1726
self.send_error(403, "Forbidden")
1729
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1730
"""Test smart client behaviour against an http server without smarts."""
1732
_req_handler_class = ForbiddenRequestHandler
1734
def test_probe_smart_server(self):
1735
"""Test error handling against server refusing smart requests."""
1736
server = self.get_readonly_server()
1737
t = self._transport(server.get_url())
1738
# No need to build a valid smart request here, the server will not even
1739
# try to interpret it.
1740
self.assertRaises(errors.SmartProtocolError,
1741
t.send_http_smart_request, 'whatever')
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
