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