/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: John Arbash Meinel
  • Date: 2007-12-04 18:11:51 UTC
  • mfrom: (3074 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3075.
  • Revision ID: john@arbash-meinel.com-20071204181151-br85qwsgshso16q5
[merge] bzr.dev 3074

Show diffs side-by-side

added added

removed removed

Lines of Context:
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, self.get_readonly_server().GET_request_nb)
 
664
 
656
665
 
657
666
class TestSingleRangeRequestServer(TestRangeRequestServer):
658
667
    """Test readv against a server which accept only single range requests"""
709
718
 
710
719
 
711
720
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
 
                               TestNoRangeRequestServer,
713
 
                               TestCaseWithWebserver):
 
721
                                      TestNoRangeRequestServer,
 
722
                                      TestCaseWithWebserver):
714
723
    """Tests range requests refusing server for pycurl implementation"""
715
724
 
716
725
 
738
747
        TestCaseWithWebserver.setUp(self)
739
748
        # We need to manipulate ranges that correspond to real chunks in the
740
749
        # response, so we build a content appropriately.
741
 
        filler = ''.join(['abcdefghij' for _ in range(102)])
 
750
        filler = ''.join(['abcdefghij' for x in range(102)])
742
751
        content = ''.join(['%04d' % v + filler for v in range(16)])
743
752
        self.build_tree_contents([('a', content)],)
744
753
 
749
758
        self.assertEqual(l[1], (1024, '0001'))
750
759
        self.assertEqual(1, self.get_readonly_server().GET_request_nb)
751
760
 
752
 
    def test_a_lot_of_ranges(self):
 
761
    def test_more_ranges(self):
753
762
        t = self.get_transport()
754
763
        l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
755
764
        self.assertEqual(l[0], (0, '0000'))
775
784
 
776
785
 
777
786
 
778
 
class TestHttpProxyWhiteBox(TestCase):
 
787
class TestHttpProxyWhiteBox(tests.TestCase):
779
788
    """Whitebox test proxy http authorization.
780
789
 
781
790
    Only the urllib implementation is tested here.
782
791
    """
783
792
 
784
793
    def setUp(self):
785
 
        TestCase.setUp(self)
 
794
        tests.TestCase.setUp(self)
786
795
        self._old_env = {}
787
796
 
788
797
    def tearDown(self):
933
942
        # pycurl does not check HTTP_PROXY for security reasons
934
943
        # (for use in a CGI context that we do not care
935
944
        # about. Should we ?)
936
 
        raise TestSkipped('pycurl does not check HTTP_PROXY '
937
 
            'for security reasons')
 
945
        raise tests.TestNotApplicable(
 
946
            'pycurl does not check HTTP_PROXY for security reasons')
938
947
 
939
948
    def test_HTTP_PROXY_with_NO_PROXY(self):
940
 
        raise TestSkipped('pycurl does not check HTTP_PROXY '
941
 
            'for security reasons')
 
949
        raise tests.TestNotApplicable(
 
950
            'pycurl does not check HTTP_PROXY for security reasons')
942
951
 
943
952
    def test_http_proxy_without_scheme(self):
944
953
        # pycurl *ignores* invalid proxy env variables. If that
1272
1281
    def test_prompt_for_password(self):
1273
1282
        self.server.add_user('joe', 'foo')
1274
1283
        t = self.get_user_transport('joe', None)
1275
 
        stdout = StringIOWrapper()
1276
 
        ui.ui_factory = TestUIFactory(stdin='foo\n', stdout=stdout)
 
1284
        stdout = tests.StringIOWrapper()
 
1285
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1277
1286
        self.assertEqual('contents of a\n',t.get('a').read())
1278
1287
        # stdin should be empty
1279
1288
        self.assertEqual('', ui.ui_factory.stdin.readline())
1302
1311
        stdin_content = 'bar\n'  # Not the right password
1303
1312
        self.server.add_user(user, password)
1304
1313
        t = self.get_user_transport(user, None)
1305
 
        ui.ui_factory = TestUIFactory(stdin=stdin_content,
1306
 
                                      stdout=StringIOWrapper())
 
1314
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
 
1315
                                            stdout=tests.StringIOWrapper())
1307
1316
        # Create a minimal config file with the right password
1308
1317
        conf = config.AuthenticationConfig()
1309
1318
        conf._get_config().update(