← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-12-04 17:06:17 UTC
  • mfrom: (4862.2.2 integration2)
  • Revision ID: pqm@pqm.ubuntu.com-20091204170617-8llyv1nv7g9cmiv7
(vila) Start fixing bug #395714 when authentication is required after
        an http redirection

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
269
269
 
270
270
    def test_empty_header(self):
271
271
        scheme, remainder = self.parse_header('')
272
 
        self.assertEquals('', scheme)
 
272
        self.assertEqual('', scheme)
273
273
        self.assertIs(None, remainder)
274
274
 
275
275
    def test_negotiate_header(self):
276
276
        scheme, remainder = self.parse_header('Negotiate')
277
 
        self.assertEquals('negotiate', scheme)
 
277
        self.assertEqual('negotiate', scheme)
278
278
        self.assertIs(None, remainder)
279
279
 
280
280
    def test_basic_header(self):
281
281
        scheme, remainder = self.parse_header(
282
282
            'Basic realm="Thou should not pass"')
283
 
        self.assertEquals('basic', scheme)
284
 
        self.assertEquals('realm="Thou should not pass"', remainder)
 
283
        self.assertEqual('basic', scheme)
 
284
        self.assertEqual('realm="Thou should not pass"', remainder)
285
285
 
286
286
    def test_basic_extract_realm(self):
287
287
        scheme, remainder = self.parse_header(
289
289
            _urllib2_wrappers.BasicAuthHandler)
290
290
        match, realm = self.auth_handler.extract_realm(remainder)
291
291
        self.assertTrue(match is not None)
292
 
        self.assertEquals('Thou should not pass', realm)
 
292
        self.assertEqual('Thou should not pass', realm)
293
293
 
294
294
    def test_digest_header(self):
295
295
        scheme, remainder = self.parse_header(
296
296
            'Digest realm="Thou should not pass"')
297
 
        self.assertEquals('digest', scheme)
298
 
        self.assertEquals('realm="Thou should not pass"', remainder)
 
297
        self.assertEqual('digest', scheme)
 
298
        self.assertEqual('realm="Thou should not pass"', remainder)
299
299
 
300
300
 
301
301
class TestHTTPServer(tests.TestCase):
393
393
    def test_url_parsing(self):
394
394
        f = FakeManager()
395
395
        url = http.extract_auth('http://example.com', f)
396
 
        self.assertEquals('http://example.com', url)
397
 
        self.assertEquals(0, len(f.credentials))
 
396
        self.assertEqual('http://example.com', url)
 
397
        self.assertEqual(0, len(f.credentials))
398
398
        url = http.extract_auth(
399
399
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
400
 
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
401
 
        self.assertEquals(1, len(f.credentials))
402
 
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
403
 
                          f.credentials[0])
 
