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, parse_headers
28
from http.server import SimpleHTTPRequestHandler
29
except ImportError: # python < 3
30
from httplib import UnknownProtocol
31
from SimpleHTTPServer import SimpleHTTPRequestHandler
26
from http.client import UnknownProtocol, parse_headers
27
from http.server import SimpleHTTPRequestHandler
227
221
def parse_header(self, header, auth_handler_class=None):
228
222
if auth_handler_class is None:
229
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
223
auth_handler_class = http.AbstractAuthHandler
230
224
self.auth_handler = auth_handler_class()
231
225
return self.auth_handler._parse_auth_header(header)
247
241
self.assertEqual('realm="Thou should not pass"', remainder)
249
243
def test_build_basic_header_with_long_creds(self):
250
handler = _urllib2_wrappers.BasicAuthHandler()
244
handler = http.BasicAuthHandler()
251
245
user = 'user' * 10 # length 40
252
246
password = 'password' * 5 # length 40
253
247
header = handler.build_auth_header(
259
253
def test_basic_extract_realm(self):
260
254
scheme, remainder = self.parse_header(
261
255
'Basic realm="Thou should not pass"',
262
_urllib2_wrappers.BasicAuthHandler)
256
http.BasicAuthHandler)
263
257
match, realm = self.auth_handler.extract_realm(remainder)
264
258
self.assertTrue(match is not None)
265
259
self.assertEqual(u'Thou should not pass', realm)
1132
1126
def _proxied_request(self):
1133
handler = _urllib2_wrappers.ProxyHandler()
1134
request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
1127
handler = http.ProxyHandler()
1128
request = http.Request('GET', 'http://baz/buzzle')
1135
1129
handler.set_proxy(request, 'http')
1138
1132
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1139
handler = _urllib2_wrappers.ProxyHandler()
1133
handler = http.ProxyHandler()
1140
1134
self.assertEqual(expected,
1141
1135
handler.evaluate_proxy_bypass(host, no_proxy))
1328
1322
self.get_new_transport().get('a').read())
1331
class RedirectedRequest(_urllib2_wrappers.Request):
1325
class RedirectedRequest(http.Request):
1332
1326
"""Request following redirections. """
1334
init_orig = _urllib2_wrappers.Request.__init__
1328
init_orig = http.Request.__init__
1336
1330
def __init__(self, method, url, *args, **kwargs):
1337
1331
"""Constructor.
1340
1334
# Since the tests using this class will replace
1341
# _urllib2_wrappers.Request, we can't just call the base class __init__
1335
# http.Request, we can't just call the base class __init__
1342
1336
# or we'll loop.
1343
1337
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1344
1338
self.follow_redirections = True
1347
1341
def install_redirected_request(test):
1348
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1342
test.overrideAttr(http, 'Request', RedirectedRequest)
1351
1345
def cleanup_http_redirection_connections(test):
1352
1346
# Some sockets are opened but never seen by _urllib, so we trap them at
1353
# the _urllib2_wrappers level to be able to clean them up.
1347
# the http level to be able to clean them up.
1354
1348
def socket_disconnect(sock):
1356
1350
sock.shutdown(socket.SHUT_RDWR)
1362
1356
test.http_connect_orig(connection)
1363
1357
test.addCleanup(socket_disconnect, connection.sock)
1364
1358
test.http_connect_orig = test.overrideAttr(
1365
_urllib2_wrappers.HTTPConnection, 'connect', connect)
1359
http.HTTPConnection, 'connect', connect)
1367
1361
def connect(connection):
1368
1362
test.https_connect_orig(connection)
1369
1363
test.addCleanup(socket_disconnect, connection.sock)
1370
1364
test.https_connect_orig = test.overrideAttr(
1371
_urllib2_wrappers.HTTPSConnection, 'connect', connect)
1365
http.HTTPSConnection, 'connect', connect)
1374
1368
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1377
1371
http implementations do not redirect silently anymore (they
1378
1372
do not redirect at all in fact). The mechanism is still in
1379
place at the _urllib2_wrappers.Request level and these tests
1373
place at the http.Request level and these tests
1405
1399
def test_one_redirection(self):
1406
1400
t = self.get_old_transport()
1407
req = RedirectedRequest('GET', t._remote_path('a'))
1408
1401
new_prefix = 'http://%s:%s' % (self.new_server.host,
1409
1402
self.new_server.port)
1410
1403
self.old_server.redirections = \
1411
1404
[('(.*)', r'%s/1\1' % (new_prefix), 301), ]
1412
self.assertEqual(b'redirected once', t._perform(req).read())
1407
t.request('GET', t._remote_path('a'), retries=1).read())
1414
1409
def test_five_redirections(self):
1415
1410
t = self.get_old_transport()
1416
req = RedirectedRequest('GET', t._remote_path('a'))
1417
1411
old_prefix = 'http://%s:%s' % (self.old_server.host,
1418
1412
self.old_server.port)
1419
1413
new_prefix = 'http://%s:%s' % (self.new_server.host,
1425
1419
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1426
1420
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1428
self.assertEqual(b'redirected 5 times', t._perform(req).read())
1423
b'redirected 5 times',
1424
t.request('GET', t._remote_path('a'), retries=6).read())
1431
1427
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1498
1494
password = 'foo'
1499
1495
_setup_authentication_config(scheme='http', host='localhost',
1500
1496
user=user, password=password)
1501
handler = _urllib2_wrappers.HTTPAuthHandler()
1497
handler = http.HTTPAuthHandler()
1502
1498
got_pass = handler.get_user_password(dict(
1504
1500
protocol='http',
1948
1944
def _handle_one_request(self):
1949
1945
tcs = self.server.test_case_server
1950
1946
requestline = self.rfile.readline()
1952
headers = parse_headers(self.rfile)
1953
bytes_read = len(headers.as_bytes())
1954
bytes_read += headers.as_bytes().count(b'\n')
1955
bytes_read += len(requestline)
1957
headers = self.MessageClass(self.rfile, 0)
1958
# We just read: the request, the headers, an empty line indicating the
1959
# end of the headers.
1960
bytes_read = len(requestline)
1961
for line in headers.headers:
1962
bytes_read += len(line)
1963
bytes_read += len(b'\r\n')
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)
1964
1951
if requestline.startswith(b'POST'):
1965
1952
# The body should be a single line (or we don't know where it ends
1966
1953
# and we don't want to issue a blocking read)
2235
2222
self.new_server.add_user('joe', 'foo')
2236
2223
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2237
2224
t = self.old_transport
2238
req = RedirectedRequest('GET', t.abspath('a'))
2239
2225
new_prefix = 'http://%s:%s' % (self.new_server.host,
2240
2226
self.new_server.port)
2241
2227
self.old_server.redirections = [
2242
2228
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2243
self.assertEqual(b'redirected once', t._perform(req).read())
2231
t.request('GET', t.abspath('a'), retries=3).read())
2244
2232
# stdin should be empty
2245
2233
self.assertEqual('', ui.ui_factory.stdin.readline())
2246
2234
# stdout should be empty, stderr will contains the prompts