/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 bzrlib/tests/test_http.py

  • Committer: Lukáš Lalinský
  • Date: 2007-12-17 17:28:25 UTC
  • mfrom: (3120 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3123.
  • Revision ID: lalinsky@gmail.com-20071217172825-tr3pqm1mhvs3gwnn
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
32
32
    config,
33
33
    errors,
34
34
    osutils,
 
35
    tests,
35
36
    ui,
36
37
    urlutils,
37
38
    )
38
 
from bzrlib.tests import (
39
 
    TestCase,
40
 
    TestUIFactory,
41
 
    TestSkipped,
42
 
    StringIOWrapper,
43
 
    )
44
39
from bzrlib.tests.HttpServer import (
45
40
    HttpServer,
46
41
    HttpServer_PyCurl,
158
153
            from bzrlib.transport.http._pycurl import PyCurlTransport
159
154
            return PyCurlTransport
160
155
        except errors.DependencyNotPresent:
161
 
            raise TestSkipped('pycurl not present')
 
156
            raise tests.TestSkipped('pycurl not present')
162
157
 
163
158
    _transport = property(_get_pycurl_maybe)
164
159
 
165
160
 
166
 
class TestHttpUrls(TestCase):
 
161
class TestHttpUrls(tests.TestCase):
167
162
 
168
163
    # TODO: This should be moved to authorization tests once they
169
164
    # are written.
226
221
            server.tearDown()
227
222
 
228
223
 
229
 
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
 
224
class TestHttpUrls_urllib(TestHttpTransportUrls, tests.TestCase):
230
225
    """Test http urls with urllib"""
231
226
 
232
227
    _transport = HttpTransport_urllib
235
230
 
236
231
 
237
232
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
238
 
                          TestCase):
 
233
                          tests.TestCase):
239
234
    """Test http urls with pycurl"""
240
235
 
241
236
    _server = HttpServer_PyCurl
254
249
        try:
255
250
            import pycurl
256
251
        except ImportError:
257
 
            raise TestSkipped('pycurl not present')
 
252
            raise tests.TestSkipped('pycurl not present')
258
253
        # Now that we have pycurl imported, we can fake its version_info
259
254
        # This was taken from a windows pycurl without SSL
260
255
        # (thanks to bialix)
352
347
    """Test http connections with pycurl"""
353
348
 
354
349
 
355
 
class TestHttpTransportRegistration(TestCase):
 
350
class TestHttpTransportRegistration(tests.TestCase):
356
351
    """Test registrations of various http implementations"""
357
352
 
358
353
    def test_http_registered(self):
372
367
        try:
373
368
            http_transport = get_transport(url)
374
369
        except errors.UnsupportedProtocol:
375
 
            raise TestSkipped('%s not available' % scheme)
 
370
            raise tests.TestSkipped('%s not available' % scheme)
376
371
        code, response = http_transport._post('abc def end-of-body')
377
372
        self.assertTrue(
378
373
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
384
379
            server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
385
380
 
386
381
 
387
 
class TestPost_urllib(TestCase, TestPost):
 
382
class TestPost_urllib(tests.TestCase, TestPost):
388
383
    """TestPost for urllib implementation"""
389
384
 
390
385
    _transport = HttpTransport_urllib
393
388
        self._test_post_body_is_received('http+urllib')
394
389
 
395
390
 
396
 
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
 
391
class TestPost_pycurl(TestWithTransport_pycurl, tests.TestCase, TestPost):
397
392
    """TestPost for pycurl implementation"""
398
393
 
399
394
    def test_post_body_is_received_pycurl(self):
400
395
        self._test_post_body_is_received('http+pycurl')
401
396
 
402
397
 
403
 
class TestRangeHeader(TestCase):
 
398
class TestRangeHeader(tests.TestCase):
404
399
    """Test range_header method"""
405
400
 
406
401
    def check_header(self, value, ranges=[], tail=0):
575
570
    """Tests forbidden server for pycurl implementation"""
576
571
 
577
572
 
578
 
class TestRecordingServer(TestCase):
 
573
class TestRecordingServer(tests.TestCase):
579
574
 
580
575
    def test_create(self):
581
576
        server = RecordingServer(expect_body_tail=None)
653
648
        self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
654
649
                              t.readv, 'a', [(12,2)])
655
650
 
 
651
    def test_readv_multiple_get_requests(self):
 
652
        server = self.get_readonly_server()
 
653
        t = self._transport(server.get_url())
 
654
        # force transport to issue multiple requests
 
655
        t._max_readv_combine = 1
 
656
        t._max_get_ranges = 1
 
657
        l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
 
658
        self.assertEqual(l[0], (0, '0'))
 
659
        self.assertEqual(l[1], (1, '1'))
 
660
        self.assertEqual(l[2], (3, '34'))
 
661
        self.assertEqual(l[3], (9, '9'))
 
662
        # The server should have issued 4 requests
 
663
        self.assertEqual(4, server.GET_request_nb)
 
664
 
 
665
    def test_readv_get_max_size(self):
 
666
        server = self.get_readonly_server()
 
667
        t = self._transport(server.get_url())
 
668
        # force transport to issue multiple requests by limiting the number of
 
669
        # bytes by request. Note that this apply to coalesced offsets only, a
 
670
        # single range ill keep its size even if bigger than the limit.
 
671
        t._get_max_size = 2
 
672
        l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
 
