23
23
# TODO: Should be renamed to breezy.transport.http.tests?
24
24
# TODO: What about renaming to breezy.tests.transport.http ?
26
from http.client import UnknownProtocol, parse_headers
27
from http.server import SimpleHTTPRequestHandler
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
71
76
def vary_by_http_client_implementation():
72
77
"""Test the libraries we can use, currently just urllib."""
73
78
transport_scenarios = [
74
('urllib', dict(_transport=HttpTransport,
75
_server=http_server.HttpServer,
76
_url_protocol='http',)),
79
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
80
_server=http_server.HttpServer_urllib,
81
_url_protocol='http+urllib',)),
78
83
return transport_scenarios
81
86
def vary_by_http_protocol_version():
82
87
"""Test on http/1.0 and 1.1"""
84
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
85
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
89
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
90
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
119
124
def vary_by_http_activity():
120
125
activity_scenarios = [
121
126
('urllib,http', dict(_activity_server=ActivityHTTPServer,
122
_transport=HttpTransport,)),
127
_transport=_urllib.HttpTransport_urllib,)),
124
129
if features.HTTPSServerFeature.available():
125
130
# FIXME: Until we have a better way to handle self-signed certificates
133
class HTTPS_transport(HttpTransport):
137
class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
135
139
def __init__(self, base, _from_transport=None):
136
super(HTTPS_transport, self).__init__(
140
super(HTTPS_urllib_transport, self).__init__(
137
141
base, _from_transport=_from_transport,
138
142
ca_certs=ssl_certs.build_path('ca.crt'))
140
144
activity_scenarios.append(
141
145
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
142
_transport=HTTPS_transport,)),)
146
_transport=HTTPS_urllib_transport,)),)
143
147
return activity_scenarios
221
225
def parse_header(self, header, auth_handler_class=None):
222
226
if auth_handler_class is None:
223
auth_handler_class = http.AbstractAuthHandler
224
self.auth_handler = auth_handler_class()
227
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
228
self.auth_handler = auth_handler_class()
225
229
return self.auth_handler._parse_auth_header(header)
227
231
def test_empty_header(self):
253
257
def test_basic_extract_realm(self):
254
258
scheme, remainder = self.parse_header(
255
259
'Basic realm="Thou should not pass"',
256
http.BasicAuthHandler)
260
_urllib2_wrappers.BasicAuthHandler)
257
261
match, realm = self.auth_handler.extract_realm(remainder)
258
262
self.assertTrue(match is not None)
259
self.assertEqual(u'Thou should not pass', realm)
263
self.assertEqual('Thou should not pass', realm)
261
265
def test_digest_header(self):
262
266
scheme, remainder = self.parse_header(
271
275
super(TestHTTPRangeParsing, self).setUp()
272
276
# We focus on range parsing here and ignore everything else
274
277
class RequestHandler(http_server.TestingHTTPRequestHandler):
275
278
def setup(self): pass
277
279
def handle(self): pass
279
280
def finish(self): pass
281
282
self.req_handler = RequestHandler(None, None, None)
283
284
def assertRanges(self, ranges, header, file_size):
284
285
self.assertEqual(ranges,
285
self.req_handler._parse_ranges(header, file_size))
286
self.req_handler._parse_ranges(header, file_size))
287
288
def test_simple_range(self):
288
289
self.assertRanges([(0, 2)], 'bytes=0-2', 12)
426
427
self.assertEqual(t.has('foo/bar'), True)
427
428
self.assertEqual(len(server.logs), 1)
428
429
self.assertContainsRe(server.logs[0],
429
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "Breezy/')
430
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "Breezy/')
431
432
def test_http_has_not_found(self):
432
433
server = self.get_readonly_server()
433
434
t = self.get_readonly_transport()
434
435
self.assertEqual(t.has('not-found'), False)
435
436
self.assertContainsRe(server.logs[1],
436
r'"HEAD /not-found HTTP/1.." 404 - "-" "Breezy/')
437
r'"HEAD /not-found HTTP/1.." 404 - "-" "Breezy/')
438
439
def test_http_get(self):
439
440
server = self.get_readonly_server()
484
485
def test_post_body_is_received(self):
485
server = RecordingServer(expect_body_tail=b'end-of-body',
486
server = RecordingServer(expect_body_tail='end-of-body',
486
487
scheme=self._url_protocol)
487
488
self.start_server(server)
488
489
url = server.get_url()
489
490
# FIXME: needs a cleanup -- vila 20100611
490
491
http_transport = transport.get_transport_from_url(url)
491
code, response = http_transport._post(b'abc def end-of-body')
493
server.received_bytes.startswith(b'POST /.bzr/smart HTTP/1.'))
495
b'content-length: 19\r' in server.received_bytes.lower())
496
self.assertTrue(b'content-type: application/octet-stream\r'
492
code, response = http_transport._post('abc def end-of-body')
494
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
495
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
496
self.assertTrue('content-type: application/octet-stream\r'
497
497
in server.received_bytes.lower())
498
498
# The transport should not be assuming that the server can accept
499
499
# chunked encoding the first time it connects, because HTTP/1.1, so we
500
500
# check for the literal string.
502
server.received_bytes.endswith(b'\r\n\r\nabc def end-of-body'))
502
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
505
505
class TestRangeHeader(tests.TestCase):
506
506
"""Test range_header method"""
508
508
def check_header(self, value, ranges=[], tail=0):
509
offsets = [(start, end - start + 1) for start, end in ranges]
509
offsets = [ (start, end - start + 1) for start, end in ranges]
510
510
coalesce = transport.Transport._coalesce_offsets
511
511
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
512
range_header = http.HttpTransport._range_header
512
range_header = http.HttpTransportBase._range_header
513
513
self.assertEqual(value, range_header(coalesced, tail))
515
515
def test_range_header_single(self):
628
628
def parse_request(self):
629
629
"""Fakes handling a single HTTP request, returns a bad status"""
630
630
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
631
self.wfile.write(b"Invalid status line\r\n")
631
self.wfile.write("Invalid status line\r\n")
632
632
# If we don't close the connection pycurl will hang. Since this is a
633
633
# stress test we don't *have* to respect the protocol, but we don't
634
634
# have to sabotage it too much either.
653
653
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
654
654
# Returns an invalid protocol version, but curl just
655
655
# ignores it and those cannot be tested.
656
self.wfile.write(b"%s %d %s\r\n" % (
657
b'HTTP/0.0', 404, b'Look at my protocol version'))
656
self.wfile.write("%s %d %s\r\n" % ('HTTP/0.0',
658
'Look at my protocol version'))
716
717
self.assertEqual(None, server.port)
718
719
def test_send_receive_bytes(self):
719
server = RecordingServer(expect_body_tail=b'c', scheme='http')
720
server = RecordingServer(expect_body_tail='c', scheme='http')
720
721
self.start_server(server)
721
722
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
722
723
sock.connect((server.host, server.port))
724
self.assertEqual(b'HTTP/1.1 200 OK\r\n',
725
self.assertEqual('HTTP/1.1 200 OK\r\n',
725
726
osutils.recv_all(sock, 4096))
726
self.assertEqual(b'abc', server.received_bytes)
727
self.assertEqual('abc', server.received_bytes)
729
730
class TestRangeRequestServer(TestSpecificRequestHandler):
736
737
super(TestRangeRequestServer, self).setUp()
737
self.build_tree_contents([('a', b'0123456789')],)
738
self.build_tree_contents([('a', '0123456789')],)
739
740
def test_readv(self):
740
741
t = self.get_readonly_transport()
741
742
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
742
self.assertEqual(l[0], (0, b'0'))
743
self.assertEqual(l[1], (1, b'1'))
744
self.assertEqual(l[2], (3, b'34'))
745
self.assertEqual(l[3], (9, b'9'))
743
self.assertEqual(l[0], (0, '0'))
744
self.assertEqual(l[1], (1, '1'))
745
self.assertEqual(l[2], (3, '34'))
746
self.assertEqual(l[3], (9, '9'))
747
748
def test_readv_out_of_order(self):
748
749
t = self.get_readonly_transport()
749
750
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
750
self.assertEqual(l[0], (1, b'1'))
751
self.assertEqual(l[1], (9, b'9'))
752
self.assertEqual(l[2], (0, b'0'))
753
self.assertEqual(l[3], (3, b'34'))
751
self.assertEqual(l[0], (1, '1'))
752
self.assertEqual(l[1], (9, '9'))
753
self.assertEqual(l[2], (0, '0'))
754
self.assertEqual(l[3], (3, '34'))
755
756
def test_readv_invalid_ranges(self):
756
757
t = self.get_readonly_transport()
772
773
t._max_readv_combine = 1
773
774
t._max_get_ranges = 1
774
775
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
775
self.assertEqual(l[0], (0, b'0'))
776
self.assertEqual(l[1], (1, b'1'))
777
self.assertEqual(l[2], (3, b'34'))
778
self.assertEqual(l[3], (9, b'9'))
776
self.assertEqual(l[0], (0, '0'))
777
self.assertEqual(l[1], (1, '1'))
778
self.assertEqual(l[2], (3, '34'))
779
self.assertEqual(l[3], (9, '9'))
779
780
# The server should have issued 4 requests
780
781
self.assertEqual(4, server.GET_request_nb)
787
788
# single range will keep its size even if bigger than the limit.
788
789
t._get_max_size = 2
789
790
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
790
self.assertEqual(l[0], (0, b'0'))
791
self.assertEqual(l[1], (1, b'1'))
792
self.assertEqual(l[2], (2, b'2345'))
793
self.assertEqual(l[3], (6, b'6789'))
791
self.assertEqual(l[0], (0, '0'))
792
self.assertEqual(l[1], (1, '1'))
793
self.assertEqual(l[2], (2, '2345'))
794
self.assertEqual(l[3], (6, '6789'))
794
795
# The server should have issued 3 requests
795
796
self.assertEqual(3, server.GET_request_nb)
802
803
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
803
804
# The server should have issued 3 requests
804
805
self.assertEqual(3, server.GET_request_nb)
805
self.assertEqual(b'0123456789', t.get_bytes('a'))
806
self.assertEqual('0123456789', t.get_bytes('a'))
806
807
self.assertEqual(4, server.GET_request_nb)
808
809
def test_incomplete_readv_leave_pipe_clean(self):
813
814
# Don't collapse readv results into a list so that we leave unread
814
815
# bytes on the socket
815
816
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
816
self.assertEqual((0, b'0'), next(ireadv))
817
self.assertEqual((0, '0'), next(ireadv))
817
818
# The server should have issued one request so far
818
819
self.assertEqual(1, server.GET_request_nb)
819
self.assertEqual(b'0123456789', t.get_bytes('a'))
820
self.assertEqual('0123456789', t.get_bytes('a'))
820
821
# get_bytes issued an additional request, the readv pending ones are
822
823
self.assertEqual(2, server.GET_request_nb)
889
890
"multipart/byteranges; boundary=%s" % boundary)
890
891
self.end_headers()
891
892
for (start, end) in ranges:
892
self.wfile.write(b"--%s\r\n" % boundary.encode('ascii'))
893
self.wfile.write("--%s\r\n" % boundary)
893
894
self.send_header("Content-type", 'application/octet-stream')
894
895
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
931
932
'Content-type', 'application/octet-stream')
932
933
content_length += self._header_line_length(
933
934
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
934
content_length += len('\r\n') # end headers
935
content_length += end - start # + 1
935
content_length += len('\r\n') # end headers
936
content_length += end - start # + 1
936
937
content_length += len(boundary_line)
937
938
self.send_header('Content-length', content_length)
938
939
self.end_headers()
969
970
# Force separate ranges for each offset
970
971
t._bytes_to_read_before_seek = 0
971
972
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
972
self.assertEqual((0, b'0'), next(ireadv))
973
self.assertEqual((2, b'2'), next(ireadv))
973
self.assertEqual((0, '0'), next(ireadv))
974
self.assertEqual((2, '2'), next(ireadv))
974
975
# Only one request have been issued so far
975
976
self.assertEqual(1, server.GET_request_nb)
976
self.assertEqual((4, b'45'), next(ireadv))
977
self.assertEqual((9, b'9'), next(ireadv))
977
self.assertEqual((4, '45'), next(ireadv))
978
self.assertEqual((9, '9'), next(ireadv))
978
979
# We issue 3 requests: two multiple (4 ranges, then 2 ranges) then a
980
981
self.assertEqual(3, server.GET_request_nb)
1004
1005
'Content-type', 'application/octet-stream')
1005
1006
content_length += self._header_line_length(
1006
1007
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
1007
content_length += len('\r\n') # end headers
1008
content_length += end - start # + 1
1008
content_length += len('\r\n') # end headers
1009
content_length += end - start # + 1
1009
1010
content_length += len(boundary_line)
1010
1011
self.send_header('Content-length', content_length)
1011
1012
self.end_headers()
1043
1044
# Force separate ranges for each offset
1044
1045
t._bytes_to_read_before_seek = 0
1045
1046
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1046
self.assertEqual((0, b'0'), next(ireadv))
1047
self.assertEqual((2, b'2'), next(ireadv))
1048
self.assertEqual((4, b'45'), next(ireadv))
1049
self.assertEqual((9, b'9'), next(ireadv))
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))
1052
1053
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1094
1095
super(TestLimitedRangeRequestServer, self).setUp()
1095
1096
# We need to manipulate ranges that correspond to real chunks in the
1096
1097
# response, so we build a content appropriately.
1097
filler = b''.join([b'abcdefghij' for x in range(102)])
1098
content = b''.join([b'%04d' % v + filler for v in range(16)])
1098
filler = ''.join(['abcdefghij' for x in range(102)])
1099
content = ''.join(['%04d' % v + filler for v in range(16)])
1099
1100
self.build_tree_contents([('a', content)],)
1101
1102
def test_few_ranges(self):
1102
1103
t = self.get_readonly_transport()
1103
1104
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1104
self.assertEqual(l[0], (0, b'0000'))
1105
self.assertEqual(l[1], (1024, b'0001'))
1105
self.assertEqual(l[0], (0, '0000'))
1106
self.assertEqual(l[1], (1024, '0001'))
1106
1107
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1108
1109
def test_more_ranges(self):
1109
1110
t = self.get_readonly_transport()
1110
1111
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1111
self.assertEqual(l[0], (0, b'0000'))
1112
self.assertEqual(l[1], (1024, b'0001'))
1113
self.assertEqual(l[2], (4096, b'0004'))
1114
self.assertEqual(l[3], (8192, b'0008'))
1112
self.assertEqual(l[0], (0, '0000'))
1113
self.assertEqual(l[1], (1024, '0001'))
1114
self.assertEqual(l[2], (4096, '0004'))
1115
self.assertEqual(l[3], (8192, '0008'))
1115
1116
# The server will refuse to serve the first request (too much ranges),
1116
1117
# a second request will succeed.
1117
1118
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1126
1127
def _proxied_request(self):
1127
handler = http.ProxyHandler()
1128
request = http.Request('GET', 'http://baz/buzzle')
1128
handler = _urllib2_wrappers.ProxyHandler()
1129
request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
1129
1130
handler.set_proxy(request, 'http')
1132
1133
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1133
handler = http.ProxyHandler()
1134
handler = _urllib2_wrappers.ProxyHandler()
1134
1135
self.assertEqual(expected,
1135
handler.evaluate_proxy_bypass(host, no_proxy))
1136
handler.evaluate_proxy_bypass(host, no_proxy))
1137
1138
def test_empty_user(self):
1138
1139
self.overrideEnv('http_proxy', 'http://bar.com')
1191
1192
def setUp(self):
1192
1193
super(TestProxyHttpServer, self).setUp()
1193
1194
self.transport_secondary_server = http_utils.ProxyServer
1194
self.build_tree_contents([('foo', b'contents of foo\n'),
1195
('foo-proxied', b'proxied contents of foo\n')])
1195
self.build_tree_contents([('foo', 'contents of foo\n'),
1196
('foo-proxied', 'proxied contents of foo\n')])
1196
1197
# Let's setup some attributes for tests
1197
1198
server = self.get_readonly_server()
1198
1199
self.server_host_port = '%s:%d' % (server.host, server.port)
1203
1204
def assertProxied(self):
1204
1205
t = self.get_readonly_transport()
1205
self.assertEqual(b'proxied contents of foo\n', t.get('foo').read())
1206
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1207
1208
def assertNotProxied(self):
1208
1209
t = self.get_readonly_transport()
1209
self.assertEqual(b'contents of foo\n', t.get('foo').read())
1210
self.assertEqual('contents of foo\n', t.get('foo').read())
1211
1212
def test_http_proxy(self):
1212
1213
self.overrideEnv('http_proxy', self.proxy_url)
1260
1261
def setUp(self):
1261
1262
super(TestRanges, self).setUp()
1262
self.build_tree_contents([('a', b'0123456789')],)
1263
self.build_tree_contents([('a', '0123456789')],)
1264
1265
def create_transport_readonly_server(self):
1265
1266
return http_server.HttpServer(protocol_version=self._protocol_version)
1267
1268
def _file_contents(self, relpath, ranges):
1268
1269
t = self.get_readonly_transport()
1269
offsets = [(start, end - start + 1) for start, end in ranges]
1270
offsets = [ (start, end - start + 1) for start, end in ranges]
1270
1271
coalesce = t._coalesce_offsets
1271
1272
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1272
1273
code, data = t._get(relpath, coalesced)
1285
1286
def test_range_header(self):
1287
1288
self.assertEqual(
1288
[b'0', b'234'], list(self._file_contents('a', [(0, 0), (2, 4)])))
1289
['0', '234'], list(self._file_contents('a', [(0, 0), (2, 4)])))
1290
1291
def test_range_header_tail(self):
1291
self.assertEqual(b'789', self._file_tail('a', 3))
1292
self.assertEqual('789', self._file_tail('a', 3))
1293
1294
def test_syntactically_invalid_range_header(self):
1294
1295
self.assertListRaises(errors.InvalidHttpRange,
1295
self._file_contents, 'a', [(4, 3)])
1296
self._file_contents, 'a', [(4, 3)])
1297
1298
def test_semantically_invalid_range_header(self):
1298
1299
self.assertListRaises(errors.InvalidHttpRange,
1299
self._file_contents, 'a', [(42, 128)])
1300
self._file_contents, 'a', [(42, 128)])
1302
1303
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1310
1311
def setUp(self):
1311
1312
super(TestHTTPRedirections, self).setUp()
1312
self.build_tree_contents([('a', b'0123456789'),
1313
self.build_tree_contents([('a', '0123456789'),
1314
b'# Bazaar revision bundle v0.9\n#\n')
1315
'# Bazaar revision bundle v0.9\n#\n')
1317
1318
def test_redirected(self):
1318
1319
self.assertRaises(errors.RedirectRequested,
1319
1320
self.get_old_transport().get, 'a')
1322
self.get_new_transport().get('a').read())
1325
class RedirectedRequest(http.Request):
1321
self.assertEqual('0123456789', self.get_new_transport().get('a').read())
1324
class RedirectedRequest(_urllib2_wrappers.Request):
1326
1325
"""Request following redirections. """
1328
init_orig = http.Request.__init__
1327
init_orig = _urllib2_wrappers.Request.__init__
1330
1329
def __init__(self, method, url, *args, **kwargs):
1331
1330
"""Constructor.
1334
1333
# Since the tests using this class will replace
1335
# http.Request, we can't just call the base class __init__
1334
# _urllib2_wrappers.Request, we can't just call the base class __init__
1336
1335
# or we'll loop.
1337
1336
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1338
1337
self.follow_redirections = True
1341
1340
def install_redirected_request(test):
1342
test.overrideAttr(http, 'Request', RedirectedRequest)
1341
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1345
1344
def cleanup_http_redirection_connections(test):
1346
1345
# Some sockets are opened but never seen by _urllib, so we trap them at
1347
# the http level to be able to clean them up.
1346
# the _urllib2_wrappers level to be able to clean them up.
1348
1347
def socket_disconnect(sock):
1350
1349
sock.shutdown(socket.SHUT_RDWR)
1352
1351
except socket.error:
1355
1353
def connect(connection):
1356
1354
test.http_connect_orig(connection)
1357
1355
test.addCleanup(socket_disconnect, connection.sock)
1358
1356
test.http_connect_orig = test.overrideAttr(
1359
http.HTTPConnection, 'connect', connect)
1357
_urllib2_wrappers.HTTPConnection, 'connect', connect)
1361
1358
def connect(connection):
1362
1359
test.https_connect_orig(connection)
1363
1360
test.addCleanup(socket_disconnect, connection.sock)
1364
1361
test.https_connect_orig = test.overrideAttr(
1365
http.HTTPSConnection, 'connect', connect)
1362
_urllib2_wrappers.HTTPSConnection, 'connect', connect)
1368
1365
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1383
1380
super(TestHTTPSilentRedirections, self).setUp()
1384
1381
install_redirected_request(self)
1385
1382
cleanup_http_redirection_connections(self)
1386
self.build_tree_contents([('a', b'a'),
1383
self.build_tree_contents([('a', 'a'),
1388
('1/a', b'redirected once'),
1385
('1/a', 'redirected once'),
1390
('2/a', b'redirected twice'),
1387
('2/a', 'redirected twice'),
1392
('3/a', b'redirected thrice'),
1389
('3/a', 'redirected thrice'),
1394
('4/a', b'redirected 4 times'),
1391
('4/a', 'redirected 4 times'),
1396
('5/a', b'redirected 5 times'),
1393
('5/a', 'redirected 5 times'),
1399
1396
def test_one_redirection(self):
1400
1397
t = self.get_old_transport()
1398
req = RedirectedRequest('GET', t._remote_path('a'))
1401
1399
new_prefix = 'http://%s:%s' % (self.new_server.host,
1402
1400
self.new_server.port)
1403
1401
self.old_server.redirections = \
1404
[('(.*)', r'%s/1\1' % (new_prefix), 301), ]
1407
t.request('GET', t._remote_path('a'), retries=1).read())
1402
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1403
self.assertEqual('redirected once', t._perform(req).read())
1409
1405
def test_five_redirections(self):
1410
1406
t = self.get_old_transport()
1407
req = RedirectedRequest('GET', t._remote_path('a'))
1411
1408
old_prefix = 'http://%s:%s' % (self.old_server.host,
1412
1409
self.old_server.port)
1413
1410
new_prefix = 'http://%s:%s' % (self.new_server.host,
1419
1416
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1420
1417
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1423
b'redirected 5 times',
1424
t.request('GET', t._remote_path('a'), retries=6).read())
1419
self.assertEqual('redirected 5 times', t._perform(req).read())
1427
1422
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1435
1430
def setUp(self):
1436
1431
super(TestDoCatchRedirections, self).setUp()
1437
self.build_tree_contents([('a', b'0123456789'), ],)
1432
self.build_tree_contents([('a', '0123456789'),],)
1438
1433
cleanup_http_redirection_connections(self)
1440
1435
self.old_transport = self.get_old_transport()
1446
1441
t = self.get_new_transport()
1448
1443
# We use None for redirected so that we fail if redirected
1449
self.assertEqual(b'0123456789',
1444
self.assertEqual('0123456789',
1450
1445
transport.do_catching_redirections(
1451
self.get_a, t, None).read())
1446
self.get_a, t, None).read())
1453
1448
def test_one_redirection(self):
1454
1449
self.redirections = 0
1458
1453
redirected_t = t._redirected_to(exception.source, exception.target)
1459
1454
return redirected_t
1461
self.assertEqual(b'0123456789',
1456
self.assertEqual('0123456789',
1462
1457
transport.do_catching_redirections(
1463
self.get_a, self.old_transport, redirected).read())
1458
self.get_a, self.old_transport, redirected).read())
1464
1459
self.assertEqual(1, self.redirections)
1466
1461
def test_redirection_loop(self):
1494
1489
password = 'foo'
1495
1490
_setup_authentication_config(scheme='http', host='localhost',
1496
1491
user=user, password=password)
1497
handler = http.HTTPAuthHandler()
1492
handler = _urllib2_wrappers.HTTPAuthHandler()
1498
1493
got_pass = handler.get_user_password(dict(
1500
1495
protocol='http',
1501
1496
host='localhost',
1505
1500
self.assertEqual((user, password), got_pass)
1517
1512
def setUp(self):
1518
1513
super(TestAuth, self).setUp()
1519
1514
self.server = self.get_readonly_server()
1520
self.build_tree_contents([('a', b'contents of a\n'),
1521
('b', b'contents of b\n'), ])
1515
self.build_tree_contents([('a', 'contents of a\n'),
1516
('b', 'contents of b\n'),])
1523
1518
def create_transport_readonly_server(self):
1524
1519
server = self._auth_server(protocol_version=self._protocol_version)
1551
1546
def test_empty_pass(self):
1552
1547
self.server.add_user('joe', '')
1553
1548
t = self.get_user_transport('joe', '')
1554
self.assertEqual(b'contents of a\n', t.get('a').read())
1549
self.assertEqual('contents of a\n', t.get('a').read())
1555
1550
# Only one 'Authentication Required' error should occur
1556
1551
self.assertEqual(1, self.server.auth_required_errors)
1558
1553
def test_user_pass(self):
1559
1554
self.server.add_user('joe', 'foo')
1560
1555
t = self.get_user_transport('joe', 'foo')
1561
self.assertEqual(b'contents of a\n', t.get('a').read())
1556
self.assertEqual('contents of a\n', t.get('a').read())
1562
1557
# Only one 'Authentication Required' error should occur
1563
1558
self.assertEqual(1, self.server.auth_required_errors)
1584
1579
t = self.get_user_transport(None, None)
1585
1580
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
1586
1581
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1587
self.assertEqual(b'contents of a\n', t.get('a').read())
1582
self.assertEqual('contents of a\n', t.get('a').read())
1588
1583
# stdin should be empty
1589
1584
self.assertEqual('', ui.ui_factory.stdin.readline())
1599
1594
t = self.get_user_transport('joe', None)
1600
1595
ui.ui_factory = tests.TestUIFactory(stdin='foo\n')
1601
1596
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1602
self.assertEqual(b'contents of a\n', t.get('a').read())
1597
self.assertEqual('contents of a\n', t.get('a').read())
1603
1598
# stdin should be empty
1604
1599
self.assertEqual('', ui.ui_factory.stdin.readline())
1605
1600
self._check_password_prompt(t._unqualified_scheme, 'joe',
1607
1602
self.assertEqual('', stdout.getvalue())
1608
1603
# And we shouldn't prompt again for a different request
1609
1604
# against the same transport.
1610
self.assertEqual(b'contents of b\n', t.get('b').read())
1605
self.assertEqual('contents of b\n', t.get('b').read())
1612
1607
# And neither against a clone
1613
self.assertEqual(b'contents of b\n', t2.get('b').read())
1608
self.assertEqual('contents of b\n', t2.get('b').read())
1614
1609
# Only one 'Authentication Required' error should occur
1615
1610
self.assertEqual(1, self.server.auth_required_errors)
1625
1620
def _expected_username_prompt(self, scheme):
1626
1621
return (self._username_prompt_prefix
1627
1622
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1628
self.server.host, self.server.port,
1629
self.server.auth_realm))
1623
self.server.host, self.server.port,
1624
self.server.auth_realm))
1631
1626
def test_no_prompt_for_password_when_using_auth_config(self):
1633
1628
password = 'foo'
1634
1629
stdin_content = 'bar\n' # Not the right password
1635
1630
self.server.add_user(user, password)
1639
1634
_setup_authentication_config(scheme='http', port=self.server.port,
1640
1635
user=user, password=password)
1641
1636
# Issue a request to the server to connect
1642
with t.get('a') as f:
1643
self.assertEqual(b'contents of a\n', f.read())
1637
self.assertEqual('contents of a\n', t.get('a').read())
1644
1638
# stdin should have been left untouched
1645
1639
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1646
1640
# Only one 'Authentication Required' error should occur
1652
1646
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1653
1647
self.server.add_user('joe', 'foo')
1654
1648
t = self.get_user_transport('joe', 'foo')
1655
with t.get('a') as f:
1656
self.assertEqual(b'contents of a\n', f.read())
1657
with t.get('b') as f:
1658
self.assertEqual(b'contents of b\n', f.read())
1649
self.assertEqual('contents of a\n', t.get('a').read())
1650
self.assertEqual('contents of b\n', t.get('b').read())
1659
1651
# Only one 'Authentication Required' error should have
1660
1652
# occured so far
1661
1653
self.assertEqual(1, self.server.auth_required_errors)
1662
1654
# The server invalidates the current nonce
1663
1655
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1664
self.assertEqual(b'contents of a\n', t.get('a').read())
1656
self.assertEqual('contents of a\n', t.get('a').read())
1665
1657
# Two 'Authentication Required' errors should occur (the
1666
1658
# initial 'who are you' and a second 'who are you' with the new nonce)
1667
1659
self.assertEqual(2, self.server.auth_required_errors)
1674
1666
user=user, password=password)
1675
1667
t = self.get_user_transport(None, None)
1676
1668
# Issue a request to the server to connect
1677
with t.get('a') as f:
1678
self.assertEqual(b'contents of a\n', f.read())
1669
self.assertEqual('contents of a\n', t.get('a').read())
1679
1670
# Only one 'Authentication Required' error should occur
1680
1671
self.assertEqual(1, self.server.auth_required_errors)
1723
1713
def setUp(self):
1724
1714
super(TestProxyAuth, self).setUp()
1725
1715
# Override the contents to avoid false positives
1726
self.build_tree_contents([('a', b'not proxied contents of a\n'),
1727
('b', b'not proxied contents of b\n'),
1728
('a-proxied', b'contents of a\n'),
1729
('b-proxied', b'contents of b\n'),
1716
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1717
('b', 'not proxied contents of b\n'),
1718
('a-proxied', 'contents of a\n'),
1719
('b-proxied', 'contents of b\n'),
1732
1722
def get_user_transport(self, user, password):
1805
1792
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1807
1794
self.assertEqual(
1808
[(0, b"c")], list(remote_transport.readv("data-file", [(0, 1)])))
1795
[(0, "c")], list(remote_transport.readv("data-file", [(0, 1)])))
1810
1797
def test_http_send_smart_request(self):
1812
post_body = b'hello\n'
1813
expected_reply_body = b'ok\x012\n'
1799
post_body = 'hello\n'
1800
expected_reply_body = 'ok\x012\n'
1815
1802
http_transport = transport.get_transport_from_url(
1816
1803
self.http_server.get_url())
1823
1810
httpd = self.http_server.server
1825
1812
socket = SampleSocket(
1826
b'POST /.bzr/smart %s \r\n' % self._protocol_version.encode('ascii') +
1813
'POST /.bzr/smart %s \r\n' % self._protocol_version
1827
1814
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1829
b'Content-Length: 6\r\n'
1816
+ 'Content-Length: 6\r\n'
1832
1819
# Beware: the ('localhost', 80) below is the
1833
1820
# client_address parameter, but we don't have one because
1834
1821
# we have defined a socket which is not bound to an
1838
1825
('localhost', 80),
1840
1827
response = socket.writefile.getvalue()
1841
self.assertStartsWith(
1843
b'%s 200 ' % self._protocol_version.encode('ascii'))
1828
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1844
1829
# This includes the end of the HTTP headers, and all the body.
1845
expected_end_of_response = b'\r\n\r\nok\x012\n'
1830
expected_end_of_response = '\r\n\r\nok\x012\n'
1846
1831
self.assertEndsWith(response, expected_end_of_response)
1905
1890
'https://www.example.com/foo')
1906
1891
self.assertIsInstance(r, type(t))
1907
1892
self.assertEqual('https://www.example.com/foo/',
1910
1895
def test_redirected_to_same_host_different_protocol(self):
1911
1896
t = self._transport('http://www.example.com/foo')
1912
1897
r = t._redirected_to('http://www.example.com/foo',
1913
'bzr://www.example.com/foo')
1898
'ftp://www.example.com/foo')
1914
1899
self.assertNotEqual(type(r), type(t))
1915
self.assertEqual('bzr://www.example.com/foo/', r.external_url())
1900
self.assertEqual('ftp://www.example.com/foo/', r.external_url())
1917
1902
def test_redirected_to_same_host_specific_implementation(self):
1918
1903
t = self._transport('http://www.example.com/foo')
1944
1929
def _handle_one_request(self):
1945
1930
tcs = self.server.test_case_server
1946
1931
requestline = self.rfile.readline()
1947
headers = parse_headers(self.rfile)
1948
bytes_read = len(headers.as_bytes())
1949
bytes_read += headers.as_bytes().count(b'\n')
1950
bytes_read += len(requestline)
1951
if requestline.startswith(b'POST'):
1932
headers = self.MessageClass(self.rfile, 0)
1933
# We just read: the request, the headers, an empty line indicating the
1934
# end of the headers.
1935
bytes_read = len(requestline)
1936
for line in headers.headers:
1937
bytes_read += len(line)
1938
bytes_read += len('\r\n')
1939
if requestline.startswith('POST'):
1952
1940
# The body should be a single line (or we don't know where it ends
1953
1941
# and we don't want to issue a blocking read)
1954
1942
body = self.rfile.readline()
2000
1987
self.server = self._activity_server(self._protocol_version)
2001
1988
self.server.start_server()
2002
1989
self.addCleanup(self.server.stop_server)
2003
_activities = {} # Don't close over self and create a cycle
1990
_activities = {} # Don't close over self and create a cycle
2005
1991
def report_activity(t, bytes, direction):
2006
1992
count = _activities.get(direction, 0)
2037
2023
Bazaar-NG meta directory, format 1
2039
2025
t = self.get_transport()
2040
self.assertEqual(b'Bazaar-NG meta directory, format 1\n',
2026
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2041
2027
t.get('foo/bar').read())
2042
2028
self.assertActivitiesMatch()
2044
2030
def test_has(self):
2045
self.server.canned_response = b'''HTTP/1.1 200 OK\r
2031
self.server.canned_response = '''HTTP/1.1 200 OK\r
2046
2032
Server: SimpleHTTP/0.6 Python/2.5.2\r
2047
2033
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2048
2034
Content-type: application/octet-stream\r
2108
2094
# Remember that the request is ignored and that the ranges below
2109
2095
# doesn't have to match the canned response.
2110
2096
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2111
# Force consumption of the last bytesrange boundary
2112
t._get_connection().cleanup_pipe()
2113
2097
self.assertEqual(2, len(l))
2114
2098
self.assertActivitiesMatch()
2116
2100
def test_post(self):
2117
self.server.canned_response = b'''HTTP/1.1 200 OK\r
2101
self.server.canned_response = '''HTTP/1.1 200 OK\r
2118
2102
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2119
2103
Server: Apache/2.0.54 (Fedora)\r
2120
2104
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2129
2113
t = self.get_transport()
2130
2114
# We must send a single line of body bytes, see
2131
2115
# PredefinedRequestHandler._handle_one_request
2132
code, f = t._post(b'abc def end-of-body\n')
2133
self.assertEqual(b'lalala whatever as long as itsssss\n', f.read())
2116
code, f = t._post('abc def end-of-body\n')
2117
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2134
2118
self.assertActivitiesMatch()
2175
2159
_password_prompt_prefix = ''
2176
2160
_username_prompt_prefix = ''
2177
2161
_auth_server = http_utils.HTTPBasicAuthServer
2178
_transport = HttpTransport
2162
_transport = _urllib.HttpTransport_urllib
2180
2164
def setUp(self):
2181
2165
super(TestAuthOnRedirected, self).setUp()
2182
self.build_tree_contents([('a', b'a'),
2166
self.build_tree_contents([('a', 'a'),
2184
('1/a', b'redirected once'),
2168
('1/a', 'redirected once'),
2186
2170
new_prefix = 'http://%s:%s' % (self.new_server.host,
2187
2171
self.new_server.port)
2188
2172
self.old_server.redirections = [
2189
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2173
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2190
2174
self.old_transport = self.get_old_transport()
2191
2175
self.new_server.add_user('joe', 'foo')
2192
2176
cleanup_http_redirection_connections(self)
2209
2193
return redirected_t
2211
2195
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2212
self.assertEqual(b'redirected once',
2196
self.assertEqual('redirected once',
2213
2197
transport.do_catching_redirections(
2214
self.get_a, self.old_transport, redirected).read())
2198
self.get_a, self.old_transport, redirected).read())
2215
2199
self.assertEqual(1, self.redirections)
2216
2200
# stdin should be empty
2217
2201
self.assertEqual('', ui.ui_factory.stdin.readline())
2222
2206
self.new_server.add_user('joe', 'foo')
2223
2207
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2224
2208
t = self.old_transport
2209
req = RedirectedRequest('GET', t.abspath('a'))
2225
2210
new_prefix = 'http://%s:%s' % (self.new_server.host,
2226
2211
self.new_server.port)
2227
2212
self.old_server.redirections = [
2228
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2231
t.request('GET', t.abspath('a'), retries=3).read())
2213
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2214
self.assertEqual('redirected once', t._perform(req).read())
2232
2215
# stdin should be empty
2233
2216
self.assertEqual('', ui.ui_factory.stdin.readline())
2234
2217
# stdout should be empty, stderr will contains the prompts
2235
2218
self.assertEqual('', ui.ui_factory.stdout.getvalue())
added
removed
breezy/tests/test_http.py
