/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_http.py

  • Committer: Jelmer Vernooij
  • Date: 2019-06-03 03:10:29 UTC
  • mfrom: (7312 work)
  • mto: This revision was merged to the branch mainline in revision 7318.
  • Revision ID: jelmer@jelmer.uk-20190603031029-b34je03bjulxxdwj
Merge trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
67
67
    )
68
68
from ..transport.http import (
69
69
    HttpTransport,
70
 
    _urllib2_wrappers,
71
70
    )
72
71
 
73
72
 
226
225
 
227
226
    def parse_header(self, header, auth_handler_class=None):
228
227
        if auth_handler_class is None:
229
 
            auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
 
228
            auth_handler_class = http.AbstractAuthHandler
230
229
        self.auth_handler = auth_handler_class()
231
230
        return self.auth_handler._parse_auth_header(header)
232
231
 
247
246
        self.assertEqual('realm="Thou should not pass"', remainder)
248
247
 
249
248
    def test_build_basic_header_with_long_creds(self):
250
 
        handler = _urllib2_wrappers.BasicAuthHandler()
 
249
        handler = http.BasicAuthHandler()
251
250
        user = 'user' * 10  # length 40
252
251
        password = 'password' * 5  # length 40
253
252
        header = handler.build_auth_header(
259
258
    def test_basic_extract_realm(self):
260
259
        scheme, remainder = self.parse_header(
261
260
            'Basic realm="Thou should not pass"',
262
 
            _urllib2_wrappers.BasicAuthHandler)
 
261
            http.BasicAuthHandler)
263
262
        match, realm = self.auth_handler.extract_realm(remainder)
264
263
        self.assertTrue(match is not None)
265
264
        self.assertEqual(u'Thou should not pass', realm)
1130
1129
    """
1131
1130
 
1132
1131
    def _proxied_request(self):
1133
 
        handler = _urllib2_wrappers.ProxyHandler()
1134
 
        request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
 
1132
        handler = http.ProxyHandler()
 
1133
        request = http.Request('GET', 'http://baz/buzzle')
1135
1134
        handler.set_proxy(request, 'http')
1136
1135
        return request
1137
1136
 
1138
1137
    def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1139
 
        handler = _urllib2_wrappers.ProxyHandler()
 
1138
        handler = http.ProxyHandler()
1140
1139
        self.assertEqual(expected,
1141
1140
                         handler.evaluate_proxy_bypass(host, no_proxy))
1142
1141
 
1328
1327
            self.get_new_transport().get('a').read())
1329
1328
 
1330
1329
 
1331
 
class RedirectedRequest(_urllib2_wrappers.Request):
 
1330
class RedirectedRequest(http.Request):
1332
1331
    """Request following redirections. """
1333
1332
 
1334
 
    init_orig = _urllib2_wrappers.Request.__init__
 
1333
    init_orig = http.Request.__init__
1335
1334
 
1336
1335
    def __init__(self, method, url, *args, **kwargs):
1337
1336
        """Constructor.
1338
1337
 
1339
1338
        """
1340
1339
        # Since the tests using this class will replace
1341
 
        # _urllib2_wrappers.Request, we can't just call the base class __init__
 
1340
        # http.Request, we can't just call the base class __init__
1342
1341
        # or we'll loop.
1343
1342
        RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1344
1343
        self.follow_redirections = True
1345
1344
 
1346
1345
 
1347
1346
def install_redirected_request(test):
1348
 
    test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
 
1347
    test.overrideAttr(http, 'Request', RedirectedRequest)
1349
1348
 
1350
1349
 
1351
1350
def cleanup_http_redirection_connections(test):
1352
1351
    # 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.
 
1352
    # the http level to be able to clean them up.
1354
1353
    def socket_disconnect(sock):
1355
1354
        try:
1356
1355
            sock.shutdown(socket.SHUT_RDWR)
1362
1361
        test.http_connect_orig(connection)
1363
1362
        test.addCleanup(socket_disconnect, connection.sock)
1364
1363
    test.http_connect_orig = test.overrideAttr(
1365
 
        _urllib2_wrappers.HTTPConnection, 'connect', connect)
 
1364
        http.HTTPConnection, 'connect', connect)
1366
1365
 
1367
1366
    def connect(connection):
1368
1367
        test.https_connect_orig(connection)
1369
1368
        test.addCleanup(socket_disconnect, connection.sock)
1370
1369
    test.https_connect_orig = test.overrideAttr(
1371
 
        _urllib2_wrappers.HTTPSConnection, 'connect', connect)
 
1370
        http.HTTPSConnection, 'connect', connect)
1372
1371
 
1373
1372
 
1374
1373
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1376
1375
 
1377
1376
    http implementations do not redirect silently anymore (they
1378
1377
    do not redirect at all in fact). The mechanism is still in
1379
 
    place at the _urllib2_wrappers.Request level and these tests
 
1378
    place at the http.Request level and these tests
1380
1379
    exercise it.
1381
1380
    """
1382
1381
 
1404
1403
 
1405
1404
    def test_one_redirection(self):
1406
1405
        t = self.get_old_transport()
1407
 
        req = RedirectedRequest('GET', t._remote_path('a'))
1408
1406
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1409
1407
                                       self.new_server.port)
1410
1408
        self.old_server.redirections = \
1411
1409
            [('(.*)', r'%s/1\1' % (new_prefix), 301), ]
1412
 
        self.assertEqual(b'redirected once', t._perform(req).read())
 
1410
        self.assertEqual(
 
1411
            b'redirected once',
 
1412
            t.request('GET', t._remote_path('a'), retries=1).read())
1413
1413
 
1414
1414
    def test_five_redirections(self):
1415
1415
        t = self.get_old_transport()
1416
 
        req = RedirectedRequest('GET', t._remote_path('a'))
1417
1416
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1418
1417
                                       self.old_server.port)
1419
1418
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1425
1424
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1426
1425
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1427
1426
            ]
1428
 
        self.assertEqual(b'redirected 5 times', t._perform(req).read())
 
1427
        self.assertEqual(
 
1428
            b'redirected 5 times',
 
1429
            t.request('GET', t._remote_path('a'), retries=6).read())
1429
1430
 
1430
1431
 
1431
1432
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1498
1499
        password = 'foo'
1499
1500
        _setup_authentication_config(scheme='http', host='localhost',
1500
1501
                                     user=user, password=password)
1501
 
        handler = _urllib2_wrappers.HTTPAuthHandler()
 
1502
        handler = http.HTTPAuthHandler()
1502
1503
        got_pass = handler.get_user_password(dict(
1503
1504
            user='joe',
1504
1505
            protocol='http',
2235
2236
        self.new_server.add_user('joe', 'foo')
2236
2237
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2237
2238
        t = self.old_transport
2238
 
        req = RedirectedRequest('GET', t.abspath('a'))
2239
2239
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2240
2240
                                       self.new_server.port)
2241
2241
        self.old_server.redirections = [
2242
2242
            ('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2243
 
        self.assertEqual(b'redirected once', t._perform(req).read())
 
2243
        self.assertEqual(
 
2244
            b'redirected once',
 
2245
            t.request('GET', t.abspath('a'), retries=3).read())
2244
2246
        # stdin should be empty
2245
2247
        self.assertEqual('', ui.ui_factory.stdin.readline())
2246
2248
        # stdout should be empty, stderr will contains the prompts