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
# 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 = [
96
135
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
97
136
('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():
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((
125
149
activity_scenarios = [
126
150
('urllib,http', dict(_activity_server=ActivityHTTPServer,
127
_transport=HttpTransport,)),
151
_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'))
153
if tests.HTTPSServerFeature.available():
145
154
activity_scenarios.append(
146
155
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
147
_transport=HTTPS_transport,)),)
148
return activity_scenarios
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)
151
190
class FakeManager(object):
368
366
server = http_server.HttpServer(RequestHandlerOneOne,
369
367
protocol_version='HTTP/1.0')
370
368
self.start_server(server)
371
self.assertIsInstance(server.server,
369
self.assertIsInstance(server._httpd,
372
370
http_server.TestingHTTPServer)
373
class TestWithTransport_pycurl(object):
374
"""Test case to inherit from if pycurl is present"""
376
def _get_pycurl_maybe(self):
377
self.requireFeature(features.pycurl)
378
return PyCurlTransport
380
_transport = property(_get_pycurl_maybe)
383
class TestHttpUrls(tests.TestCase):
385
# TODO: This should be moved to authorization tests once they
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'],
375
401
class TestHttpTransportUrls(tests.TestCase):
376
402
"""Test the http urls."""
378
scenarios = vary_by_http_client_implementation()
380
404
def test_abs_url(self):
381
405
"""Construction of absolute http URLs"""
382
t = self._transport('http://example.com/bzr/bzr.dev/')
406
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
383
407
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')
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')
387
411
eq(t.abspath('.bzr/1//2/./3'),
388
'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
412
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
390
414
def test_invalid_http_urls(self):
391
415
"""Trap invalid construction of urls"""
392
self._transport('http://example.com/bzr/bzr.dev/')
393
self.assertRaises(urlutils.InvalidURL,
416
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
417
self.assertRaises(errors.InvalidURL,
395
'http://example.com:port/bzr/bzr.dev/')
419
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
397
421
def test_http_root_urls(self):
398
422
"""Construction of URLs from server root"""
399
t = self._transport('http://example.com/')
423
t = self._transport('http://bzr.ozlabs.org/')
400
424
eq = self.assertEqualDiff
401
425
eq(t.abspath('.bzr/tree-version'),
402
'http://example.com/.bzr/tree-version')
426
'http://bzr.ozlabs.org/.bzr/tree-version')
404
428
def test_http_impl_urls(self):
405
429
"""There are servers which ask for particular clients to connect"""
407
431
server.start_server()
409
433
url = server.get_url()
410
self.assertTrue(url.startswith('%s://' % self._url_protocol))
434
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
412
436
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')
415
475
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
416
476
"""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()
479
http_utils.TestCaseWithWebserver.setUp(self)
425
480
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
426
481
transport=self.get_transport())
428
483
def test_http_has(self):
429
484
server = self.get_readonly_server()
430
t = self.get_readonly_transport()
485
t = self._transport(server.get_url())
431
486
self.assertEqual(t.has('foo/bar'), True)
432
487
self.assertEqual(len(server.logs), 1)
433
488
self.assertContainsRe(server.logs[0],
434
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "Breezy/')
489
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
436
491
def test_http_has_not_found(self):
437
492
server = self.get_readonly_server()
438
t = self.get_readonly_transport()
493
t = self._transport(server.get_url())
439
494
self.assertEqual(t.has('not-found'), False)
440
495
self.assertContainsRe(server.logs[1],
441
r'"HEAD /not-found HTTP/1.." 404 - "-" "Breezy/')
496
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
443
498
def test_http_get(self):
444
499
server = self.get_readonly_server()
445
t = self.get_readonly_transport()
500
t = self._transport(server.get_url())
446
501
fp = t.get('foo/bar')
447
502
self.assertEqualDiff(
449
b'contents of foo/bar\n')
504
'contents of foo/bar\n')
450
505
self.assertEqual(len(server.logs), 1)
451
506
self.assertTrue(server.logs[0].find(
452
'"GET /foo/bar HTTP/1.1" 200 - "-" "Breezy/%s'
453
% breezy.__version__) > -1)
507
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
508
% bzrlib.__version__) > -1)
455
510
def test_has_on_bogus_host(self):
456
511
# Get a free address and don't 'accept' on it, so that we
470
525
class TestHttpTransportRegistration(tests.TestCase):
471
526
"""Test registrations of various http implementations"""
473
scenarios = vary_by_http_client_implementation()
475
528
def test_http_registered(self):
476
t = transport.get_transport_from_url(
477
'%s://foo.com/' % self._url_protocol)
529
t = transport.get_transport('%s://foo.com/' % self._qualified_prefix)
478
530
self.assertIsInstance(t, transport.Transport)
479
531
self.assertIsInstance(t, self._transport)
482
534
class TestPost(tests.TestCase):
484
scenarios = multiply_scenarios(
485
vary_by_http_client_implementation(),
486
vary_by_http_protocol_version(),
489
536
def test_post_body_is_received(self):
490
server = RecordingServer(expect_body_tail=b'end-of-body',
491
scheme=self._url_protocol)
537
server = RecordingServer(expect_body_tail='end-of-body',
538
scheme=self._qualified_prefix)
492
539
self.start_server(server)
493
540
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())
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())
503
546
# The transport should not be assuming that the server can accept
504
547
# chunked encoding the first time it connects, because HTTP/1.1, so we
505
548
# check for the literal string.
507
server.received_bytes.endswith(b'\r\n\r\nabc def end-of-body'))
550
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
510
553
class TestRangeHeader(tests.TestCase):
511
554
"""Test range_header method"""
513
556
def check_header(self, value, ranges=[], tail=0):
514
offsets = [(start, end - start + 1) for start, end in ranges]
557
offsets = [ (start, end - start + 1) for start, end in ranges]
515
558
coalesce = transport.Transport._coalesce_offsets
516
559
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
517
range_header = http.HttpTransport._range_header
560
range_header = http.HttpTransportBase._range_header
518
561
self.assertEqual(value, range_header(coalesced, tail))
520
563
def test_range_header_single(self):
521
self.check_header('0-9', ranges=[(0, 9)])
522
self.check_header('100-109', ranges=[(100, 109)])
564
self.check_header('0-9', ranges=[(0,9)])
565
self.check_header('100-109', ranges=[(100,109)])
524
567
def test_range_header_tail(self):
525
568
self.check_header('-10', tail=10)
741
791
super(TestRangeRequestServer, self).setUp()
742
self.build_tree_contents([('a', b'0123456789')],)
792
self.build_tree_contents([('a', '0123456789')],)
744
794
def test_readv(self):
745
t = self.get_readonly_transport()
795
server = self.get_readonly_server()
796
t = self._transport(server.get_url())
746
797
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'))
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'))
752
803
def test_readv_out_of_order(self):
753
t = self.get_readonly_transport()
804
server = self.get_readonly_server()
805
t = self._transport(server.get_url())
754
806
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'))
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'))
760
812
def test_readv_invalid_ranges(self):
761
t = self.get_readonly_transport()
813
server = self.get_readonly_server()
814
t = self._transport(server.get_url())
763
816
# This is intentionally reading off the end of the file
764
817
# since we are sure that it cannot get there
765
818
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
766
t.readv, 'a', [(1, 1), (8, 10)])
819
t.readv, 'a', [(1,1), (8,10)])
768
821
# This is trying to seek past the end of the file, it should
769
822
# also raise a special error
770
823
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
771
t.readv, 'a', [(12, 2)])
824
t.readv, 'a', [(12,2)])
773
826
def test_readv_multiple_get_requests(self):
774
827
server = self.get_readonly_server()
775
t = self.get_readonly_transport()
828
t = self._transport(server.get_url())
776
829
# force transport to issue multiple requests
777
830
t._max_readv_combine = 1
778
831
t._max_get_ranges = 1
779
832
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'))
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'))
784
837
# The server should have issued 4 requests
785
838
self.assertEqual(4, server.GET_request_nb)
787
840
def test_readv_get_max_size(self):
788
841
server = self.get_readonly_server()
789
t = self.get_readonly_transport()
842
t = self._transport(server.get_url())
790
843
# force transport to issue multiple requests by limiting the number of
791
844
# bytes by request. Note that this apply to coalesced offsets only, a
792
845
# single range will keep its size even if bigger than the limit.
793
846
t._get_max_size = 2
794
847
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'))
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'))
799
852
# The server should have issued 3 requests
800
853
self.assertEqual(3, server.GET_request_nb)
802
855
def test_complete_readv_leave_pipe_clean(self):
803
856
server = self.get_readonly_server()
804
t = self.get_readonly_transport()
857
t = self._transport(server.get_url())
805
858
# force transport to issue multiple requests
806
859
t._get_max_size = 2
807
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
860
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
808
861
# The server should have issued 3 requests
809
862
self.assertEqual(3, server.GET_request_nb)
810
self.assertEqual(b'0123456789', t.get_bytes('a'))
863
self.assertEqual('0123456789', t.get_bytes('a'))
811
864
self.assertEqual(4, server.GET_request_nb)
813
866
def test_incomplete_readv_leave_pipe_clean(self):
814
867
server = self.get_readonly_server()
815
t = self.get_readonly_transport()
868
t = self._transport(server.get_url())
816
869
# force transport to issue multiple requests
817
870
t._get_max_size = 2
818
871
# Don't collapse readv results into a list so that we leave unread
819
872
# bytes on the socket
820
873
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
821
self.assertEqual((0, b'0'), next(ireadv))
874
self.assertEqual((0, '0'), ireadv.next())
822
875
# The server should have issued one request so far
823
876
self.assertEqual(1, server.GET_request_nb)
824
self.assertEqual(b'0123456789', t.get_bytes('a'))
877
self.assertEqual('0123456789', t.get_bytes('a'))
825
878
# get_bytes issued an additional request, the readv pending ones are
827
880
self.assertEqual(2, server.GET_request_nb)
967
1018
def setUp(self):
968
1019
super(TestTruncatedMultipleRangeServer, self).setUp()
969
self.build_tree_contents([('a', b'0123456789')],)
1020
self.build_tree_contents([('a', '0123456789')],)
971
1022
def test_readv_with_short_reads(self):
972
1023
server = self.get_readonly_server()
973
t = self.get_readonly_transport()
1024
t = self._transport(server.get_url())
974
1025
# Force separate ranges for each offset
975
1026
t._bytes_to_read_before_seek = 0
976
1027
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
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
985
1040
self.assertEqual(3, server.GET_request_nb)
986
1041
# Finally the client have tried a single range request and stays in
988
1043
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
1045
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1058
1046
"""Errors out when range specifiers exceed the limit"""
1095
1078
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1096
1079
protocol_version=self._protocol_version)
1081
def get_transport(self):
1082
return self._transport(self.get_readonly_server().get_url())
1098
1084
def setUp(self):
1099
super(TestLimitedRangeRequestServer, self).setUp()
1085
http_utils.TestCaseWithWebserver.setUp(self)
1100
1086
# We need to manipulate ranges that correspond to real chunks in the
1101
1087
# 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)])
1088
filler = ''.join(['abcdefghij' for x in range(102)])
1089
content = ''.join(['%04d' % v + filler for v in range(16)])
1104
1090
self.build_tree_contents([('a', content)],)
1106
1092
def test_few_ranges(self):
1107
t = self.get_readonly_transport()
1093
t = self.get_transport()
1108
1094
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'))
1095
self.assertEqual(l[0], (0, '0000'))
1096
self.assertEqual(l[1], (1024, '0001'))
1111
1097
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1113
1099
def test_more_ranges(self):
1114
t = self.get_readonly_transport()
1100
t = self.get_transport()
1115
1101
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'))
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'))
1120
1106
# The server will refuse to serve the first request (too much ranges),
1121
1107
# a second request will succeed.
1122
1108
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1128
1114
Only the urllib implementation is tested here.
1118
tests.TestCase.setUp(self)
1123
tests.TestCase.tearDown(self)
1125
def _install_env(self, env):
1126
for name, value in env.iteritems():
1127
self._old_env[name] = osutils.set_or_unset_env(name, value)
1129
def _restore_env(self):
1130
for name, value in self._old_env.iteritems():
1131
osutils.set_or_unset_env(name, value)
1131
1133
def _proxied_request(self):
1132
handler = http.ProxyHandler()
1133
request = http.Request('GET', 'http://baz/buzzle')
1134
handler = _urllib2_wrappers.ProxyHandler()
1135
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
1134
1136
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
1139
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)
1140
self._install_env({'http_proxy': 'http://bar.com'})
1141
request = self._proxied_request()
1142
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1153
1144
def test_invalid_proxy(self):
1154
1145
"""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')
1146
self._install_env({'http_proxy': 'host:1234'})
1147
self.assertRaises(errors.InvalidURL, self._proxied_request)
1179
1150
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1185
1156
to the file names).
1188
scenarios = multiply_scenarios(
1189
vary_by_http_client_implementation(),
1190
vary_by_http_protocol_version(),
1193
1159
# FIXME: We don't have an https server available, so we don't
1194
# test https connections. --vila toolongago
1160
# test https connections.
1196
1162
def setUp(self):
1197
1163
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')])
1164
self.build_tree_contents([('foo', 'contents of foo\n'),
1165
('foo-proxied', 'proxied contents of foo\n')])
1201
1166
# 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
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
1205
1175
# 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())
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())
1216
1217
def test_http_proxy(self):
1217
self.overrideEnv('http_proxy', self.proxy_url)
1218
self.assertProxied()
1218
self.proxied_in_env({'http_proxy': self.proxy_url})
1220
1220
def test_HTTP_PROXY(self):
1221
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1222
self.assertProxied()
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})
1224
1229
def test_all_proxy(self):
1225
self.overrideEnv('all_proxy', self.proxy_url)
1226
self.assertProxied()
1230
self.proxied_in_env({'all_proxy': self.proxy_url})
1228
1232
def test_ALL_PROXY(self):
1229
self.overrideEnv('ALL_PROXY', self.proxy_url)
1230
self.assertProxied()
1233
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1232
1235
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()
1236
self.not_proxied_in_env({'http_proxy': self.proxy_url,
1237
'no_proxy': self.no_proxy_host})
1237
1239
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()
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})
1242
1246
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()
1247
self.not_proxied_in_env({'all_proxy': self.proxy_url,
1248
'no_proxy': self.no_proxy_host})
1247
1250
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()
1251
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
1252
'NO_PROXY': self.no_proxy_host})
1252
1254
def test_http_proxy_without_scheme(self):
1253
self.overrideEnv('http_proxy', self.server_host_port)
1254
self.assertRaises(urlutils.InvalidURL, self.assertProxied)
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})
1257
1266
class TestRanges(http_utils.TestCaseWithWebserver):
1258
1267
"""Test the Range header in GET methods."""
1260
scenarios = multiply_scenarios(
1261
vary_by_http_client_implementation(),
1262
vary_by_http_protocol_version(),
1265
1269
def setUp(self):
1266
super(TestRanges, self).setUp()
1267
self.build_tree_contents([('a', b'0123456789')],)
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())
1269
1275
def create_transport_readonly_server(self):
1270
1276
return http_server.HttpServer(protocol_version=self._protocol_version)
1272
1278
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
1279
offsets = [ (start, end - start + 1) for start, end in ranges]
1280
coalesce = self.transport._coalesce_offsets
1276
1281
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)
1282
code, data = self.transport._get(relpath, coalesced)
1283
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1279
1284
for start, end in ranges:
1280
1285
data.seek(start)
1281
1286
yield data.read(end - start + 1)
1283
1288
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)
1289
code, data = self.transport._get(relpath, [], tail_amount)
1290
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1287
1291
data.seek(-tail_amount, 2)
1288
1292
return data.read(tail_amount)
1290
1294
def test_range_header(self):
1293
[b'0', b'234'], list(self._file_contents('a', [(0, 0), (2, 4)])))
1296
map(self.assertEqual,['0', '234'],
1297
list(self._file_contents('a', [(0,0), (2,4)])),)
1295
1299
def test_range_header_tail(self):
1296
self.assertEqual(b'789', self._file_tail('a', 3))
1300
self.assertEqual('789', self._file_tail('a', 3))
1298
1302
def test_syntactically_invalid_range_header(self):
1299
1303
self.assertListRaises(errors.InvalidHttpRange,
1300
self._file_contents, 'a', [(4, 3)])
1304
self._file_contents, 'a', [(4, 3)])
1302
1306
def test_semantically_invalid_range_header(self):
1303
1307
self.assertListRaises(errors.InvalidHttpRange,
1304
self._file_contents, 'a', [(42, 128)])
1308
self._file_contents, 'a', [(42, 128)])
1307
1311
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1308
1312
"""Test redirection between http servers."""
1310
scenarios = multiply_scenarios(
1311
vary_by_http_client_implementation(),
1312
vary_by_http_protocol_version(),
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)
1315
1323
def setUp(self):
1316
1324
super(TestHTTPRedirections, self).setUp()
1317
self.build_tree_contents([('a', b'0123456789'),
1325
self.build_tree_contents([('a', '0123456789'),
1319
b'# Bazaar revision bundle v0.9\n#\n')
1327
'# Bazaar revision bundle v0.9\n#\n')
1329
# The requests to the old server will be redirected to the new server
1330
self.old_transport = self._transport(self.old_server.get_url())
1322
1332
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):
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):
1331
1347
"""Request following redirections. """
1333
init_orig = http.Request.__init__
1349
init_orig = _urllib2_wrappers.Request.__init__
1335
1351
def __init__(self, method, url, *args, **kwargs):
1336
1352
"""Constructor.
1339
1355
# Since the tests using this class will replace
1340
# http.Request, we can't just call the base class __init__
1356
# _urllib2_wrappers.Request, we can't just call the base class __init__
1341
1357
# or we'll loop.
1342
1358
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1343
1359
self.follow_redirections = True
1346
1362
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)
1363
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1373
1366
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1376
1369
http implementations do not redirect silently anymore (they
1377
1370
do not redirect at all in fact). The mechanism is still in
1378
place at the http.Request level and these tests
1371
place at the _urllib2_wrappers.Request level and these tests
1374
For the pycurl implementation
1375
the redirection have been deleted as we may deprecate pycurl
1376
and I have no place to keep a working implementation.
1382
scenarios = multiply_scenarios(
1383
vary_by_http_client_implementation(),
1384
vary_by_http_protocol_version(),
1387
1380
def setUp(self):
1381
if (features.pycurl.available()
1382
and self._transport == PyCurlTransport):
1383
raise tests.TestNotApplicable(
1384
"pycurl doesn't redirect silently annymore")
1388
1385
super(TestHTTPSilentRedirections, self).setUp()
1389
1386
install_redirected_request(self)
1390
cleanup_http_redirection_connections(self)
1391
self.build_tree_contents([('a', b'a'),
1387
self.build_tree_contents([('a','a'),
1393
('1/a', b'redirected once'),
1389
('1/a', 'redirected once'),
1395
('2/a', b'redirected twice'),
1391
('2/a', 'redirected twice'),
1397
('3/a', b'redirected thrice'),
1393
('3/a', 'redirected thrice'),
1399
('4/a', b'redirected 4 times'),
1395
('4/a', 'redirected 4 times'),
1401
('5/a', b'redirected 5 times'),
1397
('5/a', 'redirected 5 times'),
1400
self.old_transport = self._transport(self.old_server.get_url())
1402
def create_transport_secondary_server(self):
1403
"""Create the secondary server, redirections are defined in the tests"""
1404
return http_utils.HTTPServerRedirecting(
1405
protocol_version=self._protocol_version)
1404
1407
def test_one_redirection(self):
1405
t = self.get_old_transport()
1408
t = self.old_transport
1410
req = RedirectedRequest('GET', t.abspath('a'))
1406
1411
new_prefix = 'http://%s:%s' % (self.new_server.host,
1407
1412
self.new_server.port)
1408
1413
self.old_server.redirections = \
1409
[('(.*)', r'%s/1\1' % (new_prefix), 301), ]
1412
t.request('GET', t._remote_path('a'), retries=1).read())
1414
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1415
self.assertEqual('redirected once',t._perform(req).read())
1414
1417
def test_five_redirections(self):
1415
t = self.get_old_transport()
1418
t = self.old_transport
1420
req = RedirectedRequest('GET', t.abspath('a'))
1416
1421
old_prefix = 'http://%s:%s' % (self.old_server.host,
1417
1422
self.old_server.port)
1418
1423
new_prefix = 'http://%s:%s' % (self.new_server.host,
1424
1429
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1425
1430
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1428
b'redirected 5 times',
1429
t.request('GET', t._remote_path('a'), retries=6).read())
1432
self.assertEqual('redirected 5 times',t._perform(req).read())
1432
1435
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1433
1436
"""Test transport.do_catching_redirections."""
1435
scenarios = multiply_scenarios(
1436
vary_by_http_client_implementation(),
1437
vary_by_http_protocol_version(),
1440
1438
def setUp(self):
1441
1439
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()
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')
1450
1447
def test_no_redirection(self):
1451
t = self.get_new_transport()
1448
t = self._transport(self.new_server.get_url())
1453
1450
# We use None for redirected so that we fail if redirected
1454
self.assertEqual(b'0123456789',
1451
self.assertEqual('0123456789',
1455
1452
transport.do_catching_redirections(
1456
self.get_a, t, None).read())
1453
self.get_a, t, None).read())
1458
1455
def test_one_redirection(self):
1459
1456
self.redirections = 0
1461
def redirected(t, exception, redirection_notice):
1458
def redirected(transport, exception, redirection_notice):
1462
1459
self.redirections += 1
1463
redirected_t = t._redirected_to(exception.source, exception.target)
1460
dir, file = urlutils.split(exception.target)
1461
return self._transport(dir)
1466
self.assertEqual(b'0123456789',
1463
self.assertEqual('0123456789',
1467
1464
transport.do_catching_redirections(
1468
self.get_a, self.old_transport, redirected).read())
1465
self.get_a, self.old_transport, redirected).read())
1469
1466
self.assertEqual(1, self.redirections)
1471
1468
def test_redirection_loop(self):
1630
1613
def _expected_username_prompt(self, scheme):
1631
1614
return (self._username_prompt_prefix
1632
1615
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1633
self.server.host, self.server.port,
1634
self.server.auth_realm))
1616
self.server.host, self.server.port,
1617
self.server.auth_realm))
1636
1619
def test_no_prompt_for_password_when_using_auth_config(self):
1620
if self._testing_pycurl():
1621
raise tests.TestNotApplicable(
1622
'pycurl does not support authentication.conf'
1623
' since it cannot prompt')
1638
1626
password = 'foo'
1639
1627
stdin_content = 'bar\n' # Not the right password
1640
1628
self.server.add_user(user, password)
1641
1629
t = self.get_user_transport(user, None)
1642
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1630
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1631
stderr=tests.StringIOWrapper())
1643
1632
# Create a minimal config file with the right password
1644
_setup_authentication_config(scheme='http', port=self.server.port,
1645
user=user, password=password)
1633
conf = config.AuthenticationConfig()
1634
conf._get_config().update(
1635
{'httptest': {'scheme': 'http', 'port': self.server.port,
1636
'user': user, 'password': password}})
1646
1638
# 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())
1639
self.assertEqual('contents of a\n',t.get('a').read())
1649
1640
# stdin should have been left untouched
1650
1641
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1651
1642
# Only one 'Authentication Required' error should occur
1652
1643
self.assertEqual(1, self.server.auth_required_errors)
1645
def test_user_from_auth_conf(self):
1646
if self._testing_pycurl():
1647
raise tests.TestNotApplicable(
1648
'pycurl does not support authentication.conf')
1651
self.server.add_user(user, password)
1652
# Create a minimal config file with the right password
1653
conf = config.AuthenticationConfig()
1654
conf._get_config().update(
1655
{'httptest': {'scheme': 'http', 'port': self.server.port,
1656
'user': user, 'password': password}})
1658
t = self.get_user_transport(None, None)
1659
# Issue a request to the server to connect
1660
self.assertEqual('contents of a\n', t.get('a').read())
1661
# Only one 'Authentication Required' error should occur
1662
self.assertEqual(1, self.server.auth_required_errors)
1654
1664
def test_changing_nonce(self):
1655
1665
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1656
1666
http_utils.ProxyDigestAuthServer):
1657
1667
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1668
if self._testing_pycurl():
1669
raise tests.KnownFailure(
1670
'pycurl does not handle a nonce change')
1658
1671
self.server.add_user('joe', 'foo')
1659
1672
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())
1673
self.assertEqual('contents of a\n', t.get('a').read())
1674
self.assertEqual('contents of b\n', t.get('b').read())
1664
1675
# Only one 'Authentication Required' error should have
1665
1676
# occured so far
1666
1677
self.assertEqual(1, self.server.auth_required_errors)
1667
1678
# The server invalidates the current nonce
1668
1679
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1669
self.assertEqual(b'contents of a\n', t.get('a').read())
1680
self.assertEqual('contents of a\n', t.get('a').read())
1670
1681
# Two 'Authentication Required' errors should occur (the
1671
1682
# initial 'who are you' and a second 'who are you' with the new nonce)
1672
1683
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
1687
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(),
1688
"""Test proxy authentication schemes."""
1690
_auth_header = 'Proxy-authorization'
1691
_password_prompt_prefix = 'Proxy '
1692
_username_prompt_prefix = 'Proxy '
1728
1694
def setUp(self):
1729
1695
super(TestProxyAuth, self).setUp()
1697
self.addCleanup(self._restore_env)
1730
1698
# 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'),
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'),
1737
1705
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."""
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()
1749
1726
class SampleSocket(object):
2143
2087
t = self.get_transport()
2144
2088
# 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())
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())
2148
2092
self.assertActivitiesMatch()
2151
2095
class TestActivity(tests.TestCase, TestActivityMixin):
2153
scenarios = multiply_scenarios(
2154
vary_by_http_activity(),
2155
vary_by_http_protocol_version(),
2158
2097
def setUp(self):
2159
super(TestActivity, self).setUp()
2160
TestActivityMixin.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)
2163
2119
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
2121
def setUp(self):
2174
super(TestNoReportActivity, self).setUp()
2175
self._transport = HttpTransport
2176
TestActivityMixin.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)
2178
2144
def assertActivitiesMatch(self):
2179
2145
# Nothing to check here
2183
2149
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2184
2150
"""Test authentication on the redirected http server."""
2186
scenarios = vary_by_http_protocol_version()
2188
2152
_auth_header = 'Authorization'
2189
2153
_password_prompt_prefix = ''
2190
2154
_username_prompt_prefix = ''
2191
2155
_auth_server = http_utils.HTTPBasicAuthServer
2192
_transport = HttpTransport
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)
2194
2169
def setUp(self):
2195
2170
super(TestAuthOnRedirected, self).setUp()
2196
self.build_tree_contents([('a', b'a'),
2171
self.build_tree_contents([('a','a'),
2198
('1/a', b'redirected once'),
2173
('1/a', 'redirected once'),
2200
2175
new_prefix = 'http://%s:%s' % (self.new_server.host,
2201
2176
self.new_server.port)
2202
2177
self.old_server.redirections = [
2203
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2204
self.old_transport = self.get_old_transport()
2178
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2179
self.old_transport = self._transport(self.old_server.get_url())
2205
2180
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
2182
def get_a(self, transport):
2183
return transport.get('a')
2216
2185
def test_auth_on_redirected_via_do_catching_redirections(self):
2217
2186
self.redirections = 0
2219
def redirected(t, exception, redirection_notice):
2188
def redirected(transport, exception, redirection_notice):
2220
2189
self.redirections += 1
2221
redirected_t = t._redirected_to(exception.source, exception.target)
2222
self.addCleanup(redirected_t.disconnect)
2190
dir, file = urlutils.split(exception.target)
2191
return self._transport(dir)
2225
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2226
self.assertEqual(b'redirected once',
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',
2227
2198
transport.do_catching_redirections(
2228
self.get_a, self.old_transport, redirected).read())
2199
self.get_a, self.old_transport, redirected).read())
2229
2200
self.assertEqual(1, self.redirections)
2230
2201
# stdin should be empty
2231
2202
self.assertEqual('', ui.ui_factory.stdin.readline())
2232
2203
# stdout should be empty, stderr will contains the prompts
2233
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2204
self.assertEqual('', stdout.getvalue())
2235
2206
def test_auth_on_redirected_via_following_redirections(self):
2236
2207
self.new_server.add_user('joe', 'foo')
2237
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2208
stdout = tests.StringIOWrapper()
2209
stderr = tests.StringIOWrapper()
2210
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2211
stdout=stdout, stderr=stderr)
2238
2212
t = self.old_transport
2213
req = RedirectedRequest('GET', t.abspath('a'))
2239
2214
new_prefix = 'http://%s:%s' % (self.new_server.host,
2240
2215
self.new_server.port)
2241
2216
self.old_server.redirections = [
2242
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2245
t.request('GET', t.abspath('a'), retries=3).read())
2217
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2218
self.assertEqual('redirected once',t._perform(req).read())
2246
2219
# stdin should be empty
2247
2220
self.assertEqual('', ui.ui_factory.stdin.readline())
2248
2221
# stdout should be empty, stderr will contains the prompts
2249
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2222
self.assertEqual('', stdout.getvalue())
added
removed
bzrlib/tests/test_http.py
