« back to all changes in this revision
Viewing changes to bzrlib/tests/test_http.py
- browse files at revision 4759
- compare with another revision
- download diff
- download tarball
- view history from revision 4759
- Committer: Vincent Ladeuil
- Date: 2009-12-09 08:21:55 UTC
- mfrom: (4747.2.129 2.1.0b4-win32-test-imports)
- Revision ID: v.ladeuil+lp@free.fr-20091209082155-2ai0k52k996ayibh
Merge jam's fix for termios import
- files modified:
- NEWS
added
removed
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
Loggerhead is a web-based interface for Breezy
Version: 2.0.1