53
53
HTTPDigestAuthServer,
54
54
HTTPServerRedirecting,
55
55
InvalidStatusRequestHandler,
56
LimitedRangeHTTPServer,
56
57
NoRangeRequestHandler,
57
58
ProxyBasicAuthServer,
58
59
ProxyDigestAuthServer,
60
61
SingleRangeRequestHandler,
62
SingleOnlyRangeRequestHandler,
61
63
TestCaseWithRedirectedWebserver,
62
64
TestCaseWithTwoWebservers,
63
65
TestCaseWithWebserver,
64
66
WallRequestHandler,
66
68
from bzrlib.transport import (
67
70
do_catching_redirections,
360
363
self.assertIsInstance(t, HttpTransport_urllib)
363
class TestOffsets(TestCase):
364
"""Test offsets_to_ranges method"""
366
def test_offsets_to_ranges_simple(self):
367
to_range = HttpTransportBase.offsets_to_ranges
368
ranges = to_range([(10, 1)])
369
self.assertEqual([[10, 10]], ranges)
371
ranges = to_range([(0, 1), (1, 1)])
372
self.assertEqual([[0, 1]], ranges)
374
ranges = to_range([(1, 1), (0, 1)])
375
self.assertEqual([[0, 1]], ranges)
377
def test_offset_to_ranges_overlapped(self):
378
to_range = HttpTransportBase.offsets_to_ranges
380
ranges = to_range([(10, 1), (20, 2), (22, 5)])
381
self.assertEqual([[10, 10], [20, 26]], ranges)
383
ranges = to_range([(10, 1), (11, 2), (22, 5)])
384
self.assertEqual([[10, 12], [22, 26]], ranges)
387
366
class TestPost(object):
389
368
def _test_post_body_is_received(self, scheme):
426
405
"""Test range_header method"""
428
407
def check_header(self, value, ranges=[], tail=0):
429
range_header = HttpTransportBase.range_header
430
self.assertEqual(value, range_header(ranges, tail))
408
offsets = [ (start, end - start + 1) for start, end in ranges]
409
coalesce = Transport._coalesce_offsets
410
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
411
range_header = HttpTransportBase._range_header
412
self.assertEqual(value, range_header(coalesced, tail))
432
414
def test_range_header_single(self):
433
self.check_header('0-9', ranges=[[0,9]])
434
self.check_header('100-109', ranges=[[100,109]])
415
self.check_header('0-9', ranges=[(0,9)])
416
self.check_header('100-109', ranges=[(100,109)])
436
418
def test_range_header_tail(self):
437
419
self.check_header('-10', tail=10)
693
675
"""Tests single range requests accepting server for pycurl implementation"""
678
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
679
"""Test readv against a server which only accept single range requests"""
681
def create_transport_readonly_server(self):
682
return HttpServer(SingleOnlyRangeRequestHandler)
685
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
686
TestCaseWithWebserver):
687
"""Tests single range requests accepting server for urllib implementation"""
689
_transport = HttpTransport_urllib
692
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
693
TestSingleOnlyRangeRequestServer,
694
TestCaseWithWebserver):
695
"""Tests single range requests accepting server for pycurl implementation"""
696
698
class TestNoRangeRequestServer(TestRangeRequestServer):
697
699
"""Test readv against a server which do not accept range requests"""
713
715
"""Tests range requests refusing server for pycurl implementation"""
718
class TestLimitedRangeRequestServer(object):
719
"""Tests readv requests against server that errors out on too much ranges.
721
This MUST be used by daughter classes that also inherit from
722
TestCaseWithWebserver.
724
We can't inherit directly from TestCaseWithWebserver or the
725
test framework will try to create an instance which cannot
726
run, its implementation being incomplete.
731
def create_transport_readonly_server(self):
732
# Requests with more range specifiers will error out
733
return LimitedRangeHTTPServer(range_limit=self.range_limit)
735
def get_transport(self):
736
return self._transport(self.get_readonly_server().get_url())
739
TestCaseWithWebserver.setUp(self)
740
# We need to manipulate ranges that correspond to real chunks in the
741
# response, so we build a content appropriately.
742
filler = ''.join(['abcdefghij' for _ in range(102)])
743
content = ''.join(['%04d' % v + filler for v in range(16)])
744
self.build_tree_contents([('a', content)],)
746
def test_few_ranges(self):
747
t = self.get_transport()
748
l = list(t.readv('a', ((0, 4), (1024, 4), )))
749
self.assertEqual(l[0], (0, '0000'))
750
self.assertEqual(l[1], (1024, '0001'))
751
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
753
def test_a_lot_of_ranges(self):
754
t = self.get_transport()
755
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
756
self.assertEqual(l[0], (0, '0000'))
757
self.assertEqual(l[1], (1024, '0001'))
758
self.assertEqual(l[2], (4096, '0004'))
759
self.assertEqual(l[3], (8192, '0008'))
760
# The server will refuse to serve the first request (too much ranges),
761
# a second request will succeeds.
762
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
765
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
766
TestCaseWithWebserver):
767
"""Tests limited range requests server for urllib implementation"""
769
_transport = HttpTransport_urllib
772
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
773
TestLimitedRangeRequestServer,
774
TestCaseWithWebserver):
775
"""Tests limited range requests server for pycurl implementation"""
716
779
class TestHttpProxyWhiteBox(TestCase):
717
780
"""Whitebox test proxy http authorization.
905
968
server = self.get_readonly_server()
906
969
self.transport = self._transport(server.get_url())
908
def _file_contents(self, relpath, ranges, tail_amount=0):
909
code, data = self.transport._get(relpath, ranges)
910
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
911
for start, end in ranges:
913
yield data.read(end - start + 1)
971
def _file_contents(self, relpath, ranges):
972
offsets = [ (start, end - start + 1) for start, end in ranges]
973
coalesce = self.transport._coalesce_offsets
974
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
975
code, data = self.transport._get(relpath, coalesced)
976
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
977
for start, end in ranges:
979
yield data.read(end - start + 1)
915
981
def _file_tail(self, relpath, tail_amount):
916
code, data = self.transport._get(relpath, [], tail_amount)
917
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
918
data.seek(-tail_amount + 1, 2)
919
return data.read(tail_amount)
982
code, data = self.transport._get(relpath, [], tail_amount)
983
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
984
data.seek(-tail_amount + 1, 2)
985
return data.read(tail_amount)
921
987
def test_range_header(self):
926
992
self.assertEqual('789', self._file_tail('a', 3))
927
993
# Syntactically invalid range
928
self.assertRaises(errors.InvalidRange,
929
self.transport._get, 'a', [(4, 3)])
994
self.assertListRaises(errors.InvalidRange,
995
self._file_contents, 'a', [(4, 3)])
930
996
# Semantically invalid range
931
self.assertRaises(errors.InvalidRange,
932
self.transport._get, 'a', [(42, 128)])
997
self.assertListRaises(errors.InvalidRange,
998
self._file_contents, 'a', [(42, 128)])
935
1001
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1155
1221
self.build_tree_contents([('a', 'contents of a\n'),
1156
1222
('b', 'contents of b\n'),])
1157
1223
self.old_factory = ui.ui_factory
1224
# The following has the unfortunate side-effect of hiding any ouput
1225
# during the tests (including pdb prompts). Feel free to comment them
1226
# for debugging purposes but leave them in place, there are needed to
1227
# run the tests without any console
1158
1228
self.old_stdout = sys.stdout
1159
1229
sys.stdout = StringIOWrapper()
1160
1230
self.addCleanup(self.restoreUIFactory)
added
removed
bzrlib/tests/test_http.py