673
        self.assertEqual(l[0], (0, '0'))
 
674
        self.assertEqual(l[1], (1, '1'))
 
675
        self.assertEqual(l[2], (2, '2345'))
 
676
        self.assertEqual(l[3], (6, '6789'))
 
677
        # The server should have issued 3 requests
 
678
        self.assertEqual(3, server.GET_request_nb)
 
679
 
656
680
 
657
681
class TestSingleRangeRequestServer(TestRangeRequestServer):
658
682
    """Test readv against a server which accept only single range requests"""
709
733
 
710
734
 
711
735
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
 
                               TestNoRangeRequestServer,
713
 
                               TestCaseWithWebserver):
 
736
                                      TestNoRangeRequestServer,
 
737
                                      TestCaseWithWebserver):
714
738
    """Tests range requests refusing server for pycurl implementation"""
715
739
 
716
740
 
738
762
        TestCaseWithWebserver.setUp(self)
739
763
        # We need to manipulate ranges that correspond to real chunks in the
740
764
        # response, so we build a content appropriately.
741
 
        filler = ''.join(['abcdefghij' for _ in range(102)])
 
765
        filler = ''.join(['abcdefghij' for x in range(102)])
742
766
        content = ''.join(['%04d' % v + filler for v in range(16)])
743
767
        self.build_tree_contents([('a', content)],)
744
768
 
749
773
        self.assertEqual(l[1], (1024, '0001'))
750
774
        self.assertEqual(1, self.get_readonly_server().GET_request_nb)
751
775
 
752
 
    def test_a_lot_of_ranges(self):
 
776
    def test_more_ranges(self):
753
777
        t = self.get_transport()
754
778
        l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
755
779
        self.assertEqual(l[0], (0, '0000'))
775
799
 
776
800
 
777
801
 
778
 
class TestHttpProxyWhiteBox(TestCase):
 
802
class TestHttpProxyWhiteBox(tests.TestCase):
779
803
    """Whitebox test proxy http authorization.
780
804
 
781
805
    Only the urllib implementation is tested here.
782
806
    """
783
807
 
784
808
    def setUp(self):
785
 
        TestCase.setUp(self)
 
809
        tests.TestCase.setUp(self)
786
810
        self._old_env = {}
787
811
 
788
812
    def tearDown(self):
933
957
        # pycurl does not check HTTP_PROXY for security reasons
934
958
        # (for use in a CGI context that we do not care
935
959
        # about. Should we ?)
936
 
        raise TestSkipped('pycurl does not check HTTP_PROXY '
937
 
            'for security reasons')
 
960
        raise tests.TestNotApplicable(
 
961
            'pycurl does not check HTTP_PROXY for security reasons')
938
962
 
939
963
    def test_HTTP_PROXY_with_NO_PROXY(self):
940
 
        raise TestSkipped('pycurl does not check HTTP_PROXY '
941
 
            'for security reasons')
 
964
        raise tests.TestNotApplicable(
 
965
            'pycurl does not check HTTP_PROXY for security reasons')
942
966
 
943
967
    def test_http_proxy_without_scheme(self):
944
968
        # pycurl *ignores* invalid proxy env variables. If that
978
1002
    def _file_tail(self, relpath, tail_amount):
979
1003
        code, data = self.transport._get(relpath, [], tail_amount)
980
1004
        self.assertTrue(code in (200, 206),'_get returns: %d' % code)
981
 
        data.seek(-tail_amount + 1, 2)
 
1005
        data.seek(-tail_amount, 2)
982
1006
        return data.read(tail_amount)
983
1007
 
984
1008
    def test_range_header(self):
988
1012
        # Tail
989
1013
        self.assertEqual('789', self._file_tail('a', 3))
990
1014
        # Syntactically invalid range
991
 
        self.assertListRaises(errors.InvalidRange,
 
1015
        self.assertListRaises(errors.InvalidHttpRange,
992
1016
                          self._file_contents, 'a', [(4, 3)])
993
1017
        # Semantically invalid range
994
 
        self.assertListRaises(errors.InvalidRange,
 
1018
        self.assertListRaises(errors.InvalidHttpRange,
995
1019
                          self._file_contents, 'a', [(42, 128)])
996
1020
 
997
1021
 
1272
1296
    def test_prompt_for_password(self):
1273
1297
        self.server.add_user('joe', 'foo')
1274
1298
        t = self.get_user_transport('joe', None)
1275
 
        stdout = StringIOWrapper()
1276
 
        ui.ui_factory = TestUIFactory(stdin='foo\n', stdout=stdout)
 
1299
        stdout = tests.StringIOWrapper()
 
1300
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1277
1301
        self.assertEqual('contents of a\n',t.get('a').read())
1278
1302
        # stdin should be empty
1279
1303
        self.assertEqual('', ui.ui_factory.stdin.readline())
1302
1326
        stdin_content = 'bar\n'  # Not the right password
1303
1327
        self.server.add_user(user, password)
1304
1328
        t = self.get_user_transport(user, None)
1305
 
        ui.ui_factory = TestUIFactory(stdin=stdin_content,
1306
 
                                      stdout=StringIOWrapper())
 
1329
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
 
1330
                                            stdout=tests.StringIOWrapper())
1307
1331
        # Create a minimal config file with the right password
1308
1332
        conf = config.AuthenticationConfig()
1309
1333
        conf._get_config().update(