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