346
362
self.assertIsInstance(t, HttpTransport_urllib)
349
class TestOffsets(TestCase):
350
"""Test offsets_to_ranges method"""
352
def test_offsets_to_ranges_simple(self):
353
to_range = HttpTransportBase.offsets_to_ranges
354
ranges = to_range([(10, 1)])
355
self.assertEqual([[10, 10]], ranges)
357
ranges = to_range([(0, 1), (1, 1)])
358
self.assertEqual([[0, 1]], ranges)
360
ranges = to_range([(1, 1), (0, 1)])
361
self.assertEqual([[0, 1]], ranges)
363
def test_offset_to_ranges_overlapped(self):
364
to_range = HttpTransportBase.offsets_to_ranges
366
ranges = to_range([(10, 1), (20, 2), (22, 5)])
367
self.assertEqual([[10, 10], [20, 26]], ranges)
369
ranges = to_range([(10, 1), (11, 2), (22, 5)])
370
self.assertEqual([[10, 12], [22, 26]], ranges)
373
365
class TestPost(object):
375
367
def _test_post_body_is_received(self, scheme):
699
714
"""Tests range requests refusing server for pycurl implementation"""
717
class TestLimitedRangeRequestServer(object):
718
"""Tests readv requests against server that errors out on too much ranges.
720
This MUST be used by daughter classes that also inherit from
721
TestCaseWithWebserver.
723
We can't inherit directly from TestCaseWithWebserver or the
724
test framework will try to create an instance which cannot
725
run, its implementation being incomplete.
730
def create_transport_readonly_server(self):
731
# Requests with more range specifiers will error out
732
return LimitedRangeHTTPServer(range_limit=self.range_limit)
734
def get_transport(self):
735
return self._transport(self.get_readonly_server().get_url())
738
TestCaseWithWebserver.setUp(self)
739
# We need to manipulate ranges that correspond to real chunks in the
740
# response, so we build a content appropriately.
741
filler = ''.join(['abcdefghij' for _ in range(102)])
742
content = ''.join(['%04d' % v + filler for v in range(16)])
743
self.build_tree_contents([('a', content)],)
745
def test_few_ranges(self):
746
t = self.get_transport()
747
l = list(t.readv('a', ((0, 4), (1024, 4), )))
748
self.assertEqual(l[0], (0, '0000'))
749
self.assertEqual(l[1], (1024, '0001'))
750
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
752
def test_a_lot_of_ranges(self):
753
t = self.get_transport()
754
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
755
self.assertEqual(l[0], (0, '0000'))
756
self.assertEqual(l[1], (1024, '0001'))
757
self.assertEqual(l[2], (4096, '0004'))
758
self.assertEqual(l[3], (8192, '0008'))
759
# The server will refuse to serve the first request (too much ranges),
760
# a second request will succeeds.
761
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
764
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
765
TestCaseWithWebserver):
766
"""Tests limited range requests server for urllib implementation"""
768
_transport = HttpTransport_urllib
771
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
772
TestLimitedRangeRequestServer,
773
TestCaseWithWebserver):
774
"""Tests limited range requests server for pycurl implementation"""
702
778
class TestHttpProxyWhiteBox(TestCase):
703
779
"""Whitebox test proxy http authorization.
705
These tests concern urllib implementation only.
781
Only the urllib implementation is tested here.
916
967
server = self.get_readonly_server()
917
968
self.transport = self._transport(server.get_url())
919
def _file_contents(self, relpath, ranges, tail_amount=0):
920
code, data = self.transport._get(relpath, ranges)
921
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
922
for start, end in ranges:
924
yield data.read(end - start + 1)
970
def _file_contents(self, relpath, ranges):
971
offsets = [ (start, end - start + 1) for start, end in ranges]
972
coalesce = self.transport._coalesce_offsets
973
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
974
code, data = self.transport._get(relpath, coalesced)
975
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
976
for start, end in ranges:
978
yield data.read(end - start + 1)
926
980
def _file_tail(self, relpath, tail_amount):
927
code, data = self.transport._get(relpath, [], tail_amount)
928
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
929
data.seek(-tail_amount + 1, 2)
930
return data.read(tail_amount)
981
code, data = self.transport._get(relpath, [], tail_amount)
982
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
983
data.seek(-tail_amount + 1, 2)
984
return data.read(tail_amount)
932
986
def test_range_header(self):
1146
1200
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1147
1201
self.get_a, self.old_transport, redirected)
1204
class TestAuth(object):
1205
"""Test some authentication scheme specified by daughter class.
1207
This MUST be used by daughter classes that also inherit from
1208
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1212
"""Set up the test environment
1214
Daughter classes should set up their own environment
1215
(including self.server) and explicitely call this
1216
method. This is needed because we want to reuse the same
1217
tests for proxy and no-proxy accesses which have
1218
different ways of setting self.server.
1220
self.build_tree_contents([('a', 'contents of a\n'),
1221
('b', 'contents of b\n'),])
1222
self.old_factory = ui.ui_factory
1223
# The following has the unfortunate side-effect of hiding any ouput
1224
# during the tests (including pdb prompts). Feel free to comment them
1225
# for debugging purposes but leave them in place, there are needed to
1226
# run the tests without any console
1227
self.old_stdout = sys.stdout
1228
sys.stdout = StringIOWrapper()
1229
self.addCleanup(self.restoreUIFactory)
1231
def restoreUIFactory(self):
1232
ui.ui_factory = self.old_factory
1233
sys.stdout = self.old_stdout
1235
def get_user_url(self, user=None, password=None):
1236
"""Build an url embedding user and password"""
1237
url = '%s://' % self.server._url_protocol
1238
if user is not None:
1240
if password is not None:
1241
url += ':' + password
1243
url += '%s:%s/' % (self.server.host, self.server.port)
1246
def test_no_user(self):
1247
self.server.add_user('joe', 'foo')
1248
t = self.get_user_transport()
1249
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1250
# Only one 'Authentication Required' error should occur
1251
self.assertEqual(1, self.server.auth_required_errors)
1253
def test_empty_pass(self):
1254
self.server.add_user('joe', '')
1255
t = self.get_user_transport('joe', '')
1256
self.assertEqual('contents of a\n', t.get('a').read())
1257
# Only one 'Authentication Required' error should occur
1258
self.assertEqual(1, self.server.auth_required_errors)
1260
def test_user_pass(self):
1261
self.server.add_user('joe', 'foo')
1262
t = self.get_user_transport('joe', 'foo')
1263
self.assertEqual('contents of a\n', t.get('a').read())
1264
# Only one 'Authentication Required' error should occur
1265
self.assertEqual(1, self.server.auth_required_errors)
1267
def test_unknown_user(self):
1268
self.server.add_user('joe', 'foo')
1269
t = self.get_user_transport('bill', 'foo')
1270
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1271
# Two 'Authentication Required' errors should occur (the
1272
# initial 'who are you' and 'I don't know you, who are
1274
self.assertEqual(2, self.server.auth_required_errors)
1276
def test_wrong_pass(self):
1277
self.server.add_user('joe', 'foo')
1278
t = self.get_user_transport('joe', 'bar')
1279
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1280
# Two 'Authentication Required' errors should occur (the
1281
# initial 'who are you' and 'this is not you, who are you')
1282
self.assertEqual(2, self.server.auth_required_errors)
1284
def test_prompt_for_password(self):
1285
self.server.add_user('joe', 'foo')
1286
t = self.get_user_transport('joe', None)
1287
ui.ui_factory = TestUIFactory(stdin='foo\n')
1288
self.assertEqual('contents of a\n',t.get('a').read())
1289
# stdin should be empty
1290
self.assertEqual('', ui.ui_factory.stdin.readline())
1291
# And we shouldn't prompt again for a different request
1292
# against the same transport.
1293
self.assertEqual('contents of b\n',t.get('b').read())
1295
# And neither against a clone
1296
self.assertEqual('contents of b\n',t2.get('b').read())
1297
# Only one 'Authentication Required' error should occur
1298
self.assertEqual(1, self.server.auth_required_errors)
1301
class TestHTTPAuth(TestAuth):
1302
"""Test HTTP authentication schemes.
1304
Daughter classes MUST inherit from TestCaseWithWebserver too.
1307
_auth_header = 'Authorization'
1310
TestCaseWithWebserver.setUp(self)
1311
self.server = self.get_readonly_server()
1312
TestAuth.setUp(self)
1314
def get_user_transport(self, user=None, password=None):
1315
return self._transport(self.get_user_url(user, password))
1318
class TestProxyAuth(TestAuth):
1319
"""Test proxy authentication schemes.
1321
Daughter classes MUST also inherit from TestCaseWithWebserver.
1323
_auth_header = 'Proxy-authorization'
1326
TestCaseWithWebserver.setUp(self)
1327
self.server = self.get_readonly_server()
1329
self.addCleanup(self._restore_env)
1330
TestAuth.setUp(self)
1331
# Override the contents to avoid false positives
1332
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1333
('b', 'not proxied contents of b\n'),
1334
('a-proxied', 'contents of a\n'),
1335
('b-proxied', 'contents of b\n'),
1338
def get_user_transport(self, user=None, password=None):
1339
self._install_env({'all_proxy': self.get_user_url(user, password)})
1340
return self._transport(self.server.get_url())
1342
def _install_env(self, env):
1343
for name, value in env.iteritems():
1344
self._old_env[name] = osutils.set_or_unset_env(name, value)
1346
def _restore_env(self):
1347
for name, value in self._old_env.iteritems():
1348
osutils.set_or_unset_env(name, value)
1351
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1352
"""Test http basic authentication scheme"""
1354
_transport = HttpTransport_urllib
1356
def create_transport_readonly_server(self):
1357
return HTTPBasicAuthServer()
1360
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1361
"""Test proxy basic authentication scheme"""
1363
_transport = HttpTransport_urllib
1365
def create_transport_readonly_server(self):
1366
return ProxyBasicAuthServer()
1369
class TestDigestAuth(object):
1370
"""Digest Authentication specific tests"""
1372
def test_changing_nonce(self):
1373
self.server.add_user('joe', 'foo')
1374
t = self.get_user_transport('joe', 'foo')
1375
self.assertEqual('contents of a\n', t.get('a').read())
1376
self.assertEqual('contents of b\n', t.get('b').read())
1377
# Only one 'Authentication Required' error should have
1379
self.assertEqual(1, self.server.auth_required_errors)
1380
# The server invalidates the current nonce
1381
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1382
self.assertEqual('contents of a\n', t.get('a').read())
1383
# Two 'Authentication Required' errors should occur (the
1384
# initial 'who are you' and a second 'who are you' with the new nonce)
1385
self.assertEqual(2, self.server.auth_required_errors)
1388
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1389
"""Test http digest authentication scheme"""
1391
_transport = HttpTransport_urllib
1393
def create_transport_readonly_server(self):
1394
return HTTPDigestAuthServer()
1397
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1398
TestCaseWithWebserver):
1399
"""Test proxy digest authentication scheme"""
1401
_transport = HttpTransport_urllib
1403
def create_transport_readonly_server(self):
1404
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
