14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
17
"""Tests for HTTP implementations.
19
This module defines a load_tests() method that parametrize tests classes for
20
transport implementation, http protocol versions and authentication schemes.
20
23
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
24
# TODO: What about renaming to bzrlib.tests.transport.http ?
23
26
from cStringIO import StringIO
30
import SimpleHTTPServer
31
36
from bzrlib import (
37
45
from bzrlib.tests import (
43
from bzrlib.tests.HttpServer import (
48
from bzrlib.tests.HTTPTestUtil import (
49
BadProtocolRequestHandler,
50
BadStatusRequestHandler,
51
ForbiddenRequestHandler,
54
HTTPServerRedirecting,
55
InvalidStatusRequestHandler,
56
NoRangeRequestHandler,
58
ProxyDigestAuthServer,
60
SingleRangeRequestHandler,
61
SingleOnlyRangeRequestHandler,
62
TestCaseWithRedirectedWebserver,
63
TestCaseWithTwoWebservers,
64
TestCaseWithWebserver,
67
49
from bzrlib.transport import (
68
do_catching_redirections,
72
53
from bzrlib.transport.http import (
77
from bzrlib.transport.http._urllib import HttpTransport_urllib
78
from bzrlib.transport.http._urllib2_wrappers import (
60
from bzrlib.transport.http._pycurl import PyCurlTransport
62
except errors.DependencyNotPresent:
63
pycurl_present = False
66
class TransportAdapter(tests.TestScenarioApplier):
67
"""Generate the same test for each transport implementation."""
70
transport_scenarios = [
71
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
72
_server=http_server.HttpServer_urllib,
73
_qualified_prefix='http+urllib',)),
76
transport_scenarios.append(
77
('pycurl', dict(_transport=PyCurlTransport,
78
_server=http_server.HttpServer_PyCurl,
79
_qualified_prefix='http+pycurl',)))
80
self.scenarios = transport_scenarios
83
class TransportProtocolAdapter(TransportAdapter):
84
"""Generate the same test for each protocol implementation.
86
In addition to the transport adaptatation that we inherit from.
90
super(TransportProtocolAdapter, self).__init__()
91
protocol_scenarios = [
92
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
93
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
95
self.scenarios = tests.multiply_scenarios(self.scenarios,
99
class TransportProtocolAuthenticationAdapter(TransportProtocolAdapter):
100
"""Generate the same test for each authentication scheme implementation.
102
In addition to the protocol adaptatation that we inherit from.
106
super(TransportProtocolAuthenticationAdapter, self).__init__()
107
auth_scheme_scenarios = [
108
('basic', dict(_auth_scheme='basic')),
109
('digest', dict(_auth_scheme='digest')),
112
self.scenarios = tests.multiply_scenarios(self.scenarios,
113
auth_scheme_scenarios)
115
def load_tests(standard_tests, module, loader):
116
"""Multiply tests for http clients and protocol versions."""
117
# one for each transport
118
t_adapter = TransportAdapter()
119
t_classes= (TestHttpTransportRegistration,
120
TestHttpTransportUrls,
122
is_testing_for_transports = tests.condition_isinstance(t_classes)
124
# multiplied by one for each protocol version
125
tp_adapter = TransportProtocolAdapter()
126
tp_classes= (SmartHTTPTunnellingTest,
127
TestDoCatchRedirections,
129
TestHTTPRedirections,
130
TestHTTPSilentRedirections,
131
TestLimitedRangeRequestServer,
135
TestSpecificRequestHandler,
137
is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)
139
# multiplied by one for each authentication scheme
140
tpa_adapter = TransportProtocolAuthenticationAdapter()
141
tpa_classes = (TestAuth,
143
is_also_testing_for_authentication = tests.condition_isinstance(
146
result = loader.suiteClass()
147
for test_class in tests.iter_suite_tests(standard_tests):
148
# Each test class is either standalone or testing for some combination
149
# of transport, protocol version, authentication scheme. Use the right
150
# adpater (or none) depending on the class.
151
if is_testing_for_transports(test_class):
152
result.addTests(t_adapter.adapt(test_class))
153
elif is_also_testing_for_protocols(test_class):
154
result.addTests(tp_adapter.adapt(test_class))
155
elif is_also_testing_for_authentication(test_class):
156
result.addTests(tpa_adapter.adapt(test_class))
158
result.addTest(test_class)
85
162
class FakeManager(object):
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)
151
304
class TestWithTransport_pycurl(object):
152
305
"""Test case to inherit from if pycurl is present"""
255
387
except ImportError:
256
raise TestSkipped('pycurl not present')
257
# Now that we have pycurl imported, we can fake its version_info
258
# This was taken from a windows pycurl without SSL
260
pycurl.version_info = lambda : (2,
268
('ftp', 'gopher', 'telnet',
269
'dict', 'ldap', 'http', 'file'),
273
self.assertRaises(errors.DependencyNotPresent, self._transport,
274
'https://launchpad.net')
276
class TestHttpConnections(object):
277
"""Test the http connections.
279
This MUST be used by daughter classes that also inherit from
280
TestCaseWithWebserver.
282
We can't inherit directly from TestCaseWithWebserver or the
283
test framework will try to create an instance which cannot
284
run, its implementation being incomplete.
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."""
288
TestCaseWithWebserver.setUp(self)
289
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
419
http_utils.TestCaseWithWebserver.setUp(self)
420
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
290
421
transport=self.get_transport())
292
423
def test_http_has(self):
338
469
socket.setdefaulttimeout(default_timeout)
341
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
342
"""Test http connections with urllib"""
344
_transport = HttpTransport_urllib
348
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
350
TestCaseWithWebserver):
351
"""Test http connections with pycurl"""
354
class TestHttpTransportRegistration(TestCase):
472
class TestHttpTransportRegistration(tests.TestCase):
355
473
"""Test registrations of various http implementations"""
357
475
def test_http_registered(self):
358
# urlllib should always be present
359
t = get_transport('http+urllib://bzr.google.com/')
360
self.assertIsInstance(t, Transport)
361
self.assertIsInstance(t, HttpTransport_urllib)
364
class TestOffsets(TestCase):
365
"""Test offsets_to_ranges method"""
367
def test_offsets_to_ranges_simple(self):
368
to_range = HttpTransportBase.offsets_to_ranges
369
ranges = to_range([(10, 1)])
370
self.assertEqual([[10, 10]], ranges)
372
ranges = to_range([(0, 1), (1, 1)])
373
self.assertEqual([[0, 1]], ranges)
375
ranges = to_range([(1, 1), (0, 1)])
376
self.assertEqual([[0, 1]], ranges)
378
def test_offset_to_ranges_overlapped(self):
379
to_range = HttpTransportBase.offsets_to_ranges
381
ranges = to_range([(10, 1), (20, 2), (22, 5)])
382
self.assertEqual([[10, 10], [20, 26]], ranges)
384
ranges = to_range([(10, 1), (11, 2), (22, 5)])
385
self.assertEqual([[10, 12], [22, 26]], ranges)
388
class TestPost(object):
390
def _test_post_body_is_received(self, scheme):
476
t = transport.get_transport('%s://foo.com/' % self._qualified_prefix)
477
self.assertIsInstance(t, transport.Transport)
478
self.assertIsInstance(t, self._transport)
481
class TestPost(tests.TestCase):
483
def test_post_body_is_received(self):
391
484
server = RecordingServer(expect_body_tail='end-of-body')
393
486
self.addCleanup(server.tearDown)
487
scheme = self._qualified_prefix
394
488
url = '%s://%s:%s/' % (scheme, server.host, server.port)
396
http_transport = get_transport(url)
397
except errors.UnsupportedProtocol:
398
raise TestSkipped('%s not available' % scheme)
489
http_transport = self._transport(url)
399
490
code, response = http_transport._post('abc def end-of-body')
401
492
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
673
767
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
674
768
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)
677
839
class TestSingleRangeRequestServer(TestRangeRequestServer):
678
840
"""Test readv against a server which accept only single range requests"""
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleRangeRequestHandler)
684
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
688
_transport = HttpTransport_urllib
691
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleRangeRequestServer,
693
TestCaseWithWebserver):
694
"""Tests single range requests accepting server for pycurl implementation"""
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)
697
858
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
698
859
"""Test readv against a server which only accept single range requests"""
700
def create_transport_readonly_server(self):
701
return HttpServer(SingleOnlyRangeRequestHandler)
704
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests single range requests accepting server for urllib implementation"""
708
_transport = HttpTransport_urllib
711
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestSingleOnlyRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests single range requests accepting server for pycurl implementation"""
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)
717
874
class TestNoRangeRequestServer(TestRangeRequestServer):
718
875
"""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
907
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
908
"""Errors out when range specifiers exceed the limit"""
910
def get_multiple_ranges(self, file, file_size, ranges):
911
"""Refuses the multiple ranges request"""
912
tcs = self.server.test_case_server
913
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
915
# Emulate apache behavior
916
self.send_error(400, "Bad Request")
918
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
919
self, file, file_size, ranges)
922
class LimitedRangeHTTPServer(http_server.HttpServer):
923
"""An HttpServer erroring out on requests with too much range specifiers"""
925
def __init__(self, request_handler=LimitedRangeRequestHandler,
926
protocol_version=None,
928
http_server.HttpServer.__init__(self, request_handler,
929
protocol_version=protocol_version)
930
self.range_limit = range_limit
933
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
934
"""Tests readv requests against a server erroring out on too much ranges."""
936
# Requests with more range specifiers will error out
720
939
def create_transport_readonly_server(self):
721
return HttpServer(NoRangeRequestHandler)
724
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
725
TestCaseWithWebserver):
726
"""Tests range requests refusing server for urllib implementation"""
728
_transport = HttpTransport_urllib
731
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
732
TestNoRangeRequestServer,
733
TestCaseWithWebserver):
734
"""Tests range requests refusing server for pycurl implementation"""
737
class TestHttpProxyWhiteBox(TestCase):
940
return LimitedRangeHTTPServer(range_limit=self.range_limit,
941
protocol_version=self._protocol_version)
943
def get_transport(self):
944
return self._transport(self.get_readonly_server().get_url())
947
http_utils.TestCaseWithWebserver.setUp(self)
948
# We need to manipulate ranges that correspond to real chunks in the
949
# response, so we build a content appropriately.
950
filler = ''.join(['abcdefghij' for x in range(102)])
951
content = ''.join(['%04d' % v + filler for v in range(16)])
952
self.build_tree_contents([('a', content)],)
954
def test_few_ranges(self):
955
t = self.get_transport()
956
l = list(t.readv('a', ((0, 4), (1024, 4), )))
957
self.assertEqual(l[0], (0, '0000'))
958
self.assertEqual(l[1], (1024, '0001'))
959
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
961
def test_more_ranges(self):
962
t = self.get_transport()
963
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
964
self.assertEqual(l[0], (0, '0000'))
965
self.assertEqual(l[1], (1024, '0001'))
966
self.assertEqual(l[2], (4096, '0004'))
967
self.assertEqual(l[3], (8192, '0008'))
968
# The server will refuse to serve the first request (too much ranges),
969
# a second request will succeeds.
970
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
973
class TestHttpProxyWhiteBox(tests.TestCase):
738
974
"""Whitebox test proxy http authorization.
740
976
Only the urllib implementation is tested here.
980
tests.TestCase.setUp(self)
745
981
self._old_env = {}
747
983
def tearDown(self):
869
1111
'NO_PROXY': self.no_proxy_host})
871
1113
def test_http_proxy_without_scheme(self):
872
self.assertRaises(errors.InvalidURL,
874
{'http_proxy': self.proxy_address})
877
class TestProxyHttpServer_urllib(TestProxyHttpServer,
878
TestCaseWithTwoWebservers):
879
"""Tests proxy server for urllib implementation"""
881
_transport = HttpTransport_urllib
884
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
886
TestCaseWithTwoWebservers):
887
"""Tests proxy server for pycurl implementation"""
890
TestProxyHttpServer.setUp(self)
891
# Oh my ! pycurl does not check for the port as part of
892
# no_proxy :-( So we just test the host part
893
self.no_proxy_host = 'localhost'
895
def test_HTTP_PROXY(self):
896
# pycurl do not check HTTP_PROXY for security reasons
897
# (for use in a CGI context that we do not care
898
# about. Should we ?)
901
def test_HTTP_PROXY_with_NO_PROXY(self):
904
def test_http_proxy_without_scheme(self):
905
# pycurl *ignores* invalid proxy env variables. If that
906
# ever change in the future, this test will fail
907
# indicating that pycurl do not ignore anymore such
909
self.not_proxied_in_env({'http_proxy': self.proxy_address})
912
class TestRanges(object):
913
"""Test the Range header in GET methods..
915
This MUST be used by daughter classes that also inherit from
916
TestCaseWithWebserver.
918
We can't inherit directly from TestCaseWithWebserver or the
919
test framework will try to create an instance which cannot
920
run, its implementation being incomplete.
924
TestCaseWithWebserver.setUp(self)
1114
if self._testing_pycurl():
1115
# pycurl *ignores* invalid proxy env variables. If that ever change
1116
# in the future, this test will fail indicating that pycurl do not
1117
# ignore anymore such variables.
1118
self.not_proxied_in_env({'http_proxy': self.proxy_address})
1120
self.assertRaises(errors.InvalidURL,
1121
self.proxied_in_env,
1122
{'http_proxy': self.proxy_address})
1125
class TestRanges(http_utils.TestCaseWithWebserver):
1126
"""Test the Range header in GET methods."""
1129
http_utils.TestCaseWithWebserver.setUp(self)
925
1130
self.build_tree_contents([('a', '0123456789')],)
926
1131
server = self.get_readonly_server()
927
1132
self.transport = self._transport(server.get_url())
929
def _file_contents(self, relpath, ranges, tail_amount=0):
930
code, data = self.transport._get(relpath, ranges)
931
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
932
for start, end in ranges:
934
yield data.read(end - start + 1)
1134
def create_transport_readonly_server(self):
1135
return http_server.HttpServer(protocol_version=self._protocol_version)
1137
def _file_contents(self, relpath, ranges):
1138
offsets = [ (start, end - start + 1) for start, end in ranges]
1139
coalesce = self.transport._coalesce_offsets
1140
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1141
code, data = self.transport._get(relpath, coalesced)
1142
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1143
for start, end in ranges:
1145
yield data.read(end - start + 1)
936
1147
def _file_tail(self, relpath, tail_amount):
937
code, data = self.transport._get(relpath, [], tail_amount)
938
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
939
data.seek(-tail_amount + 1, 2)
940
return data.read(tail_amount)
1148
code, data = self.transport._get(relpath, [], tail_amount)
1149
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1150
data.seek(-tail_amount, 2)
1151
return data.read(tail_amount)
942
1153
def test_range_header(self):
944
1155
map(self.assertEqual,['0', '234'],
945
1156
list(self._file_contents('a', [(0,0), (2,4)])),)
1158
def test_range_header_tail(self):
947
1159
self.assertEqual('789', self._file_tail('a', 3))
948
# Syntactically invalid range
949
self.assertRaises(errors.InvalidRange,
950
self.transport._get, 'a', [(4, 3)])
951
# Semantically invalid range
952
self.assertRaises(errors.InvalidRange,
953
self.transport._get, 'a', [(42, 128)])
956
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
957
"""Test the Range header in GET methods for urllib implementation"""
959
_transport = HttpTransport_urllib
962
class TestRanges_pycurl(TestWithTransport_pycurl,
964
TestCaseWithWebserver):
965
"""Test the Range header in GET methods for pycurl implementation"""
968
class TestHTTPRedirections(object):
969
"""Test redirection between http servers.
971
This MUST be used by daughter classes that also inherit from
972
TestCaseWithRedirectedWebserver.
974
We can't inherit directly from TestCaseWithTwoWebservers or the
975
test framework will try to create an instance which cannot
976
run, its implementation being incomplete.
1161
def test_syntactically_invalid_range_header(self):
1162
self.assertListRaises(errors.InvalidHttpRange,
1163
self._file_contents, 'a', [(4, 3)])
1165
def test_semantically_invalid_range_header(self):
1166
self.assertListRaises(errors.InvalidHttpRange,
1167
self._file_contents, 'a', [(42, 128)])
1170
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1171
"""Test redirection between http servers."""
979
1173
def create_transport_secondary_server(self):
980
1174
"""Create the secondary server redirecting to the primary server"""
981
1175
new = self.get_readonly_server()
983
redirecting = HTTPServerRedirecting()
1177
redirecting = http_utils.HTTPServerRedirecting(
1178
protocol_version=self._protocol_version)
984
1179
redirecting.redirect_to(new.host, new.port)
985
1180
return redirecting
1154
1337
return self.old_transport.clone(exception.target)
1156
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1339
self.assertRaises(errors.TooManyRedirections,
1340
transport.do_catching_redirections,
1157
1341
self.get_a, self.old_transport, redirected)
1160
class TestAuth(object):
1161
"""Test some authentication scheme specified by daughter class.
1344
class TestAuth(http_utils.TestCaseWithWebserver):
1345
"""Test authentication scheme"""
1163
This MUST be used by daughter classes that also inherit from
1164
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1347
_auth_header = 'Authorization'
1348
_password_prompt_prefix = ''
1167
1350
def setUp(self):
1168
"""Set up the test environment
1170
Daughter classes should set up their own environment
1171
(including self.server) and explicitely call this
1172
method. This is needed because we want to reuse the same
1173
tests for proxy and no-proxy accesses which have
1174
different ways of setting self.server.
1351
super(TestAuth, self).setUp()
1352
self.server = self.get_readonly_server()
1176
1353
self.build_tree_contents([('a', 'contents of a\n'),
1177
1354
('b', 'contents of b\n'),])
1178
self.old_factory = ui.ui_factory
1179
self.old_stdout = sys.stdout
1180
sys.stdout = StringIOWrapper()
1181
self.addCleanup(self.restoreUIFactory)
1183
def restoreUIFactory(self):
1184
ui.ui_factory = self.old_factory
1185
sys.stdout = self.old_stdout
1356
def create_transport_readonly_server(self):
1357
if self._auth_scheme == 'basic':
1358
server = http_utils.HTTPBasicAuthServer(
1359
protocol_version=self._protocol_version)
1361
if self._auth_scheme != 'digest':
1362
raise AssertionError('Unknown auth scheme: %r'
1363
% self._auth_scheme)
1364
server = http_utils.HTTPDigestAuthServer(
1365
protocol_version=self._protocol_version)
1368
def _testing_pycurl(self):
1369
return pycurl_present and self._transport == PyCurlTransport
1187
1371
def get_user_url(self, user=None, password=None):
1188
1372
"""Build an url embedding user and password"""
1249
1444
# Only one 'Authentication Required' error should occur
1250
1445
self.assertEqual(1, self.server.auth_required_errors)
1253
class TestHTTPAuth(TestAuth):
1254
"""Test HTTP authentication schemes.
1256
Daughter classes MUST inherit from TestCaseWithWebserver too.
1259
_auth_header = 'Authorization'
1262
TestCaseWithWebserver.setUp(self)
1263
self.server = self.get_readonly_server()
1264
TestAuth.setUp(self)
1266
def get_user_transport(self, user=None, password=None):
1267
return self._transport(self.get_user_url(user, password))
1447
def _check_password_prompt(self, scheme, user, actual_prompt):
1448
expected_prompt = (self._password_prompt_prefix
1449
+ ("%s %s@%s:%d, Realm: '%s' password: "
1451
user, self.server.host, self.server.port,
1452
self.server.auth_realm)))
1453
self.assertEquals(expected_prompt, actual_prompt)
1455
def test_no_prompt_for_password_when_using_auth_config(self):
1456
if self._testing_pycurl():
1457
raise tests.TestNotApplicable(
1458
'pycurl does not support authentication.conf'
1459
' since it cannot prompt')
1463
stdin_content = 'bar\n' # Not the right password
1464
self.server.add_user(user, password)
1465
t = self.get_user_transport(user, None)
1466
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1467
stdout=tests.StringIOWrapper())
1468
# Create a minimal config file with the right password
1469
conf = config.AuthenticationConfig()
1470
conf._get_config().update(
1471
{'httptest': {'scheme': 'http', 'port': self.server.port,
1472
'user': user, 'password': password}})
1474
# Issue a request to the server to connect
1475
self.assertEqual('contents of a\n',t.get('a').read())
1476
# stdin should have been left untouched
1477
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1478
# Only one 'Authentication Required' error should occur
1479
self.assertEqual(1, self.server.auth_required_errors)
1481
def test_changing_nonce(self):
1482
if self._auth_scheme != 'digest':
1483
raise tests.TestNotApplicable('HTTP auth digest only test')
1484
if self._testing_pycurl():
1485
raise tests.KnownFailure(
1486
'pycurl does not handle a nonce change')
1487
self.server.add_user('joe', 'foo')
1488
t = self.get_user_transport('joe', 'foo')
1489
self.assertEqual('contents of a\n', t.get('a').read())
1490
self.assertEqual('contents of b\n', t.get('b').read())
1491
# Only one 'Authentication Required' error should have
1493
self.assertEqual(1, self.server.auth_required_errors)
1494
# The server invalidates the current nonce
1495
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1496
self.assertEqual('contents of a\n', t.get('a').read())
1497
# Two 'Authentication Required' errors should occur (the
1498
# initial 'who are you' and a second 'who are you' with the new nonce)
1499
self.assertEqual(2, self.server.auth_required_errors)
1270
1503
class TestProxyAuth(TestAuth):
1271
"""Test proxy authentication schemes.
1504
"""Test proxy authentication schemes."""
1273
Daughter classes MUST also inherit from TestCaseWithWebserver.
1275
1506
_auth_header = 'Proxy-authorization'
1507
_password_prompt_prefix='Proxy '
1277
1509
def setUp(self):
1278
TestCaseWithWebserver.setUp(self)
1279
self.server = self.get_readonly_server()
1510
super(TestProxyAuth, self).setUp()
1280
1511
self._old_env = {}
1281
1512
self.addCleanup(self._restore_env)
1282
TestAuth.setUp(self)
1283
1513
# Override the contents to avoid false positives
1284
1514
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1285
1515
('b', 'not proxied contents of b\n'),
1299
1541
for name, value in self._old_env.iteritems():
1300
1542
osutils.set_or_unset_env(name, value)
1303
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1304
"""Test http basic authentication scheme"""
1306
_transport = HttpTransport_urllib
1308
def create_transport_readonly_server(self):
1309
return HTTPBasicAuthServer()
1312
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1313
"""Test proxy basic authentication scheme"""
1315
_transport = HttpTransport_urllib
1317
def create_transport_readonly_server(self):
1318
return ProxyBasicAuthServer()
1321
class TestDigestAuth(object):
1322
"""Digest Authentication specific tests"""
1324
def test_changing_nonce(self):
1325
self.server.add_user('joe', 'foo')
1326
t = self.get_user_transport('joe', 'foo')
1327
self.assertEqual('contents of a\n', t.get('a').read())
1328
self.assertEqual('contents of b\n', t.get('b').read())
1329
# Only one 'Authentication Required' error should have
1331
self.assertEqual(1, self.server.auth_required_errors)
1332
# The server invalidates the current nonce
1333
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1334
self.assertEqual('contents of a\n', t.get('a').read())
1335
# Two 'Authentication Required' errors should occur (the
1336
# initial 'who are you' and a second 'who are you' with the new nonce)
1337
self.assertEqual(2, self.server.auth_required_errors)
1340
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1341
"""Test http digest authentication scheme"""
1343
_transport = HttpTransport_urllib
1345
def create_transport_readonly_server(self):
1346
return HTTPDigestAuthServer()
1349
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1350
TestCaseWithWebserver):
1351
"""Test proxy digest authentication scheme"""
1353
_transport = HttpTransport_urllib
1355
def create_transport_readonly_server(self):
1356
return ProxyDigestAuthServer()
1544
def test_empty_pass(self):
1545
if self._testing_pycurl():
1547
if pycurl.version_info()[1] < '7.16.0':
1548
raise tests.KnownFailure(
1549
'pycurl < 7.16.0 does not handle empty proxy passwords')
1550
super(TestProxyAuth, self).test_empty_pass()
1553
class SampleSocket(object):
1554
"""A socket-like object for use in testing the HTTP request handler."""
1556
def __init__(self, socket_read_content):
1557
"""Constructs a sample socket.
1559
:param socket_read_content: a byte sequence
1561
# Use plain python StringIO so we can monkey-patch the close method to
1562
# not discard the contents.
1563
from StringIO import StringIO
1564
self.readfile = StringIO(socket_read_content)
1565
self.writefile = StringIO()
1566
self.writefile.close = lambda: None
1568
def makefile(self, mode='r', bufsize=None):
1570
return self.readfile
1572
return self.writefile
1575
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1578
super(SmartHTTPTunnellingTest, self).setUp()
1579
# We use the VFS layer as part of HTTP tunnelling tests.
1580
self._captureVar('BZR_NO_SMART_VFS', None)
1581
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1583
def create_transport_readonly_server(self):
1584
return http_utils.HTTPServerWithSmarts(
1585
protocol_version=self._protocol_version)
1587
def test_bulk_data(self):
1588
# We should be able to send and receive bulk data in a single message.
1589
# The 'readv' command in the smart protocol both sends and receives
1590
# bulk data, so we use that.
1591
self.build_tree(['data-file'])
1592
http_server = self.get_readonly_server()
1593
http_transport = self._transport(http_server.get_url())
1594
medium = http_transport.get_smart_medium()
1595
# Since we provide the medium, the url below will be mostly ignored
1596
# during the test, as long as the path is '/'.
1597
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1600
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1602
def test_http_send_smart_request(self):
1604
post_body = 'hello\n'
1605
expected_reply_body = 'ok\x012\n'
1607
http_server = self.get_readonly_server()
1608
http_transport = self._transport(http_server.get_url())
1609
medium = http_transport.get_smart_medium()
1610
response = medium.send_http_smart_request(post_body)
1611
reply_body = response.read()
1612
self.assertEqual(expected_reply_body, reply_body)
1614
def test_smart_http_server_post_request_handler(self):
1615
httpd = self.get_readonly_server()._get_httpd()
1617
socket = SampleSocket(
1618
'POST /.bzr/smart %s \r\n' % self._protocol_version
1619
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1621
+ 'Content-Length: 6\r\n'
1624
# Beware: the ('localhost', 80) below is the
1625
# client_address parameter, but we don't have one because
1626
# we have defined a socket which is not bound to an
1627
# address. The test framework never uses this client
1628
# address, so far...
1629
request_handler = http_utils.SmartRequestHandler(socket,
1632
response = socket.writefile.getvalue()
1633
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1634
# This includes the end of the HTTP headers, and all the body.
1635
expected_end_of_response = '\r\n\r\nok\x012\n'
1636
self.assertEndsWith(response, expected_end_of_response)
added
removed
bzrlib/tests/test_http.py