400
        self.assertEqual('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
 
401
        self.assertEqual(1, len(f.credentials))
 
402
        self.assertEqual([None, 'www.bazaar-vcs.org', 'user', 'pass'],
 
403
                         f.credentials[0])
404
404
 
405
405
 
406
406
class TestHttpTransportUrls(tests.TestCase):
1366
1366
        self.follow_redirections = True
1367
1367
 
1368
1368
 
 
1369
def install_redirected_request(test):
 
1370
    test.original_class = _urllib2_wrappers.Request
 
1371
    def restore():
 
1372
        _urllib2_wrappers.Request = test.original_class
 
1373
    _urllib2_wrappers.Request = RedirectedRequest
 
1374
    test.addCleanup(restore)
 
1375
 
 
1376
 
1369
1377
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1370
1378
    """Test redirections.
1371
1379
 
1385
1393
            raise tests.TestNotApplicable(
1386
1394
                "pycurl doesn't redirect silently annymore")
1387
1395
        super(TestHTTPSilentRedirections, self).setUp()
1388
 
        self.setup_redirected_request()
1389
 
        self.addCleanup(self.cleanup_redirected_request)
 
1396
        install_redirected_request(self)
1390
1397
        self.build_tree_contents([('a','a'),
1391
1398
                                  ('1/',),
1392
1399
                                  ('1/a', 'redirected once'),
1402
1409
 
1403
1410
        self.old_transport = self._transport(self.old_server.get_url())
1404
1411
 
1405
 
    def setup_redirected_request(self):
1406
 
        self.original_class = _urllib2_wrappers.Request
1407
 
        _urllib2_wrappers.Request = RedirectedRequest
1408
 
 
1409
 
    def cleanup_redirected_request(self):
1410
 
        _urllib2_wrappers.Request = self.original_class
1411
 
 
1412
1412
    def create_transport_secondary_server(self):
1413
1413
        """Create the secondary server, redirections are defined in the tests"""
1414
1414
        return http_utils.HTTPServerRedirecting(
1418
1418
        t = self.old_transport
1419
1419
 
1420
1420
        req = RedirectedRequest('GET', t.abspath('a'))
1421
 
        req.follow_redirections = True
1422
1421
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1423
1422
                                       self.new_server.port)
1424
1423
        self.old_server.redirections = \
1425
1424
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1426
 
        self.assertEquals('redirected once',t._perform(req).read())
 
1425
        self.assertEqual('redirected once',t._perform(req).read())
1427
1426
 
1428
1427
    def test_five_redirections(self):
1429
1428
        t = self.old_transport
1430
1429
 
1431
1430
        req = RedirectedRequest('GET', t.abspath('a'))
1432
 
        req.follow_redirections = True
1433
1431
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1434
1432
                                       self.old_server.port)
1435
1433
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1441
1439
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1442
1440
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1443
1441
            ]
1444
 
        self.assertEquals('redirected 5 times',t._perform(req).read())
 
1442
        self.assertEqual('redirected 5 times',t._perform(req).read())
1445
1443
 
1446
1444
 
1447
1445
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1460
1458
        t = self._transport(self.new_server.get_url())
1461
1459
 
1462
1460
        # We use None for redirected so that we fail if redirected
1463
 
        self.assertEquals('0123456789',
1464
 
                          transport.do_catching_redirections(
 
1461
        self.assertEqual('0123456789',
 
1462
                         transport.do_catching_redirections(
1465
1463
                self.get_a, t, None).read())
1466
1464
 
1467
1465
    def test_one_redirection(self):
1472
1470
            dir, file = urlutils.split(exception.target)
1473
1471
            return self._transport(dir)
1474
1472
 
1475
 
        self.assertEquals('0123456789',
1476
 
                          transport.do_catching_redirections(
 
1473
        self.assertEqual('0123456789',
 
1474
                         transport.do_catching_redirections(
1477
1475
                self.get_a, self.old_transport, redirected).read())
1478
 
        self.assertEquals(1, self.redirections)
 
1476
        self.assertEqual(1, self.redirections)
1479
1477
 
1480
1478
    def test_redirection_loop(self):
1481
1479
 
1580
1578
        self.assertEqual('', ui.ui_factory.stdin.readline())
1581
1579
        stderr.seek(0)
1582
1580
        expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1583
 
        self.assertEquals(expected_prompt, stderr.read(len(expected_prompt)))
1584
 
        self.assertEquals('', stdout.getvalue())
 
1581
        self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
 
1582
        self.assertEqual('', stdout.getvalue())
1585
1583
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1586
1584
                                    stderr.readline())
1587
1585
 
1602
1600
        self.assertEqual('', ui.ui_factory.stdin.readline())
1603
1601
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1604
1602
                                    stderr.getvalue())
1605
 
        self.assertEquals('', stdout.getvalue())
 
1603
        self.assertEqual('', stdout.getvalue())
1606
1604
        # And we shouldn't prompt again for a different request
1607
1605
        # against the same transport.
1608
1606
        self.assertEqual('contents of b\n',t.get('b').read())
1618
1616
                              % (scheme.upper(),
1619
1617
                                 user, self.server.host, self.server.port,
1620
1618
                                 self.server.auth_realm)))
1621
 
        self.assertEquals(expected_prompt, actual_prompt)
 
1619
        self.assertEqual(expected_prompt, actual_prompt)
1622
1620
 
1623
1621
    def _expected_username_prompt(self, scheme):
1624
1622
        return (self._username_prompt_prefix
1638
1636
        self.server.add_user(user, password)
1639
1637
        t = self.get_user_transport(user, None)
1640
1638
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1641
 
                                            stdout=tests.StringIOWrapper())
 
1639
                                            stderr=tests.StringIOWrapper())
1642
1640
        # Create a minimal config file with the right password
1643
1641
        conf = config.AuthenticationConfig()
1644
1642
        conf._get_config().update(
1856
1854
                             'http://www.example.com/foo/subdir')
1857
1855
        self.assertIsInstance(r, type(t))
1858
1856
        # Both transports share the some connection
1859
 
        self.assertEquals(t._get_connection(), r._get_connection())
 
1857
        self.assertEqual(t._get_connection(), r._get_connection())
1860
1858
 
1861
1859
    def test_redirected_to_self_with_slash(self):
1862
1860
        t = self._transport('http://www.example.com/foo')
1866
1864
        # Both transports share the some connection (one can argue that we
1867
1865
        # should return the exact same transport here, but that seems
1868
1866
        # overkill).
1869
 
        self.assertEquals(t._get_connection(), r._get_connection())
 
1867
        self.assertEqual(t._get_connection(), r._get_connection())
1870
1868
 
1871
1869
    def test_redirected_to_host(self):
1872
1870
        t = self._transport('http://www.example.com/foo')
1891
1889
        r = t._redirected_to('http://www.example.com/foo',
1892
1890
                             'https://foo.example.com/foo')
1893
1891
        self.assertIsInstance(r, type(t))
1894
 
        self.assertEquals(t._user, r._user)
 
1892
        self.assertEqual(t._user, r._user)
1895
1893
 
1896
1894
 
1897
1895
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
2154
2152
    def assertActivitiesMatch(self):
2155
2153
        # Nothing to check here
2156
2154
        pass
 
2155
 
 
2156
 
 
2157
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
 
2158
    """Test authentication on the redirected http server."""
 
2159
 
 
2160
    _auth_header = 'Authorization'
 
2161
    _password_prompt_prefix = ''
 
2162
    _username_prompt_prefix = ''
 
2163
    _auth_server = http_utils.HTTPBasicAuthServer
 
2164
    _transport = _urllib.HttpTransport_urllib
 
2165
 
 
2166
    def create_transport_readonly_server(self):
 
2167
        return self._auth_server()
 
2168
 
 
2169
    def create_transport_secondary_server(self):
 
2170
        """Create the secondary server redirecting to the primary server"""
 
2171
        new = self.get_readonly_server()
 
2172
 
 
2173
        redirecting = http_utils.HTTPServerRedirecting()
 
2174
        redirecting.redirect_to(new.host, new.port)
 
2175
        return redirecting
 
2176
 
 
2177
    def setUp(self):
 
2178
        super(TestAuthOnRedirected, self).setUp()
 
2179
        self.build_tree_contents([('a','a'),
 
2180
                                  ('1/',),
 
2181
                                  ('1/a', 'redirected once'),
 
2182
                                  ],)
 
2183
        new_prefix = 'http://%s:%s' % (self.new_server.host,
 
2184
                                       self.new_server.port)
 
2185
        self.old_server.redirections = [
 
2186
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
 
2187
        self.old_transport = self._transport(self.old_server.get_url())
 
2188
        self.new_server.add_user('joe', 'foo')
 
2189
 
 
2190
    def get_a(self, transport):
 
2191
        return transport.get('a')
 
2192
 
 
2193
    def test_auth_on_redirected_via_do_catching_redirections(self):
 
2194
        self.redirections = 0
 
2195
 
 
2196
        def redirected(transport, exception, redirection_notice):
 
2197
            self.redirections += 1
 
2198
            dir, file = urlutils.split(exception.target)
 
2199
            return self._transport(dir)
 
2200
 
 
2201
        stdout = tests.StringIOWrapper()
 
2202
        stderr = tests.StringIOWrapper()
 
2203
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
 
2204
                                            stdout=stdout, stderr=stderr)
 
2205
        self.assertEqual('redirected once',
 
2206
                         transport.do_catching_redirections(
 
2207
                self.get_a, self.old_transport, redirected).read())
 
2208
        self.assertEqual(1, self.redirections)
 
2209
        # stdin should be empty
 
2210
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
2211
        # stdout should be empty, stderr will contains the prompts
 
2212
        self.assertEqual('', stdout.getvalue())
 
2213
 
 
2214
    def test_auth_on_redirected_via_following_redirections(self):
 
2215
        self.new_server.add_user('joe', 'foo')
 
2216
        stdout = tests.StringIOWrapper()
 
2217
        stderr = tests.StringIOWrapper()
 
2218
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
 
2219
                                            stdout=stdout, stderr=stderr)
 
2220
        t = self.old_transport
 
2221
        req = RedirectedRequest('GET', t.abspath('a'))
 
2222
        new_prefix = 'http://%s:%s' % (self.new_server.host,
 
2223
                                       self.new_server.port)
 
2224
        self.old_server.redirections = [
 
2225
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
 
2226
        self.assertEqual('redirected once',t._perform(req).read())
 
2227
        # stdin should be empty
 
2228
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
2229
        # stdout should be empty, stderr will contains the prompts
 
2230
        self.assertEqual('', stdout.getvalue())
 
2231
 
 
2232