20
20
transport implementation, http protocol versions and authentication schemes.
23
# TODO: Should be renamed to breezy.transport.http.tests?
24
# TODO: What about renaming to breezy.tests.transport.http ?
23
# TODO: Should be renamed to bzrlib.transport.http.tests?
24
# TODO: What about renaming to bzrlib.tests.transport.http ?
27
from http.client import UnknownProtocol, parse_headers
28
from http.server import SimpleHTTPRequestHandler
29
except ImportError: # python < 3
30
from httplib import UnknownProtocol
31
from SimpleHTTPServer import SimpleHTTPRequestHandler
26
from cStringIO import StringIO
30
import SimpleHTTPServer
41
remote as _mod_remote,
51
remote as _mod_remote,
47
from bzrlib.symbol_versioning import (
53
from ..sixish import PY3
50
from bzrlib.tests import (
60
from .scenarios import (
61
load_tests_apply_scenarios,
64
from ..transport import (
55
from bzrlib.transport import (
68
from ..transport.http import (
59
from bzrlib.transport.http import (
73
load_tests = load_tests_apply_scenarios
76
def vary_by_http_client_implementation():
77
"""Test the libraries we can use, currently just urllib."""
65
if features.pycurl.available():
66
from bzrlib.transport.http._pycurl import PyCurlTransport
69
def load_tests(standard_tests, module, loader):
70
"""Multiply tests for http clients and protocol versions."""
71
result = loader.suiteClass()
73
# one for each transport implementation
74
t_tests, remaining_tests = tests.split_suite_by_condition(
75
standard_tests, tests.condition_isinstance((
76
TestHttpTransportRegistration,
77
TestHttpTransportUrls,
78
80
transport_scenarios = [
79
('urllib', dict(_transport=HttpTransport,
80
_server=http_server.HttpServer,
81
_url_protocol='http',)),
83
return transport_scenarios
86
def vary_by_http_protocol_version():
87
"""Test on http/1.0 and 1.1"""
89
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
90
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
94
def vary_by_http_auth_scheme():
81
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
82
_server=http_server.HttpServer_urllib,
83
_qualified_prefix='http+urllib',)),
85
if features.pycurl.available():
86
transport_scenarios.append(
87
('pycurl', dict(_transport=PyCurlTransport,
88
_server=http_server.HttpServer_PyCurl,
89
_qualified_prefix='http+pycurl',)))
90
tests.multiply_tests(t_tests, transport_scenarios, result)
92
protocol_scenarios = [
93
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
94
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
97
# some tests are parametrized by the protocol version only
98
p_tests, remaining_tests = tests.split_suite_by_condition(
99
remaining_tests, tests.condition_isinstance((
100
TestAuthOnRedirected,
102
tests.multiply_tests(p_tests, protocol_scenarios, result)
104
# each implementation tested with each HTTP version
105
tp_tests, remaining_tests = tests.split_suite_by_condition(
106
remaining_tests, tests.condition_isinstance((
107
SmartHTTPTunnellingTest,
108
TestDoCatchRedirections,
110
TestHTTPRedirections,
111
TestHTTPSilentRedirections,
112
TestLimitedRangeRequestServer,
116
TestSpecificRequestHandler,
118
tp_scenarios = tests.multiply_scenarios(transport_scenarios,
120
tests.multiply_tests(tp_tests, tp_scenarios, result)
122
# proxy auth: each auth scheme on all http versions on all implementations.
123
tppa_tests, remaining_tests = tests.split_suite_by_condition(
124
remaining_tests, tests.condition_isinstance((
127
proxy_auth_scheme_scenarios = [
128
('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
129
('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
131
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
133
tppa_scenarios = tests.multiply_scenarios(tp_scenarios,
134
proxy_auth_scheme_scenarios)
135
tests.multiply_tests(tppa_tests, tppa_scenarios, result)
137
# auth: each auth scheme on all http versions on all implementations.
138
tpa_tests, remaining_tests = tests.split_suite_by_condition(
139
remaining_tests, tests.condition_isinstance((
142
auth_scheme_scenarios = [
96
143
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
97
144
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
99
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
101
# Add some attributes common to all scenarios
102
for scenario_id, scenario_dict in scenarios:
103
scenario_dict.update(_auth_header='Authorization',
104
_username_prompt_prefix='',
105
_password_prompt_prefix='')
109
def vary_by_http_proxy_auth_scheme():
111
('proxy-basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
112
('proxy-digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
113
('proxy-basicdigest',
114
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
116
# Add some attributes common to all scenarios
117
for scenario_id, scenario_dict in scenarios:
118
scenario_dict.update(_auth_header='Proxy-Authorization',
119
_username_prompt_prefix='Proxy ',
120
_password_prompt_prefix='Proxy ')
124
def vary_by_http_activity():
146
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
148
tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
149
auth_scheme_scenarios)
150
tests.multiply_tests(tpa_tests, tpa_scenarios, result)
152
# activity: on all http[s] versions on all implementations
153
tpact_tests, remaining_tests = tests.split_suite_by_condition(
154
remaining_tests, tests.condition_isinstance((
125
157
activity_scenarios = [
126
158
('urllib,http', dict(_activity_server=ActivityHTTPServer,
127
_transport=HttpTransport,)),
159
_transport=_urllib.HttpTransport_urllib,)),
129
if features.HTTPSServerFeature.available():
130
# FIXME: Until we have a better way to handle self-signed certificates
131
# (like allowing them in a test specific authentication.conf for
132
# example), we need some specialized urllib transport for tests.
138
class HTTPS_transport(HttpTransport):
140
def __init__(self, base, _from_transport=None):
141
super(HTTPS_transport, self).__init__(
142
base, _from_transport=_from_transport,
143
ca_certs=ssl_certs.build_path('ca.crt'))
161
if tests.HTTPSServerFeature.available():
145
162
activity_scenarios.append(
146
163
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
147
_transport=HTTPS_transport,)),)
148
return activity_scenarios
164
_transport=_urllib.HttpTransport_urllib,)),)
165
if features.pycurl.available():
166
activity_scenarios.append(
167
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
168
_transport=PyCurlTransport,)),)
169
if tests.HTTPSServerFeature.available():
170
from bzrlib.tests import (
173
# FIXME: Until we have a better way to handle self-signed
174
# certificates (like allowing them in a test specific
175
# authentication.conf for example), we need some specialized pycurl
176
# transport for tests.
177
class HTTPS_pycurl_transport(PyCurlTransport):
179
def __init__(self, base, _from_transport=None):
180
super(HTTPS_pycurl_transport, self).__init__(
181
base, _from_transport)
182
self.cabundle = str(ssl_certs.build_path('ca.crt'))
184
activity_scenarios.append(
185
('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
186
_transport=HTTPS_pycurl_transport,)),)
188
tpact_scenarios = tests.multiply_scenarios(activity_scenarios,
190
tests.multiply_tests(tpact_tests, tpact_scenarios, result)
192
# No parametrization for the remaining tests
193
result.addTests(remaining_tests)
151
198
class FakeManager(object):
368
374
server = http_server.HttpServer(RequestHandlerOneOne,
369
375
protocol_version='HTTP/1.0')
370
376
self.start_server(server)
371
self.assertIsInstance(server.server,
377
self.assertIsInstance(server._httpd,
372
378
http_server.TestingHTTPServer)
381
class TestWithTransport_pycurl(object):
382
"""Test case to inherit from if pycurl is present"""
384
def _get_pycurl_maybe(self):
385
self.requireFeature(features.pycurl)
386
return PyCurlTransport
388
_transport = property(_get_pycurl_maybe)
391
class TestHttpUrls(tests.TestCase):
393
# TODO: This should be moved to authorization tests once they
396
def test_url_parsing(self):
398
url = http.extract_auth('http://example.com', f)
399
self.assertEqual('http://example.com', url)
400
self.assertEqual(0, len(f.credentials))
401
url = http.extract_auth(
402
'http://user:pass@example.com/bzr/bzr.dev', f)
403
self.assertEqual('http://example.com/bzr/bzr.dev', url)
404
self.assertEqual(1, len(f.credentials))
405
self.assertEqual([None, 'example.com', 'user', 'pass'],
375
409
class TestHttpTransportUrls(tests.TestCase):
376
410
"""Test the http urls."""
378
scenarios = vary_by_http_client_implementation()
380
412
def test_abs_url(self):
381
413
"""Construction of absolute http URLs"""
382
t = self._transport('http://example.com/bzr/bzr.dev/')
414
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
383
415
eq = self.assertEqualDiff
384
eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
385
eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
386
eq(t.abspath('.bzr'), 'http://example.com/bzr/bzr.dev/.bzr')
416
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
417
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
418
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
387
419
eq(t.abspath('.bzr/1//2/./3'),
388
'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
420
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
390
422
def test_invalid_http_urls(self):
391
423
"""Trap invalid construction of urls"""
392
self._transport('http://example.com/bzr/bzr.dev/')
393
self.assertRaises(urlutils.InvalidURL,
424
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
425
self.assertRaises(errors.InvalidURL,
395
'http://example.com:port/bzr/bzr.dev/')
427
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
397
429
def test_http_root_urls(self):
398
430
"""Construction of URLs from server root"""
399
t = self._transport('http://example.com/')
431
t = self._transport('http://bzr.ozlabs.org/')
400
432
eq = self.assertEqualDiff
401
433
eq(t.abspath('.bzr/tree-version'),
402
'http://example.com/.bzr/tree-version')
434
'http://bzr.ozlabs.org/.bzr/tree-version')
404
436
def test_http_impl_urls(self):
405
437
"""There are servers which ask for particular clients to connect"""
407
439
server.start_server()
409
441
url = server.get_url()
410
self.assertTrue(url.startswith('%s://' % self._url_protocol))
442
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
412
444
server.stop_server()
447
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
449
# TODO: This should really be moved into another pycurl
450
# specific test. When https tests will be implemented, take
451
# this one into account.
452
def test_pycurl_without_https_support(self):
453
"""Test that pycurl without SSL do not fail with a traceback.
455
For the purpose of the test, we force pycurl to ignore
456
https by supplying a fake version_info that do not
459
self.requireFeature(features.pycurl)
460
# Import the module locally now that we now it's available.
461
pycurl = features.pycurl.module
463
self.overrideAttr(pycurl, 'version_info',
464
# Fake the pycurl version_info This was taken from
465
# a windows pycurl without SSL (thanks to bialix)
474
('ftp', 'gopher', 'telnet',
475
'dict', 'ldap', 'http', 'file'),
479
self.assertRaises(errors.DependencyNotPresent, self._transport,
480
'https://launchpad.net')
415
483
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
416
484
"""Test the http connections."""
418
scenarios = multiply_scenarios(
419
vary_by_http_client_implementation(),
420
vary_by_http_protocol_version(),
424
super(TestHTTPConnections, self).setUp()
487
http_utils.TestCaseWithWebserver.setUp(self)
425
488
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
426
489
transport=self.get_transport())
428
491
def test_http_has(self):
429
492
server = self.get_readonly_server()
430
t = self.get_readonly_transport()
493
t = self._transport(server.get_url())
431
494
self.assertEqual(t.has('foo/bar'), True)
432
495
self.assertEqual(len(server.logs), 1)
433
496
self.assertContainsRe(server.logs[0],
434
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "Breezy/')
497
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
436
499
def test_http_has_not_found(self):
437
500
server = self.get_readonly_server()
438
t = self.get_readonly_transport()
501
t = self._transport(server.get_url())
439
502
self.assertEqual(t.has('not-found'), False)
440
503
self.assertContainsRe(server.logs[1],
441
r'"HEAD /not-found HTTP/1.." 404 - "-" "Breezy/')
504
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
443
506
def test_http_get(self):
444
507
server = self.get_readonly_server()
445
t = self.get_readonly_transport()
508
t = self._transport(server.get_url())
446
509
fp = t.get('foo/bar')
447
510
self.assertEqualDiff(
449
b'contents of foo/bar\n')
512
'contents of foo/bar\n')
450
513
self.assertEqual(len(server.logs), 1)
451
514
self.assertTrue(server.logs[0].find(
452
'"GET /foo/bar HTTP/1.1" 200 - "-" "Breezy/%s'
453
% breezy.__version__) > -1)
515
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
516
% bzrlib.__version__) > -1)
455
518
def test_has_on_bogus_host(self):
456
519
# Get a free address and don't 'accept' on it, so that we
470
533
class TestHttpTransportRegistration(tests.TestCase):
471
534
"""Test registrations of various http implementations"""
473
scenarios = vary_by_http_client_implementation()
475
536
def test_http_registered(self):
476
t = transport.get_transport_from_url(
477
'%s://foo.com/' % self._url_protocol)
537
t = transport.get_transport('%s://foo.com/' % self._qualified_prefix)
478
538
self.assertIsInstance(t, transport.Transport)
479
539
self.assertIsInstance(t, self._transport)
482
542
class TestPost(tests.TestCase):
484
scenarios = multiply_scenarios(
485
vary_by_http_client_implementation(),
486
vary_by_http_protocol_version(),
489
544
def test_post_body_is_received(self):
490
server = RecordingServer(expect_body_tail=b'end-of-body',
491
scheme=self._url_protocol)
545
server = RecordingServer(expect_body_tail='end-of-body',
546
scheme=self._qualified_prefix)
492
547
self.start_server(server)
493
548
url = server.get_url()
494
# FIXME: needs a cleanup -- vila 20100611
495
http_transport = transport.get_transport_from_url(url)
496
code, response = http_transport._post(b'abc def end-of-body')
498
server.received_bytes.startswith(b'POST /.bzr/smart HTTP/1.'))
500
b'content-length: 19\r' in server.received_bytes.lower())
501
self.assertTrue(b'content-type: application/octet-stream\r'
502
in server.received_bytes.lower())
549
http_transport = self._transport(url)
550
code, response = http_transport._post('abc def end-of-body')
552
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
553
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
503
554
# The transport should not be assuming that the server can accept
504
555
# chunked encoding the first time it connects, because HTTP/1.1, so we
505
556
# check for the literal string.
507
server.received_bytes.endswith(b'\r\n\r\nabc def end-of-body'))
558
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
510
561
class TestRangeHeader(tests.TestCase):
511
562
"""Test range_header method"""
513
564
def check_header(self, value, ranges=[], tail=0):
514
offsets = [(start, end - start + 1) for start, end in ranges]
565
offsets = [ (start, end - start + 1) for start, end in ranges]
515
566
coalesce = transport.Transport._coalesce_offsets
516
567
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
517
range_header = http.HttpTransport._range_header
568
range_header = http.HttpTransportBase._range_header
518
569
self.assertEqual(value, range_header(coalesced, tail))
520
571
def test_range_header_single(self):
521
self.check_header('0-9', ranges=[(0, 9)])
522
self.check_header('100-109', ranges=[(100, 109)])
572
self.check_header('0-9', ranges=[(0,9)])
573
self.check_header('100-109', ranges=[(100,109)])
524
575
def test_range_header_tail(self):
525
576
self.check_header('-10', tail=10)
741
799
super(TestRangeRequestServer, self).setUp()
742
self.build_tree_contents([('a', b'0123456789')],)
800
self.build_tree_contents([('a', '0123456789')],)
744
802
def test_readv(self):
745
t = self.get_readonly_transport()
803
server = self.get_readonly_server()
804
t = self._transport(server.get_url())
746
805
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
747
self.assertEqual(l[0], (0, b'0'))
748
self.assertEqual(l[1], (1, b'1'))
749
self.assertEqual(l[2], (3, b'34'))
750
self.assertEqual(l[3], (9, b'9'))
806
self.assertEqual(l[0], (0, '0'))
807
self.assertEqual(l[1], (1, '1'))
808
self.assertEqual(l[2], (3, '34'))
809
self.assertEqual(l[3], (9, '9'))
752
811
def test_readv_out_of_order(self):
753
t = self.get_readonly_transport()
812
server = self.get_readonly_server()
813
t = self._transport(server.get_url())
754
814
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
755
self.assertEqual(l[0], (1, b'1'))
756
self.assertEqual(l[1], (9, b'9'))
757
self.assertEqual(l[2], (0, b'0'))
758
self.assertEqual(l[3], (3, b'34'))
815
self.assertEqual(l[0], (1, '1'))
816
self.assertEqual(l[1], (9, '9'))
817
self.assertEqual(l[2], (0, '0'))
818
self.assertEqual(l[3], (3, '34'))
760
820
def test_readv_invalid_ranges(self):
761
t = self.get_readonly_transport()
821
server = self.get_readonly_server()
822
t = self._transport(server.get_url())
763
824
# This is intentionally reading off the end of the file
764
825
# since we are sure that it cannot get there
765
826
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
766
t.readv, 'a', [(1, 1), (8, 10)])
827
t.readv, 'a', [(1,1), (8,10)])
768
829
# This is trying to seek past the end of the file, it should
769
830
# also raise a special error
770
831
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
771
t.readv, 'a', [(12, 2)])
832
t.readv, 'a', [(12,2)])
773
834
def test_readv_multiple_get_requests(self):
774
835
server = self.get_readonly_server()
775
t = self.get_readonly_transport()
836
t = self._transport(server.get_url())
776
837
# force transport to issue multiple requests
777
838
t._max_readv_combine = 1
778
839
t._max_get_ranges = 1
779
840
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
780
self.assertEqual(l[0], (0, b'0'))
781
self.assertEqual(l[1], (1, b'1'))
782
self.assertEqual(l[2], (3, b'34'))
783
self.assertEqual(l[3], (9, b'9'))
841
self.assertEqual(l[0], (0, '0'))
842
self.assertEqual(l[1], (1, '1'))
843
self.assertEqual(l[2], (3, '34'))
844
self.assertEqual(l[3], (9, '9'))
784
845
# The server should have issued 4 requests
785
846
self.assertEqual(4, server.GET_request_nb)
787
848
def test_readv_get_max_size(self):
788
849
server = self.get_readonly_server()
789
t = self.get_readonly_transport()
850
t = self._transport(server.get_url())
790
851
# force transport to issue multiple requests by limiting the number of
791
852
# bytes by request. Note that this apply to coalesced offsets only, a
792
853
# single range will keep its size even if bigger than the limit.
793
854
t._get_max_size = 2
794
855
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
795
self.assertEqual(l[0], (0, b'0'))
796
self.assertEqual(l[1], (1, b'1'))
797
self.assertEqual(l[2], (2, b'2345'))
798
self.assertEqual(l[3], (6, b'6789'))
856
self.assertEqual(l[0], (0, '0'))
857
self.assertEqual(l[1], (1, '1'))
858
self.assertEqual(l[2], (2, '2345'))
859
self.assertEqual(l[3], (6, '6789'))
799
860
# The server should have issued 3 requests
800
861
self.assertEqual(3, server.GET_request_nb)
802
863
def test_complete_readv_leave_pipe_clean(self):
803
864
server = self.get_readonly_server()
804
t = self.get_readonly_transport()
865
t = self._transport(server.get_url())
805
866
# force transport to issue multiple requests
806
867
t._get_max_size = 2
807
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
868
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
808
869
# The server should have issued 3 requests
809
870
self.assertEqual(3, server.GET_request_nb)
810
self.assertEqual(b'0123456789', t.get_bytes('a'))
871
self.assertEqual('0123456789', t.get_bytes('a'))
811
872
self.assertEqual(4, server.GET_request_nb)
813
874
def test_incomplete_readv_leave_pipe_clean(self):
814
875
server = self.get_readonly_server()
815
t = self.get_readonly_transport()
876
t = self._transport(server.get_url())
816
877
# force transport to issue multiple requests
817
878
t._get_max_size = 2
818
879
# Don't collapse readv results into a list so that we leave unread
819
880
# bytes on the socket
820
881
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
821
self.assertEqual((0, b'0'), next(ireadv))
882
self.assertEqual((0, '0'), ireadv.next())
822
883
# The server should have issued one request so far
823
884
self.assertEqual(1, server.GET_request_nb)
824
self.assertEqual(b'0123456789', t.get_bytes('a'))
885
self.assertEqual('0123456789', t.get_bytes('a'))
825
886
# get_bytes issued an additional request, the readv pending ones are
827
888
self.assertEqual(2, server.GET_request_nb)
967
1026
def setUp(self):
968
1027
super(TestTruncatedMultipleRangeServer, self).setUp()
969
self.build_tree_contents([('a', b'0123456789')],)
1028
self.build_tree_contents([('a', '0123456789')],)
971
1030
def test_readv_with_short_reads(self):
972
1031
server = self.get_readonly_server()
973
t = self.get_readonly_transport()
1032
t = self._transport(server.get_url())
974
1033
# Force separate ranges for each offset
975
1034
t._bytes_to_read_before_seek = 0
976
1035
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
977
self.assertEqual((0, b'0'), next(ireadv))
978
self.assertEqual((2, b'2'), next(ireadv))
979
# Only one request have been issued so far
980
self.assertEqual(1, server.GET_request_nb)
981
self.assertEqual((4, b'45'), next(ireadv))
982
self.assertEqual((9, b'9'), next(ireadv))
983
# We issue 3 requests: two multiple (4 ranges, then 2 ranges) then a
1036
self.assertEqual((0, '0'), ireadv.next())
1037
self.assertEqual((2, '2'), ireadv.next())
1038
if not self._testing_pycurl():
1039
# Only one request have been issued so far (except for pycurl that
1040
# try to read the whole response at once)
1041
self.assertEqual(1, server.GET_request_nb)
1042
self.assertEqual((4, '45'), ireadv.next())
1043
self.assertEqual((9, '9'), ireadv.next())
1044
# Both implementations issue 3 requests but:
1045
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1047
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
985
1048
self.assertEqual(3, server.GET_request_nb)
986
1049
# Finally the client have tried a single range request and stays in
988
1051
self.assertEqual('single', t._range_hint)
991
class TruncatedBeforeBoundaryRequestHandler(
992
http_server.TestingHTTPRequestHandler):
993
"""Truncation before a boundary, like in bug 198646"""
995
_truncated_ranges = 1
997
def get_multiple_ranges(self, file, file_size, ranges):
998
self.send_response(206)
999
self.send_header('Accept-Ranges', 'bytes')
1001
self.send_header('Content-Type',
1002
'multipart/byteranges; boundary=%s' % boundary)
1003
boundary_line = b'--%s\r\n' % boundary.encode('ascii')
1004
# Calculate the Content-Length
1006
for (start, end) in ranges:
1007
content_length += len(boundary_line)
1008
content_length += self._header_line_length(
1009
'Content-type', 'application/octet-stream')
1010
content_length += self._header_line_length(
1011
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
1012
content_length += len('\r\n') # end headers
1013
content_length += end - start # + 1
1014
content_length += len(boundary_line)
1015
self.send_header('Content-length', content_length)
1018
# Send the multipart body
1020
for (start, end) in ranges:
1021
if cur + self._truncated_ranges >= len(ranges):
1022
# Abruptly ends the response and close the connection
1023
self.close_connection = 1
1025
self.wfile.write(boundary_line)
1026
self.send_header('Content-type', 'application/octet-stream')
1027
self.send_header('Content-Range', 'bytes %d-%d/%d'
1028
% (start, end, file_size))
1030
self.send_range_content(file, start, end - start + 1)
1033
self.wfile.write(boundary_line)
1036
class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):
1037
"""Tests the case of bug 198646, disconnecting before a boundary."""
1039
_req_handler_class = TruncatedBeforeBoundaryRequestHandler
1042
super(TestTruncatedBeforeBoundary, self).setUp()
1043
self.build_tree_contents([('a', b'0123456789')],)
1045
def test_readv_with_short_reads(self):
1046
server = self.get_readonly_server()
1047
t = self.get_readonly_transport()
1048
# Force separate ranges for each offset
1049
t._bytes_to_read_before_seek = 0
1050
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1051
self.assertEqual((0, b'0'), next(ireadv))
1052
self.assertEqual((2, b'2'), next(ireadv))
1053
self.assertEqual((4, b'45'), next(ireadv))
1054
self.assertEqual((9, b'9'), next(ireadv))
1057
1053
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1058
1054
"""Errors out when range specifiers exceed the limit"""
1095
1086
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1096
1087
protocol_version=self._protocol_version)
1089
def get_transport(self):
1090
return self._transport(self.get_readonly_server().get_url())
1098
1092
def setUp(self):
1099
super(TestLimitedRangeRequestServer, self).setUp()
1093
http_utils.TestCaseWithWebserver.setUp(self)
1100
1094
# We need to manipulate ranges that correspond to real chunks in the
1101
1095
# response, so we build a content appropriately.
1102
filler = b''.join([b'abcdefghij' for x in range(102)])
1103
content = b''.join([b'%04d' % v + filler for v in range(16)])
1096
filler = ''.join(['abcdefghij' for x in range(102)])
1097
content = ''.join(['%04d' % v + filler for v in range(16)])
1104
1098
self.build_tree_contents([('a', content)],)
1106
1100
def test_few_ranges(self):
1107
t = self.get_readonly_transport()
1101
t = self.get_transport()
1108
1102
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1109
self.assertEqual(l[0], (0, b'0000'))
1110
self.assertEqual(l[1], (1024, b'0001'))
1103
self.assertEqual(l[0], (0, '0000'))
1104
self.assertEqual(l[1], (1024, '0001'))
1111
1105
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1113
1107
def test_more_ranges(self):
1114
t = self.get_readonly_transport()
1108
t = self.get_transport()
1115
1109
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1116
self.assertEqual(l[0], (0, b'0000'))
1117
self.assertEqual(l[1], (1024, b'0001'))
1118
self.assertEqual(l[2], (4096, b'0004'))
1119
self.assertEqual(l[3], (8192, b'0008'))
1110
self.assertEqual(l[0], (0, '0000'))
1111
self.assertEqual(l[1], (1024, '0001'))
1112
self.assertEqual(l[2], (4096, '0004'))
1113
self.assertEqual(l[3], (8192, '0008'))
1120
1114
# The server will refuse to serve the first request (too much ranges),
1121
1115
# a second request will succeed.
1122
1116
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1128
1122
Only the urllib implementation is tested here.
1126
tests.TestCase.setUp(self)
1131
tests.TestCase.tearDown(self)
1133
def _install_env(self, env):
1134
for name, value in env.iteritems():
1135
self._old_env[name] = osutils.set_or_unset_env(name, value)
1137
def _restore_env(self):
1138
for name, value in self._old_env.iteritems():
1139
osutils.set_or_unset_env(name, value)
1131
1141
def _proxied_request(self):
1132
handler = http.ProxyHandler()
1133
request = http.Request('GET', 'http://baz/buzzle')
1142
handler = _urllib2_wrappers.ProxyHandler()
1143
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
1134
1144
handler.set_proxy(request, 'http')
1137
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1138
handler = http.ProxyHandler()
1139
self.assertEqual(expected,
1140
handler.evaluate_proxy_bypass(host, no_proxy))
1142
1147
def test_empty_user(self):
1143
self.overrideEnv('http_proxy', 'http://bar.com')
1144
request = self._proxied_request()
1145
self.assertFalse('Proxy-authorization' in request.headers)
1147
def test_user_with_at(self):
1148
self.overrideEnv('http_proxy',
1149
'http://username@domain:password@proxy_host:1234')
1150
request = self._proxied_request()
1151
self.assertFalse('Proxy-authorization' in request.headers)
1148
self._install_env({'http_proxy': 'http://bar.com'})
1149
request = self._proxied_request()
1150
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1153
1152
def test_invalid_proxy(self):
1154
1153
"""A proxy env variable without scheme"""
1155
self.overrideEnv('http_proxy', 'host:1234')
1156
self.assertRaises(urlutils.InvalidURL, self._proxied_request)
1158
def test_evaluate_proxy_bypass_true(self):
1159
"""The host is not proxied"""
1160
self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
1161
self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
1163
def test_evaluate_proxy_bypass_false(self):
1164
"""The host is proxied"""
1165
self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
1167
def test_evaluate_proxy_bypass_unknown(self):
1168
"""The host is not explicitly proxied"""
1169
self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
1170
self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
1172
def test_evaluate_proxy_bypass_empty_entries(self):
1173
"""Ignore empty entries"""
1174
self.assertEvaluateProxyBypass(None, 'example.com', '')
1175
self.assertEvaluateProxyBypass(None, 'example.com', ',')
1176
self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
1154
self._install_env({'http_proxy': 'host:1234'})
1155
self.assertRaises(errors.InvalidURL, self._proxied_request)
1179
1158
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1185
1164
to the file names).
1188
scenarios = multiply_scenarios(
1189
vary_by_http_client_implementation(),
1190
vary_by_http_protocol_version(),
1193
1167
# FIXME: We don't have an https server available, so we don't
1194
# test https connections. --vila toolongago
1168
# test https connections.
1196
1170
def setUp(self):
1197
1171
super(TestProxyHttpServer, self).setUp()
1198
self.transport_secondary_server = http_utils.ProxyServer
1199
self.build_tree_contents([('foo', b'contents of foo\n'),
1200
('foo-proxied', b'proxied contents of foo\n')])
1172
self.build_tree_contents([('foo', 'contents of foo\n'),
1173
('foo-proxied', 'proxied contents of foo\n')])
1201
1174
# Let's setup some attributes for tests
1202
server = self.get_readonly_server()
1203
self.server_host_port = '%s:%d' % (server.host, server.port)
1204
self.no_proxy_host = self.server_host_port
1175
self.server = self.get_readonly_server()
1176
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
1177
if self._testing_pycurl():
1178
# Oh my ! pycurl does not check for the port as part of
1179
# no_proxy :-( So we just test the host part
1180
self.no_proxy_host = self.server.host
1182
self.no_proxy_host = self.proxy_address
1205
1183
# The secondary server is the proxy
1206
self.proxy_url = self.get_secondary_url()
1208
def assertProxied(self):
1209
t = self.get_readonly_transport()
1210
self.assertEqual(b'proxied contents of foo\n', t.get('foo').read())
1212
def assertNotProxied(self):
1213
t = self.get_readonly_transport()
1214
self.assertEqual(b'contents of foo\n', t.get('foo').read())
1184
self.proxy = self.get_secondary_server()
1185
self.proxy_url = self.proxy.get_url()
1188
def _testing_pycurl(self):
1189
# TODO: This is duplicated for lots of the classes in this file
1190
return (features.pycurl.available()
1191
and self._transport == PyCurlTransport)
1193
def create_transport_secondary_server(self):
1194
"""Creates an http server that will serve files with
1195
'-proxied' appended to their names.
1197
return http_utils.ProxyServer(protocol_version=self._protocol_version)
1199
def _install_env(self, env):
1200
for name, value in env.iteritems():
1201
self._old_env[name] = osutils.set_or_unset_env(name, value)
1203
def _restore_env(self):
1204
for name, value in self._old_env.iteritems():
1205
osutils.set_or_unset_env(name, value)
1207
def proxied_in_env(self, env):
1208
self._install_env(env)
1209
url = self.server.get_url()
1210
t = self._transport(url)
1212
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1216
def not_proxied_in_env(self, env):
1217
self._install_env(env)
1218
url = self.server.get_url()
1219
t = self._transport(url)
1221
self.assertEqual('contents of foo\n', t.get('foo').read())
1216
1225
def test_http_proxy(self):
1217
self.overrideEnv('http_proxy', self.proxy_url)
1218
self.assertProxied()
1226
self.proxied_in_env({'http_proxy': self.proxy_url})
1220
1228
def test_HTTP_PROXY(self):
1221
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1222
self.assertProxied()
1229
if self._testing_pycurl():
1230
# pycurl does not check HTTP_PROXY for security reasons
1231
# (for use in a CGI context that we do not care
1232
# about. Should we ?)
1233
raise tests.TestNotApplicable(
1234
'pycurl does not check HTTP_PROXY for security reasons')
1235
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
1224
1237
def test_all_proxy(self):
1225
self.overrideEnv('all_proxy', self.proxy_url)
1226
self.assertProxied()
1238
self.proxied_in_env({'all_proxy': self.proxy_url})
1228
1240
def test_ALL_PROXY(self):
1229
self.overrideEnv('ALL_PROXY', self.proxy_url)
1230
self.assertProxied()
1241
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1232
1243
def test_http_proxy_with_no_proxy(self):
1233
self.overrideEnv('no_proxy', self.no_proxy_host)
1234
self.overrideEnv('http_proxy', self.proxy_url)
1235
self.assertNotProxied()
1244
self.not_proxied_in_env({'http_proxy': self.proxy_url,
1245
'no_proxy': self.no_proxy_host})
1237
1247
def test_HTTP_PROXY_with_NO_PROXY(self):
1238
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1239
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1240
self.assertNotProxied()
1248
if self._testing_pycurl():
1249
raise tests.TestNotApplicable(
1250
'pycurl does not check HTTP_PROXY for security reasons')
1251
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
1252
'NO_PROXY': self.no_proxy_host})
1242
1254
def test_all_proxy_with_no_proxy(self):
1243
self.overrideEnv('no_proxy', self.no_proxy_host)
1244
self.overrideEnv('all_proxy', self.proxy_url)
1245
self.assertNotProxied()
1255
self.not_proxied_in_env({'all_proxy': self.proxy_url,
1256
'no_proxy': self.no_proxy_host})
1247
1258
def test_ALL_PROXY_with_NO_PROXY(self):
1248
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1249
self.overrideEnv('ALL_PROXY', self.proxy_url)
1250
self.assertNotProxied()
1259
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
1260
'NO_PROXY': self.no_proxy_host})
1252
1262
def test_http_proxy_without_scheme(self):
1253
self.overrideEnv('http_proxy', self.server_host_port)
1254
self.assertRaises(urlutils.InvalidURL, self.assertProxied)
1263
if self._testing_pycurl():
1264
# pycurl *ignores* invalid proxy env variables. If that ever change
1265
# in the future, this test will fail indicating that pycurl do not
1266
# ignore anymore such variables.
1267
self.not_proxied_in_env({'http_proxy': self.proxy_address})
1269
self.assertRaises(errors.InvalidURL,
1270
self.proxied_in_env,
1271
{'http_proxy': self.proxy_address})
1257
1274
class TestRanges(http_utils.TestCaseWithWebserver):
1258
1275
"""Test the Range header in GET methods."""
1260
scenarios = multiply_scenarios(
1261
vary_by_http_client_implementation(),
1262
vary_by_http_protocol_version(),
1265
1277
def setUp(self):
1266
super(TestRanges, self).setUp()
1267
self.build_tree_contents([('a', b'0123456789')],)
1278
http_utils.TestCaseWithWebserver.setUp(self)
1279
self.build_tree_contents([('a', '0123456789')],)
1280
server = self.get_readonly_server()
1281
self.transport = self._transport(server.get_url())
1269
1283
def create_transport_readonly_server(self):
1270
1284
return http_server.HttpServer(protocol_version=self._protocol_version)
1272
1286
def _file_contents(self, relpath, ranges):
1273
t = self.get_readonly_transport()
1274
offsets = [(start, end - start + 1) for start, end in ranges]
1275
coalesce = t._coalesce_offsets
1287
offsets = [ (start, end - start + 1) for start, end in ranges]
1288
coalesce = self.transport._coalesce_offsets
1276
1289
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1277
code, data = t._get(relpath, coalesced)
1278
self.assertTrue(code in (200, 206), '_get returns: %d' % code)
1290
code, data = self.transport._get(relpath, coalesced)
1291
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1279
1292
for start, end in ranges:
1280
1293
data.seek(start)
1281
1294
yield data.read(end - start + 1)
1283
1296
def _file_tail(self, relpath, tail_amount):
1284
t = self.get_readonly_transport()
1285
code, data = t._get(relpath, [], tail_amount)
1286
self.assertTrue(code in (200, 206), '_get returns: %d' % code)
1297
code, data = self.transport._get(relpath, [], tail_amount)
1298
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1287
1299
data.seek(-tail_amount, 2)
1288
1300
return data.read(tail_amount)
1290
1302
def test_range_header(self):
1293
[b'0', b'234'], list(self._file_contents('a', [(0, 0), (2, 4)])))
1304
map(self.assertEqual,['0', '234'],
1305
list(self._file_contents('a', [(0,0), (2,4)])),)
1295
1307
def test_range_header_tail(self):
1296
self.assertEqual(b'789', self._file_tail('a', 3))
1308
self.assertEqual('789', self._file_tail('a', 3))
1298
1310
def test_syntactically_invalid_range_header(self):
1299
1311
self.assertListRaises(errors.InvalidHttpRange,
1300
self._file_contents, 'a', [(4, 3)])
1312
self._file_contents, 'a', [(4, 3)])
1302
1314
def test_semantically_invalid_range_header(self):
1303
1315
self.assertListRaises(errors.InvalidHttpRange,
1304
self._file_contents, 'a', [(42, 128)])
1316
self._file_contents, 'a', [(42, 128)])
1307
1319
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1308
1320
"""Test redirection between http servers."""
1310
scenarios = multiply_scenarios(
1311
vary_by_http_client_implementation(),
1312
vary_by_http_protocol_version(),
1322
def create_transport_secondary_server(self):
1323
"""Create the secondary server redirecting to the primary server"""
1324
new = self.get_readonly_server()
1326
redirecting = http_utils.HTTPServerRedirecting(
1327
protocol_version=self._protocol_version)
1328
redirecting.redirect_to(new.host, new.port)
1315
1331
def setUp(self):
1316
1332
super(TestHTTPRedirections, self).setUp()
1317
self.build_tree_contents([('a', b'0123456789'),
1333
self.build_tree_contents([('a', '0123456789'),
1319
b'# Bazaar revision bundle v0.9\n#\n')
1335
'# Bazaar revision bundle v0.9\n#\n')
1337
# The requests to the old server will be redirected to the new server
1338
self.old_transport = self._transport(self.old_server.get_url())
1322
1340
def test_redirected(self):
1323
self.assertRaises(errors.RedirectRequested,
1324
self.get_old_transport().get, 'a')
1327
self.get_new_transport().get('a').read())
1330
class RedirectedRequest(http.Request):
1341
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1342
t = self._transport(self.new_server.get_url())
1343
self.assertEqual('0123456789', t.get('a').read())
1345
def test_read_redirected_bundle_from_url(self):
1346
from bzrlib.bundle import read_bundle_from_url
1347
url = self.old_transport.abspath('bundle')
1348
bundle = self.applyDeprecated(deprecated_in((1, 12, 0)),
1349
read_bundle_from_url, url)
1350
# If read_bundle_from_url was successful we get an empty bundle
1351
self.assertEqual([], bundle.revisions)
1354
class RedirectedRequest(_urllib2_wrappers.Request):
1331
1355
"""Request following redirections. """
1333
init_orig = http.Request.__init__
1357
init_orig = _urllib2_wrappers.Request.__init__
1335
1359
def __init__(self, method, url, *args, **kwargs):
1336
1360
"""Constructor.
1339
1363
# Since the tests using this class will replace
1340
# http.Request, we can't just call the base class __init__
1364
# _urllib2_wrappers.Request, we can't just call the base class __init__
1341
1365
# or we'll loop.
1342
1366
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1343
1367
self.follow_redirections = True
1346
1370
def install_redirected_request(test):
1347
test.overrideAttr(http, 'Request', RedirectedRequest)
1350
def cleanup_http_redirection_connections(test):
1351
# Some sockets are opened but never seen by _urllib, so we trap them at
1352
# the http level to be able to clean them up.
1353
def socket_disconnect(sock):
1355
sock.shutdown(socket.SHUT_RDWR)
1357
except socket.error:
1360
def connect(connection):
1361
test.http_connect_orig(connection)
1362
test.addCleanup(socket_disconnect, connection.sock)
1363
test.http_connect_orig = test.overrideAttr(
1364
http.HTTPConnection, 'connect', connect)
1366
def connect(connection):
1367
test.https_connect_orig(connection)
1368
test.addCleanup(socket_disconnect, connection.sock)
1369
test.https_connect_orig = test.overrideAttr(
1370
http.HTTPSConnection, 'connect', connect)
1371
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1373
1374
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1376
1377
http implementations do not redirect silently anymore (they
1377
1378
do not redirect at all in fact). The mechanism is still in
1378
place at the http.Request level and these tests
1379
place at the _urllib2_wrappers.Request level and these tests
1382
For the pycurl implementation
1383
the redirection have been deleted as we may deprecate pycurl
1384
and I have no place to keep a working implementation.
1382
scenarios = multiply_scenarios(
1383
vary_by_http_client_implementation(),
1384
vary_by_http_protocol_version(),
1387
1388
def setUp(self):
1389
if (features.pycurl.available()
1390
and self._transport == PyCurlTransport):
1391
raise tests.TestNotApplicable(
1392
"pycurl doesn't redirect silently annymore")
1388
1393
super(TestHTTPSilentRedirections, self).setUp()
1389
1394
install_redirected_request(self)
1390
cleanup_http_redirection_connections(self)
1391
self.build_tree_contents([('a', b'a'),
1395
self.build_tree_contents([('a','a'),
1393
('1/a', b'redirected once'),
1397
('1/a', 'redirected once'),
1395
('2/a', b'redirected twice'),
1399
('2/a', 'redirected twice'),
1397
('3/a', b'redirected thrice'),
1401
('3/a', 'redirected thrice'),
1399
('4/a', b'redirected 4 times'),
1403
('4/a', 'redirected 4 times'),
1401
('5/a', b'redirected 5 times'),
1405
('5/a', 'redirected 5 times'),
1408
self.old_transport = self._transport(self.old_server.get_url())
1410
def create_transport_secondary_server(self):
1411
"""Create the secondary server, redirections are defined in the tests"""
1412
return http_utils.HTTPServerRedirecting(
1413
protocol_version=self._protocol_version)
1404
1415
def test_one_redirection(self):
1405
t = self.get_old_transport()
1416
t = self.old_transport
1418
req = RedirectedRequest('GET', t.abspath('a'))
1406
1419
new_prefix = 'http://%s:%s' % (self.new_server.host,
1407
1420
self.new_server.port)
1408
1421
self.old_server.redirections = \
1409
[('(.*)', r'%s/1\1' % (new_prefix), 301), ]
1412
t.request('GET', t._remote_path('a'), retries=1).read())
1422
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1423
self.assertEqual('redirected once',t._perform(req).read())
1414
1425
def test_five_redirections(self):
1415
t = self.get_old_transport()
1426
t = self.old_transport
1428
req = RedirectedRequest('GET', t.abspath('a'))
1416
1429
old_prefix = 'http://%s:%s' % (self.old_server.host,
1417
1430
self.old_server.port)
1418
1431
new_prefix = 'http://%s:%s' % (self.new_server.host,
1424
1437
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1425
1438
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1428
b'redirected 5 times',
1429
t.request('GET', t._remote_path('a'), retries=6).read())
1440
self.assertEqual('redirected 5 times',t._perform(req).read())
1432
1443
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1433
1444
"""Test transport.do_catching_redirections."""
1435
scenarios = multiply_scenarios(
1436
vary_by_http_client_implementation(),
1437
vary_by_http_protocol_version(),
1440
1446
def setUp(self):
1441
1447
super(TestDoCatchRedirections, self).setUp()
1442
self.build_tree_contents([('a', b'0123456789'), ],)
1443
cleanup_http_redirection_connections(self)
1445
self.old_transport = self.get_old_transport()
1448
self.build_tree_contents([('a', '0123456789'),],)
1450
self.old_transport = self._transport(self.old_server.get_url())
1452
def get_a(self, transport):
1453
return transport.get('a')
1450
1455
def test_no_redirection(self):
1451
t = self.get_new_transport()
1456
t = self._transport(self.new_server.get_url())
1453
1458
# We use None for redirected so that we fail if redirected
1454
self.assertEqual(b'0123456789',
1459
self.assertEqual('0123456789',
1455
1460
transport.do_catching_redirections(
1456
self.get_a, t, None).read())
1461
self.get_a, t, None).read())
1458
1463
def test_one_redirection(self):
1459
1464
self.redirections = 0
1461
def redirected(t, exception, redirection_notice):
1466
def redirected(transport, exception, redirection_notice):
1462
1467
self.redirections += 1
1463
redirected_t = t._redirected_to(exception.source, exception.target)
1468
dir, file = urlutils.split(exception.target)
1469
return self._transport(dir)
1466
self.assertEqual(b'0123456789',
1471
self.assertEqual('0123456789',
1467
1472
transport.do_catching_redirections(
1468
self.get_a, self.old_transport, redirected).read())
1473
self.get_a, self.old_transport, redirected).read())
1469
1474
self.assertEqual(1, self.redirections)
1471
1476
def test_redirection_loop(self):
1630
1621
def _expected_username_prompt(self, scheme):
1631
1622
return (self._username_prompt_prefix
1632
1623
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1633
self.server.host, self.server.port,
1634
self.server.auth_realm))
1624
self.server.host, self.server.port,
1625
self.server.auth_realm))
1636
1627
def test_no_prompt_for_password_when_using_auth_config(self):
1628
if self._testing_pycurl():
1629
raise tests.TestNotApplicable(
1630
'pycurl does not support authentication.conf'
1631
' since it cannot prompt')
1638
1634
password = 'foo'
1639
1635
stdin_content = 'bar\n' # Not the right password
1640
1636
self.server.add_user(user, password)
1641
1637
t = self.get_user_transport(user, None)
1642
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1638
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1639
stderr=tests.StringIOWrapper())
1643
1640
# Create a minimal config file with the right password
1644
_setup_authentication_config(scheme='http', port=self.server.port,
1645
user=user, password=password)
1641
conf = config.AuthenticationConfig()
1642
conf._get_config().update(
1643
{'httptest': {'scheme': 'http', 'port': self.server.port,
1644
'user': user, 'password': password}})
1646
1646
# Issue a request to the server to connect
1647
with t.get('a') as f:
1648
self.assertEqual(b'contents of a\n', f.read())
1647
self.assertEqual('contents of a\n',t.get('a').read())
1649
1648
# stdin should have been left untouched
1650
1649
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1651
1650
# Only one 'Authentication Required' error should occur
1652
1651
self.assertEqual(1, self.server.auth_required_errors)
1653
def test_user_from_auth_conf(self):
1654
if self._testing_pycurl():
1655
raise tests.TestNotApplicable(
1656
'pycurl does not support authentication.conf')
1659
self.server.add_user(user, password)
1660
# Create a minimal config file with the right password
1661
conf = config.AuthenticationConfig()
1662
conf._get_config().update(
1663
{'httptest': {'scheme': 'http', 'port': self.server.port,
1664
'user': user, 'password': password}})
1666
t = self.get_user_transport(None, None)
1667
# Issue a request to the server to connect
1668
self.assertEqual('contents of a\n', t.get('a').read())
1669
# Only one 'Authentication Required' error should occur
1670
self.assertEqual(1, self.server.auth_required_errors)
1654
1672
def test_changing_nonce(self):
1655
1673
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1656
1674
http_utils.ProxyDigestAuthServer):
1657
1675
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1676
if self._testing_pycurl():
1677
raise tests.KnownFailure(
1678
'pycurl does not handle a nonce change')
1658
1679
self.server.add_user('joe', 'foo')
1659
1680
t = self.get_user_transport('joe', 'foo')
1660
with t.get('a') as f:
1661
self.assertEqual(b'contents of a\n', f.read())
1662
with t.get('b') as f:
1663
self.assertEqual(b'contents of b\n', f.read())
1681
self.assertEqual('contents of a\n', t.get('a').read())
1682
self.assertEqual('contents of b\n', t.get('b').read())
1664
1683
# Only one 'Authentication Required' error should have
1665
1684
# occured so far
1666
1685
self.assertEqual(1, self.server.auth_required_errors)
1667
1686
# The server invalidates the current nonce
1668
1687
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1669
self.assertEqual(b'contents of a\n', t.get('a').read())
1688
self.assertEqual('contents of a\n', t.get('a').read())
1670
1689
# Two 'Authentication Required' errors should occur (the
1671
1690
# initial 'who are you' and a second 'who are you' with the new nonce)
1672
1691
self.assertEqual(2, self.server.auth_required_errors)
1674
def test_user_from_auth_conf(self):
1677
self.server.add_user(user, password)
1678
_setup_authentication_config(scheme='http', port=self.server.port,
1679
user=user, password=password)
1680
t = self.get_user_transport(None, None)
1681
# Issue a request to the server to connect
1682
with t.get('a') as f:
1683
self.assertEqual(b'contents of a\n', f.read())
1684
# Only one 'Authentication Required' error should occur
1685
self.assertEqual(1, self.server.auth_required_errors)
1687
def test_no_credential_leaks_in_log(self):
1688
self.overrideAttr(debug, 'debug_flags', {'http'})
1690
password = 'very-sensitive-password'
1691
self.server.add_user(user, password)
1692
t = self.get_user_transport(user, password)
1693
# Capture the debug calls to mutter
1697
lines = args[0] % args[1:]
1698
# Some calls output multiple lines, just split them now since we
1699
# care about a single one later.
1700
self.mutters.extend(lines.splitlines())
1701
self.overrideAttr(trace, 'mutter', mutter)
1702
# Issue a request to the server to connect
1703
self.assertEqual(True, t.has('a'))
1704
# Only one 'Authentication Required' error should occur
1705
self.assertEqual(1, self.server.auth_required_errors)
1706
# Since the authentification succeeded, there should be a corresponding
1708
sent_auth_headers = [line for line in self.mutters
1709
if line.startswith('> %s' % (self._auth_header,))]
1710
self.assertLength(1, sent_auth_headers)
1711
self.assertStartsWith(sent_auth_headers[0],
1712
'> %s: <masked>' % (self._auth_header,))
1715
1695
class TestProxyAuth(TestAuth):
1716
"""Test proxy authentication schemes.
1718
This inherits from TestAuth to tweak the setUp and filter some failing
1722
scenarios = multiply_scenarios(
1723
vary_by_http_client_implementation(),
1724
vary_by_http_protocol_version(),
1725
vary_by_http_proxy_auth_scheme(),
1696
"""Test proxy authentication schemes."""
1698
_auth_header = 'Proxy-authorization'
1699
_password_prompt_prefix = 'Proxy '
1700
_username_prompt_prefix = 'Proxy '
1728
1702
def setUp(self):
1729
1703
super(TestProxyAuth, self).setUp()
1705
self.addCleanup(self._restore_env)
1730
1706
# Override the contents to avoid false positives
1731
self.build_tree_contents([('a', b'not proxied contents of a\n'),
1732
('b', b'not proxied contents of b\n'),
1733
('a-proxied', b'contents of a\n'),
1734
('b-proxied', b'contents of b\n'),
1707
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1708
('b', 'not proxied contents of b\n'),
1709
('a-proxied', 'contents of a\n'),
1710
('b-proxied', 'contents of b\n'),
1737
1713
def get_user_transport(self, user, password):
1738
proxy_url = self.get_user_url(user, password)
1739
self.overrideEnv('all_proxy', proxy_url)
1740
return TestAuth.get_user_transport(self, user, password)
1743
class NonClosingBytesIO(io.BytesIO):
1746
"""Ignore and leave file open."""
1714
self._install_env({'all_proxy': self.get_user_url(user, password)})
1715
return self._transport(self.server.get_url())
1717
def _install_env(self, env):
1718
for name, value in env.iteritems():
1719
self._old_env[name] = osutils.set_or_unset_env(name, value)
1721
def _restore_env(self):
1722
for name, value in self._old_env.iteritems():
1723
osutils.set_or_unset_env(name, value)
1725
def test_empty_pass(self):
1726
if self._testing_pycurl():
1728
if pycurl.version_info()[1] < '7.16.0':
1729
raise tests.KnownFailure(
1730
'pycurl < 7.16.0 does not handle empty proxy passwords')
1731
super(TestProxyAuth, self).test_empty_pass()
1749
1734
class SampleSocket(object):
2143
2095
t = self.get_transport()
2144
2096
# We must send a single line of body bytes, see
2145
# PredefinedRequestHandler._handle_one_request
2146
code, f = t._post(b'abc def end-of-body\n')
2147
self.assertEqual(b'lalala whatever as long as itsssss\n', f.read())
2097
# PredefinedRequestHandler.handle_one_request
2098
code, f = t._post('abc def end-of-body\n')
2099
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2148
2100
self.assertActivitiesMatch()
2151
2103
class TestActivity(tests.TestCase, TestActivityMixin):
2153
scenarios = multiply_scenarios(
2154
vary_by_http_activity(),
2155
vary_by_http_protocol_version(),
2158
2105
def setUp(self):
2159
super(TestActivity, self).setUp()
2160
TestActivityMixin.setUp(self)
2106
tests.TestCase.setUp(self)
2107
self.server = self._activity_server(self._protocol_version)
2108
self.server.start_server()
2109
self.activities = {}
2110
def report_activity(t, bytes, direction):
2111
count = self.activities.get(direction, 0)
2113
self.activities[direction] = count
2115
# We override at class level because constructors may propagate the
2116
# bound method and render instance overriding ineffective (an
2117
# alternative would be to define a specific ui factory instead...)
2118
self.orig_report_activity = self._transport._report_activity
2119
self._transport._report_activity = report_activity
2122
self._transport._report_activity = self.orig_report_activity
2123
self.server.stop_server()
2124
tests.TestCase.tearDown(self)
2163
2127
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2165
# Unlike TestActivity, we are really testing ReportingFileSocket and
2166
# ReportingSocket, so we don't need all the parametrization. Since
2167
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2168
# test them through their use by the transport than directly (that's a
2169
# bit less clean but far more simpler and effective).
2170
_activity_server = ActivityHTTPServer
2171
_protocol_version = 'HTTP/1.1'
2173
2129
def setUp(self):
2174
super(TestNoReportActivity, self).setUp()
2175
self._transport = HttpTransport
2176
TestActivityMixin.setUp(self)
2130
tests.TestCase.setUp(self)
2131
# Unlike TestActivity, we are really testing ReportingFileSocket and
2132
# ReportingSocket, so we don't need all the parametrization. Since
2133
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2134
# test them through their use by the transport than directly (that's a
2135
# bit less clean but far more simpler and effective).
2136
self.server = ActivityHTTPServer('HTTP/1.1')
2137
self._transport=_urllib.HttpTransport_urllib
2139
self.server.start_server()
2141
# We override at class level because constructors may propagate the
2142
# bound method and render instance overriding ineffective (an
2143
# alternative would be to define a specific ui factory instead...)
2144
self.orig_report_activity = self._transport._report_activity
2145
self._transport._report_activity = None
2148
self._transport._report_activity = self.orig_report_activity
2149
self.server.stop_server()
2150
tests.TestCase.tearDown(self)
2178
2152
def assertActivitiesMatch(self):
2179
2153
# Nothing to check here
2183
2157
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2184
2158
"""Test authentication on the redirected http server."""
2186
scenarios = vary_by_http_protocol_version()
2188
2160
_auth_header = 'Authorization'
2189
2161
_password_prompt_prefix = ''
2190
2162
_username_prompt_prefix = ''
2191
2163
_auth_server = http_utils.HTTPBasicAuthServer
2192
_transport = HttpTransport
2164
_transport = _urllib.HttpTransport_urllib
2166
def create_transport_readonly_server(self):
2167
return self._auth_server(protocol_version=self._protocol_version)
2169
def create_transport_secondary_server(self):
2170
"""Create the secondary server redirecting to the primary server"""
2171
new = self.get_readonly_server()
2173
redirecting = http_utils.HTTPServerRedirecting(
2174
protocol_version=self._protocol_version)
2175
redirecting.redirect_to(new.host, new.port)
2194
2178
def setUp(self):
2195
2179
super(TestAuthOnRedirected, self).setUp()
2196
self.build_tree_contents([('a', b'a'),
2180
self.build_tree_contents([('a','a'),
2198
('1/a', b'redirected once'),
2182
('1/a', 'redirected once'),
2200
2184
new_prefix = 'http://%s:%s' % (self.new_server.host,
2201
2185
self.new_server.port)
2202
2186
self.old_server.redirections = [
2203
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2204
self.old_transport = self.get_old_transport()
2187
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2188
self.old_transport = self._transport(self.old_server.get_url())
2205
2189
self.new_server.add_user('joe', 'foo')
2206
cleanup_http_redirection_connections(self)
2208
def create_transport_readonly_server(self):
2209
server = self._auth_server(protocol_version=self._protocol_version)
2210
server._url_protocol = self._url_protocol
2191
def get_a(self, transport):
2192
return transport.get('a')
2216
2194
def test_auth_on_redirected_via_do_catching_redirections(self):
2217
2195
self.redirections = 0
2219
def redirected(t, exception, redirection_notice):
2197
def redirected(transport, exception, redirection_notice):
2220
2198
self.redirections += 1
2221
redirected_t = t._redirected_to(exception.source, exception.target)
2222
self.addCleanup(redirected_t.disconnect)
2199
dir, file = urlutils.split(exception.target)
2200
return self._transport(dir)
2225
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2226
self.assertEqual(b'redirected once',
2202
stdout = tests.StringIOWrapper()
2203
stderr = tests.StringIOWrapper()
2204
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2205
stdout=stdout, stderr=stderr)
2206
self.assertEqual('redirected once',
2227
2207
transport.do_catching_redirections(
2228
self.get_a, self.old_transport, redirected).read())
2208
self.get_a, self.old_transport, redirected).read())
2229
2209
self.assertEqual(1, self.redirections)
2230
2210
# stdin should be empty
2231
2211
self.assertEqual('', ui.ui_factory.stdin.readline())
2232
2212
# stdout should be empty, stderr will contains the prompts
2233
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2213
self.assertEqual('', stdout.getvalue())
2235
2215
def test_auth_on_redirected_via_following_redirections(self):
2236
2216
self.new_server.add_user('joe', 'foo')
2237
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2217
stdout = tests.StringIOWrapper()
2218
stderr = tests.StringIOWrapper()
2219
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2220
stdout=stdout, stderr=stderr)
2238
2221
t = self.old_transport
2222
req = RedirectedRequest('GET', t.abspath('a'))
2239
2223
new_prefix = 'http://%s:%s' % (self.new_server.host,
2240
2224
self.new_server.port)
2241
2225
self.old_server.redirections = [
2242
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2245
t.request('GET', t.abspath('a'), retries=3).read())
2226
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2227
self.assertEqual('redirected once',t._perform(req).read())
2246
2228
# stdin should be empty
2247
2229
self.assertEqual('', ui.ui_factory.stdin.readline())
2248
2230
# stdout should be empty, stderr will contains the prompts
2249
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2231
self.assertEqual('', stdout.getvalue())
added
removed
bzrlib/tests/test_http.py
