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
28
from http.server import SimpleHTTPRequestHandler
29
except ImportError: # python < 3
30
from httplib import UnknownProtocol
31
from SimpleHTTPServer import SimpleHTTPRequestHandler
27
import SimpleHTTPServer
39
remote as _mod_remote,
51
remote as _mod_remote,
45
from bzrlib.tests import (
59
from .scenarios import (
51
from bzrlib.tests.scenarios import (
60
52
load_tests_apply_scenarios,
61
53
multiply_scenarios,
63
from ..transport import (
55
from bzrlib.transport import (
67
from ..transport.http import (
59
from bzrlib.transport.http import (
65
if features.pycurl.available():
66
from bzrlib.transport.http._pycurl import PyCurlTransport
73
69
load_tests = load_tests_apply_scenarios
76
72
def vary_by_http_client_implementation():
77
"""Test the libraries we can use, currently just urllib."""
73
"""Test the two libraries we can use, pycurl and urllib."""
78
74
transport_scenarios = [
79
('urllib', dict(_transport=HttpTransport,
80
_server=http_server.HttpServer,
81
_url_protocol='http',)),
75
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
76
_server=http_server.HttpServer_urllib,
77
_url_protocol='http+urllib',)),
79
if features.pycurl.available():
80
transport_scenarios.append(
81
('pycurl', dict(_transport=PyCurlTransport,
82
_server=http_server.HttpServer_PyCurl,
83
_url_protocol='http+pycurl',)))
83
84
return transport_scenarios
124
125
def vary_by_http_activity():
125
126
activity_scenarios = [
126
127
('urllib,http', dict(_activity_server=ActivityHTTPServer,
127
_transport=HttpTransport,)),
128
_transport=_urllib.HttpTransport_urllib,)),
130
if features.pycurl.available():
131
activity_scenarios.append(
132
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
133
_transport=PyCurlTransport,)),)
129
134
if features.HTTPSServerFeature.available():
130
135
# FIXME: Until we have a better way to handle self-signed certificates
131
136
# (like allowing them in a test specific authentication.conf for
132
# example), we need some specialized urllib transport for tests.
137
# example), we need some specialized pycurl/urllib transport for tests.
133
138
# -- vila 2012-01-20
139
from bzrlib.tests import (
137
class HTTPS_transport(HttpTransport):
142
class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
139
144
def __init__(self, base, _from_transport=None):
140
super(HTTPS_transport, self).__init__(
145
super(HTTPS_urllib_transport, self).__init__(
141
146
base, _from_transport=_from_transport,
142
147
ca_certs=ssl_certs.build_path('ca.crt'))
144
149
activity_scenarios.append(
145
150
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
146
_transport=HTTPS_transport,)),)
151
_transport=HTTPS_urllib_transport,)),)
152
if features.pycurl.available():
153
class HTTPS_pycurl_transport(PyCurlTransport):
155
def __init__(self, base, _from_transport=None):
156
super(HTTPS_pycurl_transport, self).__init__(
157
base, _from_transport)
158
self.cabundle = str(ssl_certs.build_path('ca.crt'))
160
activity_scenarios.append(
161
('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
162
_transport=HTTPS_pycurl_transport,)),)
147
163
return activity_scenarios
244
260
self.assertEqual('basic', scheme)
245
261
self.assertEqual('realm="Thou should not pass"', remainder)
247
def test_build_basic_header_with_long_creds(self):
248
handler = _urllib2_wrappers.BasicAuthHandler()
249
user = 'user' * 10 # length 40
250
password = 'password' * 5 # length 40
251
header = handler.build_auth_header(
252
dict(user=user, password=password), None)
253
# https://bugs.launchpad.net/bzr/+bug/1606203 was caused by incorrectly
254
# creating a header value with an embedded '\n'
255
self.assertFalse('\n' in header)
257
263
def test_basic_extract_realm(self):
258
264
scheme, remainder = self.parse_header(
259
265
'Basic realm="Thou should not pass"',
282
288
self.req_handler = RequestHandler(None, None, None)
284
290
def assertRanges(self, ranges, header, file_size):
285
self.assertEqual(ranges,
291
self.assertEquals(ranges,
286
292
self.req_handler._parse_ranges(header, file_size))
288
294
def test_simple_range(self):
289
self.assertRanges([(0, 2)], 'bytes=0-2', 12)
295
self.assertRanges([(0,2)], 'bytes=0-2', 12)
291
297
def test_tail(self):
292
298
self.assertRanges([(8, 11)], 'bytes=-4', 12)
312
318
protocol_version = 'HTTP/0.1'
314
self.assertRaises(UnknownProtocol,
320
self.assertRaises(httplib.UnknownProtocol,
315
321
http_server.HttpServer, BogusRequestHandler)
317
323
def test_force_invalid_protocol(self):
318
self.assertRaises(UnknownProtocol,
324
self.assertRaises(httplib.UnknownProtocol,
319
325
http_server.HttpServer, protocol_version='HTTP/0.1')
321
327
def test_server_start_and_stop(self):
368
374
http_server.TestingHTTPServer)
377
class TestWithTransport_pycurl(object):
378
"""Test case to inherit from if pycurl is present"""
380
def _get_pycurl_maybe(self):
381
self.requireFeature(features.pycurl)
382
return PyCurlTransport
384
_transport = property(_get_pycurl_maybe)
371
387
class TestHttpTransportUrls(tests.TestCase):
372
388
"""Test the http urls."""
386
402
def test_invalid_http_urls(self):
387
403
"""Trap invalid construction of urls"""
388
404
self._transport('http://example.com/bzr/bzr.dev/')
389
self.assertRaises(urlutils.InvalidURL,
405
self.assertRaises(errors.InvalidURL,
391
407
'http://http://example.com/bzr/bzr.dev/')
408
424
server.stop_server()
427
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
429
# TODO: This should really be moved into another pycurl
430
# specific test. When https tests will be implemented, take
431
# this one into account.
432
def test_pycurl_without_https_support(self):
433
"""Test that pycurl without SSL do not fail with a traceback.
435
For the purpose of the test, we force pycurl to ignore
436
https by supplying a fake version_info that do not
439
self.requireFeature(features.pycurl)
440
# Import the module locally now that we now it's available.
441
pycurl = features.pycurl.module
443
self.overrideAttr(pycurl, 'version_info',
444
# Fake the pycurl version_info This was taken from
445
# a windows pycurl without SSL (thanks to bialix)
454
('ftp', 'gopher', 'telnet',
455
'dict', 'ldap', 'http', 'file'),
459
self.assertRaises(errors.DependencyNotPresent, self._transport,
460
'https://launchpad.net')
411
463
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
412
464
"""Test the http connections."""
427
479
self.assertEqual(t.has('foo/bar'), True)
428
480
self.assertEqual(len(server.logs), 1)
429
481
self.assertContainsRe(server.logs[0],
430
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "Breezy/')
482
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
432
484
def test_http_has_not_found(self):
433
485
server = self.get_readonly_server()
434
486
t = self.get_readonly_transport()
435
487
self.assertEqual(t.has('not-found'), False)
436
488
self.assertContainsRe(server.logs[1],
437
r'"HEAD /not-found HTTP/1.." 404 - "-" "Breezy/')
489
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
439
491
def test_http_get(self):
440
492
server = self.get_readonly_server()
445
497
'contents of foo/bar\n')
446
498
self.assertEqual(len(server.logs), 1)
447
499
self.assertTrue(server.logs[0].find(
448
'"GET /foo/bar HTTP/1.1" 200 - "-" "Breezy/%s'
449
% breezy.__version__) > -1)
500
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
501
% bzrlib.__version__) > -1)
451
503
def test_has_on_bogus_host(self):
452
504
# Get a free address and don't 'accept' on it, so that we
509
561
offsets = [ (start, end - start + 1) for start, end in ranges]
510
562
coalesce = transport.Transport._coalesce_offsets
511
563
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
512
range_header = http.HttpTransport._range_header
564
range_header = http.HttpTransportBase._range_header
513
565
self.assertEqual(value, range_header(coalesced, tail))
515
567
def test_range_header_single(self):
516
self.check_header('0-9', ranges=[(0, 9)])
517
self.check_header('100-109', ranges=[(100, 109)])
568
self.check_header('0-9', ranges=[(0,9)])
569
self.check_header('100-109', ranges=[(100,109)])
519
571
def test_range_header_tail(self):
520
572
self.check_header('-10', tail=10)
523
575
def test_range_header_multi(self):
524
576
self.check_header('0-9,100-200,300-5000',
525
ranges=[(0, 9), (100, 200), (300, 5000)])
577
ranges=[(0,9), (100, 200), (300,5000)])
527
579
def test_range_header_mixed(self):
528
580
self.check_header('0-9,300-5000,-50',
529
ranges=[(0, 9), (300, 5000)],
581
ranges=[(0,9), (300,5000)],
550
602
server._url_protocol = self._url_protocol
605
def _testing_pycurl(self):
606
# TODO: This is duplicated for lots of the classes in this file
607
return (features.pycurl.available()
608
and self._transport == PyCurlTransport)
554
611
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
555
612
"""Whatever request comes in, close the connection"""
570
627
# for details) make no distinction between a closed
571
628
# socket and badly formatted status line, so we can't
572
629
# just test for ConnectionError, we have to test
573
# InvalidHttpResponse too.
574
self.assertRaises((errors.ConnectionError,
575
errors.InvalidHttpResponse),
630
# InvalidHttpResponse too. And pycurl may raise ConnectionReset
631
# instead of ConnectionError too.
632
self.assertRaises(( errors.ConnectionError, errors.ConnectionReset,
633
errors.InvalidHttpResponse),
576
634
t.has, 'foo/bar')
578
636
def test_http_get(self):
665
723
_req_handler_class = BadProtocolRequestHandler
726
if self._testing_pycurl():
727
raise tests.TestNotApplicable(
728
"pycurl doesn't check the protocol version")
729
super(TestBadProtocolServer, self).setUp()
667
731
def test_http_has(self):
668
732
t = self.get_readonly_transport()
669
733
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
759
823
# This is intentionally reading off the end of the file
760
824
# since we are sure that it cannot get there
761
825
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
762
t.readv, 'a', [(1, 1), (8, 10)])
826
t.readv, 'a', [(1,1), (8,10)])
764
828
# This is trying to seek past the end of the file, it should
765
829
# also raise a special error
766
830
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
767
t.readv, 'a', [(12, 2)])
831
t.readv, 'a', [(12,2)])
769
833
def test_readv_multiple_get_requests(self):
770
834
server = self.get_readonly_server()
814
878
# Don't collapse readv results into a list so that we leave unread
815
879
# bytes on the socket
816
880
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
817
self.assertEqual((0, '0'), next(ireadv))
881
self.assertEqual((0, '0'), ireadv.next())
818
882
# The server should have issued one request so far
819
883
self.assertEqual(1, server.GET_request_nb)
820
884
self.assertEqual('0123456789', t.get_bytes('a'))
885
949
self.send_header('Accept-Ranges', 'bytes')
886
950
# XXX: this is strange; the 'random' name below seems undefined and
887
951
# yet the tests pass -- mbp 2010-10-11 bug 658773
888
boundary = "%d" % random.randint(0, 0x7FFFFFFF)
952
boundary = "%d" % random.randint(0,0x7FFFFFFF)
889
953
self.send_header("Content-Type",
890
954
"multipart/byteranges; boundary=%s" % boundary)
891
955
self.end_headers()
963
1027
def setUp(self):
964
1028
super(TestTruncatedMultipleRangeServer, self).setUp()
965
self.build_tree_contents([('a', b'0123456789')],)
1029
self.build_tree_contents([('a', '0123456789')],)
967
1031
def test_readv_with_short_reads(self):
968
1032
server = self.get_readonly_server()
970
1034
# Force separate ranges for each offset
971
1035
t._bytes_to_read_before_seek = 0
972
1036
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
973
self.assertEqual((0, '0'), next(ireadv))
974
self.assertEqual((2, '2'), next(ireadv))
975
# Only one request have been issued so far
976
self.assertEqual(1, server.GET_request_nb)
977
self.assertEqual((4, '45'), next(ireadv))
978
self.assertEqual((9, '9'), next(ireadv))
979
# We issue 3 requests: two multiple (4 ranges, then 2 ranges) then a
1037
self.assertEqual((0, '0'), ireadv.next())
1038
self.assertEqual((2, '2'), ireadv.next())
1039
if not self._testing_pycurl():
1040
# Only one request have been issued so far (except for pycurl that
1041
# try to read the whole response at once)
1042
self.assertEqual(1, server.GET_request_nb)
1043
self.assertEqual((4, '45'), ireadv.next())
1044
self.assertEqual((9, '9'), ireadv.next())
1045
# Both implementations issue 3 requests but:
1046
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1048
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
981
1049
self.assertEqual(3, server.GET_request_nb)
982
1050
# Finally the client have tried a single range request and stays in
1037
1105
def setUp(self):
1038
1106
super(TestTruncatedBeforeBoundary, self).setUp()
1039
self.build_tree_contents([('a', b'0123456789')],)
1107
self.build_tree_contents([('a', '0123456789')],)
1041
1109
def test_readv_with_short_reads(self):
1042
1110
server = self.get_readonly_server()
1044
1112
# Force separate ranges for each offset
1045
1113
t._bytes_to_read_before_seek = 0
1046
1114
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1047
self.assertEqual((0, '0'), next(ireadv))
1048
self.assertEqual((2, '2'), next(ireadv))
1049
self.assertEqual((4, '45'), next(ireadv))
1050
self.assertEqual((9, '9'), next(ireadv))
1115
self.assertEqual((0, '0'), ireadv.next())
1116
self.assertEqual((2, '2'), ireadv.next())
1117
self.assertEqual((4, '45'), ireadv.next())
1118
self.assertEqual((9, '9'), ireadv.next())
1053
1121
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1095
1163
super(TestLimitedRangeRequestServer, self).setUp()
1096
1164
# We need to manipulate ranges that correspond to real chunks in the
1097
1165
# response, so we build a content appropriately.
1098
filler = b''.join([b'abcdefghij' for x in range(102)])
1099
content = b''.join([b'%04d' % v + filler for v in range(16)])
1166
filler = ''.join(['abcdefghij' for x in range(102)])
1167
content = ''.join(['%04d' % v + filler for v in range(16)])
1100
1168
self.build_tree_contents([('a', content)],)
1102
1170
def test_few_ranges(self):
1133
1201
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1134
1202
handler = _urllib2_wrappers.ProxyHandler()
1135
self.assertEqual(expected,
1203
self.assertEquals(expected,
1136
1204
handler.evaluate_proxy_bypass(host, no_proxy))
1138
1206
def test_empty_user(self):
1139
1207
self.overrideEnv('http_proxy', 'http://bar.com')
1140
1208
request = self._proxied_request()
1141
self.assertFalse('Proxy-authorization' in request.headers)
1209
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1143
1211
def test_user_with_at(self):
1144
1212
self.overrideEnv('http_proxy',
1145
1213
'http://username@domain:password@proxy_host:1234')
1146
1214
request = self._proxied_request()
1147
self.assertFalse('Proxy-authorization' in request.headers)
1215
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1149
1217
def test_invalid_proxy(self):
1150
1218
"""A proxy env variable without scheme"""
1151
1219
self.overrideEnv('http_proxy', 'host:1234')
1152
self.assertRaises(urlutils.InvalidURL, self._proxied_request)
1220
self.assertRaises(errors.InvalidURL, self._proxied_request)
1154
1222
def test_evaluate_proxy_bypass_true(self):
1155
1223
"""The host is not proxied"""
1192
1260
def setUp(self):
1193
1261
super(TestProxyHttpServer, self).setUp()
1194
1262
self.transport_secondary_server = http_utils.ProxyServer
1195
self.build_tree_contents([('foo', b'contents of foo\n'),
1196
('foo-proxied', b'proxied contents of foo\n')])
1263
self.build_tree_contents([('foo', 'contents of foo\n'),
1264
('foo-proxied', 'proxied contents of foo\n')])
1197
1265
# Let's setup some attributes for tests
1198
1266
server = self.get_readonly_server()
1199
1267
self.server_host_port = '%s:%d' % (server.host, server.port)
1200
self.no_proxy_host = self.server_host_port
1268
if self._testing_pycurl():
1269
# Oh my ! pycurl does not check for the port as part of
1270
# no_proxy :-( So we just test the host part
1271
self.no_proxy_host = server.host
1273
self.no_proxy_host = self.server_host_port
1201
1274
# The secondary server is the proxy
1202
1275
self.proxy_url = self.get_secondary_url()
1277
def _testing_pycurl(self):
1278
# TODO: This is duplicated for lots of the classes in this file
1279
return (features.pycurl.available()
1280
and self._transport == PyCurlTransport)
1204
1282
def assertProxied(self):
1205
1283
t = self.get_readonly_transport()
1206
1284
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1214
1292
self.assertProxied()
1216
1294
def test_HTTP_PROXY(self):
1295
if self._testing_pycurl():
1296
# pycurl does not check HTTP_PROXY for security reasons
1297
# (for use in a CGI context that we do not care
1298
# about. Should we ?)
1299
raise tests.TestNotApplicable(
1300
'pycurl does not check HTTP_PROXY for security reasons')
1217
1301
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1218
1302
self.assertProxied()
1231
1315
self.assertNotProxied()
1233
1317
def test_HTTP_PROXY_with_NO_PROXY(self):
1318
if self._testing_pycurl():
1319
raise tests.TestNotApplicable(
1320
'pycurl does not check HTTP_PROXY for security reasons')
1234
1321
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1235
1322
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1236
1323
self.assertNotProxied()
1248
1335
def test_http_proxy_without_scheme(self):
1249
1336
self.overrideEnv('http_proxy', self.server_host_port)
1250
self.assertRaises(urlutils.InvalidURL, self.assertProxied)
1337
if self._testing_pycurl():
1338
# pycurl *ignores* invalid proxy env variables. If that ever change
1339
# in the future, this test will fail indicating that pycurl do not
1340
# ignore anymore such variables.
1341
self.assertNotProxied()
1343
self.assertRaises(errors.InvalidURL, self.assertProxied)
1253
1346
class TestRanges(http_utils.TestCaseWithWebserver):
1261
1354
def setUp(self):
1262
1355
super(TestRanges, self).setUp()
1263
self.build_tree_contents([('a', b'0123456789')],)
1356
self.build_tree_contents([('a', '0123456789')],)
1265
1358
def create_transport_readonly_server(self):
1266
1359
return http_server.HttpServer(protocol_version=self._protocol_version)
1271
1364
coalesce = t._coalesce_offsets
1272
1365
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1273
1366
code, data = t._get(relpath, coalesced)
1274
self.assertTrue(code in (200, 206), '_get returns: %d' % code)
1367
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1275
1368
for start, end in ranges:
1276
1369
data.seek(start)
1277
1370
yield data.read(end - start + 1)
1279
1372
def _file_tail(self, relpath, tail_amount):
1280
1373
t = self.get_readonly_transport()
1281
1374
code, data = t._get(relpath, [], tail_amount)
1282
self.assertTrue(code in (200, 206), '_get returns: %d' % code)
1375
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1283
1376
data.seek(-tail_amount, 2)
1284
1377
return data.read(tail_amount)
1286
1379
def test_range_header(self):
1289
['0', '234'], list(self._file_contents('a', [(0, 0), (2, 4)])))
1381
map(self.assertEqual,['0', '234'],
1382
list(self._file_contents('a', [(0,0), (2,4)])),)
1291
1384
def test_range_header_tail(self):
1292
1385
self.assertEqual('789', self._file_tail('a', 3))
1311
1404
def setUp(self):
1312
1405
super(TestHTTPRedirections, self).setUp()
1313
self.build_tree_contents([('a', b'0123456789'),
1406
self.build_tree_contents([('a', '0123456789'),
1315
b'# Bazaar revision bundle v0.9\n#\n')
1408
'# Bazaar revision bundle v0.9\n#\n')
1318
1411
def test_redirected(self):
1379
1477
def setUp(self):
1478
if (features.pycurl.available()
1479
and self._transport == PyCurlTransport):
1480
raise tests.TestNotApplicable(
1481
"pycurl doesn't redirect silently anymore")
1380
1482
super(TestHTTPSilentRedirections, self).setUp()
1381
1483
install_redirected_request(self)
1382
1484
cleanup_http_redirection_connections(self)
1383
self.build_tree_contents([('a', b'a'),
1485
self.build_tree_contents([('a','a'),
1385
('1/a', b'redirected once'),
1487
('1/a', 'redirected once'),
1387
('2/a', b'redirected twice'),
1489
('2/a', 'redirected twice'),
1389
('3/a', b'redirected thrice'),
1491
('3/a', 'redirected thrice'),
1391
('4/a', b'redirected 4 times'),
1493
('4/a', 'redirected 4 times'),
1393
('5/a', b'redirected 5 times'),
1495
('5/a', 'redirected 5 times'),
1396
1498
def test_one_redirection(self):
1430
1532
def setUp(self):
1431
1533
super(TestDoCatchRedirections, self).setUp()
1432
self.build_tree_contents([('a', b'0123456789'),],)
1534
self.build_tree_contents([('a', '0123456789'),],)
1433
1535
cleanup_http_redirection_connections(self)
1435
1537
self.old_transport = self.get_old_transport()
1512
1614
def setUp(self):
1513
1615
super(TestAuth, self).setUp()
1514
1616
self.server = self.get_readonly_server()
1515
self.build_tree_contents([('a', b'contents of a\n'),
1516
('b', b'contents of b\n'),])
1617
self.build_tree_contents([('a', 'contents of a\n'),
1618
('b', 'contents of b\n'),])
1518
1620
def create_transport_readonly_server(self):
1519
1621
server = self._auth_server(protocol_version=self._protocol_version)
1520
1622
server._url_protocol = self._url_protocol
1625
def _testing_pycurl(self):
1626
# TODO: This is duplicated for lots of the classes in this file
1627
return (features.pycurl.available()
1628
and self._transport == PyCurlTransport)
1523
1630
def get_user_url(self, user, password):
1524
1631
"""Build an url embedding user and password"""
1525
1632
url = '%s://' % self.server._url_protocol
1575
1682
self.assertEqual(2, self.server.auth_required_errors)
1577
1684
def test_prompt_for_username(self):
1685
if self._testing_pycurl():
1686
raise tests.TestNotApplicable(
1687
'pycurl cannot prompt, it handles auth by embedding'
1688
' user:pass in urls only')
1578
1690
self.server.add_user('joe', 'foo')
1579
1691
t = self.get_user_transport(None, None)
1580
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
1581
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1582
self.assertEqual('contents of a\n', t.get('a').read())
1692
stdout = tests.StringIOWrapper()
1693
stderr = tests.StringIOWrapper()
1694
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1695
stdout=stdout, stderr=stderr)
1696
self.assertEqual('contents of a\n',t.get('a').read())
1583
1697
# stdin should be empty
1584
1698
self.assertEqual('', ui.ui_factory.stdin.readline())
1590
1704
stderr.readline())
1592
1706
def test_prompt_for_password(self):
1707
if self._testing_pycurl():
1708
raise tests.TestNotApplicable(
1709
'pycurl cannot prompt, it handles auth by embedding'
1710
' user:pass in urls only')
1593
1712
self.server.add_user('joe', 'foo')
1594
1713
t = self.get_user_transport('joe', None)
1595
ui.ui_factory = tests.TestUIFactory(stdin='foo\n')
1596
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1714
stdout = tests.StringIOWrapper()
1715
stderr = tests.StringIOWrapper()
1716
ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1717
stdout=stdout, stderr=stderr)
1597
1718
self.assertEqual('contents of a\n', t.get('a').read())
1598
1719
# stdin should be empty
1599
1720
self.assertEqual('', ui.ui_factory.stdin.readline())
1602
1723
self.assertEqual('', stdout.getvalue())
1603
1724
# And we shouldn't prompt again for a different request
1604
1725
# against the same transport.
1605
self.assertEqual('contents of b\n', t.get('b').read())
1726
self.assertEqual('contents of b\n',t.get('b').read())
1607
1728
# And neither against a clone
1608
self.assertEqual('contents of b\n', t2.get('b').read())
1729
self.assertEqual('contents of b\n',t2.get('b').read())
1609
1730
# Only one 'Authentication Required' error should occur
1610
1731
self.assertEqual(1, self.server.auth_required_errors)
1624
1745
self.server.auth_realm))
1626
1747
def test_no_prompt_for_password_when_using_auth_config(self):
1748
if self._testing_pycurl():
1749
raise tests.TestNotApplicable(
1750
'pycurl does not support authentication.conf'
1751
' since it cannot prompt')
1628
1754
password = 'foo'
1629
1755
stdin_content = 'bar\n' # Not the right password
1630
1756
self.server.add_user(user, password)
1631
1757
t = self.get_user_transport(user, None)
1632
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1758
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1759
stderr=tests.StringIOWrapper())
1633
1760
# Create a minimal config file with the right password
1634
1761
_setup_authentication_config(scheme='http', port=self.server.port,
1635
1762
user=user, password=password)
1636
1763
# Issue a request to the server to connect
1637
self.assertEqual('contents of a\n', t.get('a').read())
1764
self.assertEqual('contents of a\n',t.get('a').read())
1638
1765
# stdin should have been left untouched
1639
1766
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1640
1767
# Only one 'Authentication Required' error should occur
1644
1771
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1645
1772
http_utils.ProxyDigestAuthServer):
1646
1773
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1774
if self._testing_pycurl():
1776
'pycurl does not handle a nonce change')
1647
1777
self.server.add_user('joe', 'foo')
1648
1778
t = self.get_user_transport('joe', 'foo')
1649
1779
self.assertEqual('contents of a\n', t.get('a').read())
1659
1789
self.assertEqual(2, self.server.auth_required_errors)
1661
1791
def test_user_from_auth_conf(self):
1792
if self._testing_pycurl():
1793
raise tests.TestNotApplicable(
1794
'pycurl does not support authentication.conf')
1663
1796
password = 'foo'
1664
1797
self.server.add_user(user, password)
1671
1804
self.assertEqual(1, self.server.auth_required_errors)
1673
1806
def test_no_credential_leaks_in_log(self):
1674
self.overrideAttr(debug, 'debug_flags', {'http'})
1807
self.overrideAttr(debug, 'debug_flags', set(['http']))
1676
1809
password = 'very-sensitive-password'
1677
1810
self.server.add_user(user, password)
1713
1846
def setUp(self):
1714
1847
super(TestProxyAuth, self).setUp()
1715
1848
# Override the contents to avoid false positives
1716
self.build_tree_contents([('a', b'not proxied contents of a\n'),
1717
('b', b'not proxied contents of b\n'),
1718
('a-proxied', b'contents of a\n'),
1719
('b-proxied', b'contents of b\n'),
1849
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1850
('b', 'not proxied contents of b\n'),
1851
('a-proxied', 'contents of a\n'),
1852
('b-proxied', 'contents of b\n'),
1722
1855
def get_user_transport(self, user, password):
1723
proxy_url = self.get_user_url(user, password)
1724
self.overrideEnv('all_proxy', proxy_url)
1856
self.overrideEnv('all_proxy', self.get_user_url(user, password))
1725
1857
return TestAuth.get_user_transport(self, user, password)
1728
class NonClosingBytesIO(io.BytesIO):
1731
"""Ignore and leave file open."""
1859
def test_empty_pass(self):
1860
if self._testing_pycurl():
1862
if pycurl.version_info()[1] < '7.16.0':
1864
'pycurl < 7.16.0 does not handle empty proxy passwords')
1865
super(TestProxyAuth, self).test_empty_pass()
1734
1868
class SampleSocket(object):
1740
1874
:param socket_read_content: a byte sequence
1742
self.readfile = io.BytesIO(socket_read_content)
1743
self.writefile = NonClosingBytesIO()
1746
"""Ignore and leave files alone."""
1876
# Use plain python StringIO so we can monkey-patch the close method to
1877
# not discard the contents.
1878
from StringIO import StringIO
1879
self.readfile = StringIO(socket_read_content)
1880
self.writefile = StringIO()
1881
self.writefile.close = lambda: None
1882
self.close = lambda: None
1748
1884
def makefile(self, mode='r', bufsize=None):
1749
1885
if 'r' in mode:
1762
1898
def setUp(self):
1763
1899
super(SmartHTTPTunnellingTest, self).setUp()
1764
1900
# We use the VFS layer as part of HTTP tunnelling tests.
1765
self.overrideEnv('BRZ_NO_SMART_VFS', None)
1901
self.overrideEnv('BZR_NO_SMART_VFS', None)
1766
1902
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1767
1903
self.http_server = self.get_readonly_server()
1792
1928
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1794
1930
self.assertEqual(
1795
[(0, "c")], list(remote_transport.readv("data-file", [(0, 1)])))
1931
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1797
1933
def test_http_send_smart_request(self):
1864
2000
self.assertIsInstance(r, type(t))
1865
2001
# Both transports share the some connection
1866
2002
self.assertEqual(t._get_connection(), r._get_connection())
1867
self.assertEqual('http://www.example.com/foo/subdir/', r.base)
2003
self.assertEquals('http://www.example.com/foo/subdir/', r.base)
1869
2005
def test_redirected_to_self_with_slash(self):
1870
2006
t = self._transport('http://www.example.com/foo')
1881
2017
r = t._redirected_to('http://www.example.com/foo',
1882
2018
'http://foo.example.com/foo/subdir')
1883
2019
self.assertIsInstance(r, type(t))
1884
self.assertEqual('http://foo.example.com/foo/subdir/',
2020
self.assertEquals('http://foo.example.com/foo/subdir/',
1885
2021
r.external_url())
1887
2023
def test_redirected_to_same_host_sibling_protocol(self):
1889
2025
r = t._redirected_to('http://www.example.com/foo',
1890
2026
'https://www.example.com/foo')
1891
2027
self.assertIsInstance(r, type(t))
1892
self.assertEqual('https://www.example.com/foo/',
2028
self.assertEquals('https://www.example.com/foo/',
1893
2029
r.external_url())
1895
2031
def test_redirected_to_same_host_different_protocol(self):
1896
2032
t = self._transport('http://www.example.com/foo')
1897
2033
r = t._redirected_to('http://www.example.com/foo',
1898
2034
'ftp://www.example.com/foo')
1899
self.assertNotEqual(type(r), type(t))
1900
self.assertEqual('ftp://www.example.com/foo/', r.external_url())
2035
self.assertNotEquals(type(r), type(t))
2036
self.assertEquals('ftp://www.example.com/foo/', r.external_url())
1902
2038
def test_redirected_to_same_host_specific_implementation(self):
1903
2039
t = self._transport('http://www.example.com/foo')
1904
2040
r = t._redirected_to('http://www.example.com/foo',
1905
2041
'https+urllib://www.example.com/foo')
1906
self.assertEqual('https://www.example.com/foo/', r.external_url())
2042
self.assertEquals('https://www.example.com/foo/', r.external_url())
1908
2044
def test_redirected_to_different_host_same_user(self):
1909
2045
t = self._transport('http://joe@www.example.com/foo')
1911
2047
'https://foo.example.com/foo')
1912
2048
self.assertIsInstance(r, type(t))
1913
2049
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
1914
self.assertEqual('https://joe@foo.example.com/foo/', r.external_url())
2050
self.assertEquals('https://joe@foo.example.com/foo/', r.external_url())
1917
2053
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
2159
2295
_password_prompt_prefix = ''
2160
2296
_username_prompt_prefix = ''
2161
2297
_auth_server = http_utils.HTTPBasicAuthServer
2162
_transport = HttpTransport
2298
_transport = _urllib.HttpTransport_urllib
2164
2300
def setUp(self):
2165
2301
super(TestAuthOnRedirected, self).setUp()
2166
self.build_tree_contents([('a', b'a'),
2302
self.build_tree_contents([('a','a'),
2168
('1/a', b'redirected once'),
2304
('1/a', 'redirected once'),
2170
2306
new_prefix = 'http://%s:%s' % (self.new_server.host,
2171
2307
self.new_server.port)
2192
2328
self.addCleanup(redirected_t.disconnect)
2193
2329
return redirected_t
2195
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2331
stdout = tests.StringIOWrapper()
2332
stderr = tests.StringIOWrapper()
2333
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2334
stdout=stdout, stderr=stderr)
2196
2335
self.assertEqual('redirected once',
2197
2336
transport.do_catching_redirections(
2198
2337
self.get_a, self.old_transport, redirected).read())
2200
2339
# stdin should be empty
2201
2340
self.assertEqual('', ui.ui_factory.stdin.readline())
2202
2341
# stdout should be empty, stderr will contains the prompts
2203
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2342
self.assertEqual('', stdout.getvalue())
2205
2344
def test_auth_on_redirected_via_following_redirections(self):
2206
2345
self.new_server.add_user('joe', 'foo')
2207
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2346
stdout = tests.StringIOWrapper()
2347
stderr = tests.StringIOWrapper()
2348
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2349
stdout=stdout, stderr=stderr)
2208
2350
t = self.old_transport
2209
2351
req = RedirectedRequest('GET', t.abspath('a'))
2210
2352
new_prefix = 'http://%s:%s' % (self.new_server.host,
added
removed
bzrlib/tests/test_http.py
