/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: 2009-06-12 18:05:15 UTC
  • mto: (4371.4.5 vila-better-heads)
  • mto: This revision was merged to the branch mainline in revision 4449.
  • Revision ID: john@arbash-meinel.com-20090612180515-t0cwbjsnve094oik
Add a failing test for handling nodes that are in the same linear chain.

It fails because the ancestry skipping causes us to miss the fact that the two nodes
are actually directly related. We could check at the beginning, as the 
code used to do, but I think that will be incomplete for the more-than-two
heads cases.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
48
48
    deprecated_in,
49
49
    )
50
50
from bzrlib.tests import (
51
 
    features,
52
51
    http_server,
53
52
    http_utils,
54
53
    )
62
61
    )
63
62
 
64
63
 
65
 
if features.pycurl.available():
 
64
try:
66
65
    from bzrlib.transport.http._pycurl import PyCurlTransport
 
66
    pycurl_present = True
 
67
except errors.DependencyNotPresent:
 
68
    pycurl_present = False
67
69
 
68
70
 
69
71
def load_tests(standard_tests, module, loader):
82
84
                        _server=http_server.HttpServer_urllib,
83
85
                        _qualified_prefix='http+urllib',)),
84
86
        ]
85
 
    if features.pycurl.available():
 
87
    if pycurl_present:
86
88
        transport_scenarios.append(
87
89
            ('pycurl', dict(_transport=PyCurlTransport,
88
90
                            _server=http_server.HttpServer_PyCurl,
154
156
        activity_scenarios.append(
155
157
            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,
156
158
                                  _transport=_urllib.HttpTransport_urllib,)),)
157
 
    if features.pycurl.available():
 
159
    if pycurl_present:
158
160
        activity_scenarios.append(
159
161
            ('pycurl,http', dict(_activity_server=ActivityHTTPServer,
160
162
                                 _transport=PyCurlTransport,)),)
202
204
    It records the bytes sent to it, and replies with a 200.
203
205
    """
204
206
 
205
 
    def __init__(self, expect_body_tail=None, scheme=''):
 
207
    def __init__(self, expect_body_tail=None):
206
208
        """Constructor.
207
209
 
208
210
        :type expect_body_tail: str
213
215
        self.host = None
214
216
        self.port = None
215
217
        self.received_bytes = ''
216
 
        self.scheme = scheme
217
 
 
218
 
    def get_url(self):
219
 
        return '%s://%s:%s/' % (self.scheme, self.host, self.port)
220
 
 
221
 
    def start_server(self):
 
218
 
 
219
    def setUp(self):
222
220
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
223
221
        self._sock.bind(('127.0.0.1', 0))
224
222
        self.host, self.port = self._sock.getsockname()
247
245
            # The client may have already closed the socket.
248
246
            pass
249
247
 
250
 
    def stop_server(self):
 
248
    def tearDown(self):
251
249
        try:
252
250
            self._sock.close()
253
251
        except socket.error:
267
265
 
268
266
    def test_empty_header(self):
269
267
        scheme, remainder = self.parse_header('')
270
 
        self.assertEqual('', scheme)
 
268
        self.assertEquals('', scheme)
271
269
        self.assertIs(None, remainder)
272
270
 
273
271
    def test_negotiate_header(self):
274
272
        scheme, remainder = self.parse_header('Negotiate')
275
 
        self.assertEqual('negotiate', scheme)
 
273
        self.assertEquals('negotiate', scheme)
276
274
        self.assertIs(None, remainder)
277
275
 
278
276
    def test_basic_header(self):
279
277
        scheme, remainder = self.parse_header(
280
278
            'Basic realm="Thou should not pass"')
281
 
        self.assertEqual('basic', scheme)
282
 
        self.assertEqual('realm="Thou should not pass"', remainder)
 
279
        self.assertEquals('basic', scheme)
 
280
        self.assertEquals('realm="Thou should not pass"', remainder)
283
281
 
284
282
    def test_basic_extract_realm(self):
285
283
        scheme, remainder = self.parse_header(
287
285
            _urllib2_wrappers.BasicAuthHandler)
288
286
        match, realm = self.auth_handler.extract_realm(remainder)
289
287
        self.assertTrue(match is not None)
290
 
        self.assertEqual('Thou should not pass', realm)
 
288
        self.assertEquals('Thou should not pass', realm)
291
289
 
292
290
    def test_digest_header(self):
293
291
        scheme, remainder = self.parse_header(
294
292
            'Digest realm="Thou should not pass"')
295
 
        self.assertEqual('digest', scheme)
296
 
        self.assertEqual('realm="Thou should not pass"', remainder)
 
293
        self.assertEquals('digest', scheme)
 
294
        self.assertEquals('realm="Thou should not pass"', remainder)
297
295
 
298
296
 
299
297
class TestHTTPServer(tests.TestCase):
306
304
 
307
305
        server = http_server.HttpServer(BogusRequestHandler)
308
306
        try:
309
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
307
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
310
308
        except:
311
 
            server.stop_server()
 
309
            server.tearDown()
312
310
            self.fail('HTTP Server creation did not raise UnknownProtocol')
313
311
 
314
312
    def test_force_invalid_protocol(self):
315
313
        server = http_server.HttpServer(protocol_version='HTTP/0.1')
316
314
        try:
317
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
315
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
318
316
        except:
319
 
            server.stop_server()
 
317
            server.tearDown()
320
318
            self.fail('HTTP Server creation did not raise UnknownProtocol')
321
319
 
322
320
    def test_server_start_and_stop(self):
323
321
        server = http_server.HttpServer()
324
 
        server.start_server()
325
 
        try:
326
 
            self.assertTrue(server._http_running)
327
 
        finally:
328
 
            server.stop_server()
 
322
        server.setUp()
 
323
        self.assertTrue(server._http_running)
 
324
        server.tearDown()
329
325
        self.assertFalse(server._http_running)
330
326
 
331
327
    def test_create_http_server_one_zero(self):
334
330
            protocol_version = 'HTTP/1.0'
335
331
 
336
332
        server = http_server.HttpServer(RequestHandlerOneZero)
337
 
        self.start_server(server)
 
333
        server.setUp()
 
334
        self.addCleanup(server.tearDown)
338
335
        self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
339
336
 
340
337
    def test_create_http_server_one_one(self):
343
340
            protocol_version = 'HTTP/1.1'
344
341
 
345
342
        server = http_server.HttpServer(RequestHandlerOneOne)
346
 
        self.start_server(server)
 
343
        server.setUp()
 
344
        self.addCleanup(server.tearDown)
347
345
        self.assertIsInstance(server._httpd,
348
346
                              http_server.TestingThreadingHTTPServer)
349
347
 
354
352
 
355
353
        server = http_server.HttpServer(RequestHandlerOneZero,
356
354
                                        protocol_version='HTTP/1.1')
357
 
        self.start_server(server)
 
355
        server.setUp()
 
356
        self.addCleanup(server.tearDown)
358
357
        self.assertIsInstance(server._httpd,
359
358
                              http_server.TestingThreadingHTTPServer)
360
359
 
365
364
 
366
365
        server = http_server.HttpServer(RequestHandlerOneOne,
367
366
                                        protocol_version='HTTP/1.0')
368
 
        self.start_server(server)
 
367
        server.setUp()
 
368
        self.addCleanup(server.tearDown)
369
369
        self.assertIsInstance(server._httpd,
370
370
                              http_server.TestingHTTPServer)
371
371
 
374
374
    """Test case to inherit from if pycurl is present"""
375
375
 
376
376
    def _get_pycurl_maybe(self):
377
 
        self.requireFeature(features.pycurl)
378
 
        return PyCurlTransport
 
377
        try:
 
378
            from bzrlib.transport.http._pycurl import PyCurlTransport
 
379
            return PyCurlTransport
 
380
        except errors.DependencyNotPresent:
 
381
            raise tests.TestSkipped('pycurl not present')
379
382
 
380
383
    _transport = property(_get_pycurl_maybe)
381
384
 
388
391
    def test_url_parsing(self):
389
392
        f = FakeManager()
390
393
        url = http.extract_auth('http://example.com', f)
391
 
        self.assertEqual('http://example.com', url)
392
 
        self.assertEqual(0, len(f.credentials))
 
394
        self.assertEquals('http://example.com', url)
 
395
        self.assertEquals(0, len(f.credentials))
393
396
        url = http.extract_auth(
394
 
            'http://user:pass@example.com/bzr/bzr.dev', f)
395
 
        self.assertEqual('http://example.com/bzr/bzr.dev', url)
396
 
        self.assertEqual(1, len(f.credentials))
397
 
        self.assertEqual([None, 'example.com', 'user', 'pass'],
398
 
                         f.credentials[0])
 
397
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
 
398
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
 
399
        self.assertEquals(1, len(f.credentials))
 
400
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
 
401
                          f.credentials[0])
399
402
 
400
403
 
401
404
class TestHttpTransportUrls(tests.TestCase):
428
431
    def test_http_impl_urls(self):
429
432
        """There are servers which ask for particular clients to connect"""
430
433
        server = self._server()
431
 
        server.start_server()
432
434
        try:
 
435
            server.setUp()
433
436
            url = server.get_url()
434
437
            self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
435
438
        finally:
436
 
            server.stop_server()
 
439
            server.tearDown()
437
440
 
438
441
 
439
442
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
448
451
        https by supplying a fake version_info that do not
449
452
        support it.
450
453
        """
451
 
        self.requireFeature(features.pycurl)
452
 
        # Import the module locally now that we now it's available.
453
 
        pycurl = features.pycurl.module
 
454
        try:
 
455
            import pycurl
 
456
        except ImportError:
 
457
            raise tests.TestSkipped('pycurl not present')
454
458
 
455
 
        self.overrideAttr(pycurl, 'version_info',
456
 
                          # Fake the pycurl version_info This was taken from
457
 
                          # a windows pycurl without SSL (thanks to bialix)
458
 
                          lambda : (2,
459
 
                                    '7.13.2',
460
 
                                    462082,
461
 
                                    'i386-pc-win32',
462
 
                                    2576,
463
 
                                    None,
464
 
                                    0,
465
 
                                    None,
466
 
                                    ('ftp', 'gopher', 'telnet',
467
 
                                     'dict', 'ldap', 'http', 'file'),
468
 
                                    None,
469
 
                                    0,
470
 
                                    None))
471
 
        self.assertRaises(errors.DependencyNotPresent, self._transport,
472
 
                          'https://launchpad.net')
 
459
        version_info_orig = pycurl.version_info
 
460
        try:
 
461
            # Now that we have pycurl imported, we can fake its version_info
 
462
            # This was taken from a windows pycurl without SSL
 
463
            # (thanks to bialix)
 
464
            pycurl.version_info = lambda : (2,
 
465
                                            '7.13.2',
 
466
                                            462082,
 
467
                                            'i386-pc-win32',
 
468
                                            2576,
 
469
                                            None,
 
470
                                            0,
 
471
                                            None,
 
472
                                            ('ftp', 'gopher', 'telnet',
 
473
                                             'dict', 'ldap', 'http', 'file'),
 
474
                                            None,
 
475
                                            0,
 
476
                                            None)
 
477
            self.assertRaises(errors.DependencyNotPresent, self._transport,
 
478
                              'https://launchpad.net')
 
479
        finally:
 
480
            # Restore the right function
 
481
            pycurl.version_info = version_info_orig
473
482
 
474
483
 
475
484
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
534
543
class TestPost(tests.TestCase):
535
544
 
536
545
    def test_post_body_is_received(self):
537
 
        server = RecordingServer(expect_body_tail='end-of-body',
538
 
            scheme=self._qualified_prefix)
539
 
        self.start_server(server)
540
 
        url = server.get_url()
 
546
        server = RecordingServer(expect_body_tail='end-of-body')
 
547
        server.setUp()
 
548
        self.addCleanup(server.tearDown)
 
549
        scheme = self._qualified_prefix
 
550
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
541
551
        http_transport = self._transport(url)
542
552
        code, response = http_transport._post('abc def end-of-body')
543
553
        self.assertTrue(
592
602
                                      protocol_version=self._protocol_version)
593
603
 
594
604
    def _testing_pycurl(self):
595
 
        # TODO: This is duplicated for lots of the classes in this file
596
 
        return (features.pycurl.available()
597
 
                and self._transport == PyCurlTransport)
 
605
        return pycurl_present and self._transport == PyCurlTransport
598
606
 
599
607
 
600
608
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
617
625
        # for details) make no distinction between a closed
618
626
        # socket and badly formatted status line, so we can't
619
627
        # just test for ConnectionError, we have to test
620
 
        # InvalidHttpResponse too. And pycurl may raise ConnectionReset
621
 
        # instead of ConnectionError too.
622
 
        self.assertRaises(( errors.ConnectionError, errors.ConnectionReset,
623
 
                            errors.InvalidHttpResponse),
 
628
        # InvalidHttpResponse too.
 
629
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
624
630
                          t.has, 'foo/bar')
625
631
 
626
632
    def test_http_get(self):
627
633
        server = self.get_readonly_server()
628
634
        t = self._transport(server.get_url())
629
 
        self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
630
 
                           errors.InvalidHttpResponse),
 
635
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
631
636
                          t.get, 'foo/bar')
632
637
 
633
638
 
709
714
    _req_handler_class = BadProtocolRequestHandler
710
715
 
711
716
    def setUp(self):
712
 
        if self._testing_pycurl():
 
717
        if pycurl_present and self._transport == PyCurlTransport:
713
718
            raise tests.TestNotApplicable(
714
719
                "pycurl doesn't check the protocol version")
715
720
        super(TestBadProtocolServer, self).setUp()
759
764
        self.assertEqual(None, server.host)
760
765
        self.assertEqual(None, server.port)
761
766
 
762
 
    def test_setUp_and_stop(self):
 
767
    def test_setUp_and_tearDown(self):
763
768
        server = RecordingServer(expect_body_tail=None)
764
 
        server.start_server()
 
769
        server.setUp()
765
770
        try:
766
771
            self.assertNotEqual(None, server.host)
767
772
            self.assertNotEqual(None, server.port)
768
773
        finally:
769
 
            server.stop_server()
 
774
            server.tearDown()
770
775
        self.assertEqual(None, server.host)
771
776
        self.assertEqual(None, server.port)
772
777
 
773
778
    def test_send_receive_bytes(self):
774
 
        server = RecordingServer(expect_body_tail='c', scheme='http')
775
 
        self.start_server(server)
 
779
        server = RecordingServer(expect_body_tail='c')
 
780
        server.setUp()
 
781
        self.addCleanup(server.tearDown)
776
782
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
777
783
        sock.connect((server.host, server.port))
778
784
        sock.sendall('abc')
1169
1175
        if self._testing_pycurl():
1170
1176
            # Oh my ! pycurl does not check for the port as part of
1171
1177
            # no_proxy :-( So we just test the host part
1172
 
            self.no_proxy_host = self.server.host
 
1178
            self.no_proxy_host = 'localhost'
1173
1179
        else:
1174
1180
            self.no_proxy_host = self.proxy_address
1175
1181
        # The secondary server is the proxy
1178
1184
        self._old_env = {}
1179
1185
 
1180
1186
    def _testing_pycurl(self):
1181
 
        # TODO: This is duplicated for lots of the classes in this file
1182
 
        return (features.pycurl.available()
1183
 
                and self._transport == PyCurlTransport)
 
1187
        return pycurl_present and self._transport == PyCurlTransport
1184
1188
 
1185
1189
    def create_transport_secondary_server(self):
1186
1190
        """Creates an http server that will serve files with
1359
1363
        self.follow_redirections = True
1360
1364
 
1361
1365
 
1362
 
def install_redirected_request(test):
1363
 
    test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1364
 
 
1365
 
 
1366
1366
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1367
1367
    """Test redirections.
1368
1368
 
1378
1378
    """
1379
1379
 
1380
1380
    def setUp(self):
1381
 
        if (features.pycurl.available()
1382
 
            and self._transport == PyCurlTransport):
 
1381
        if pycurl_present and self._transport == PyCurlTransport:
1383
1382
            raise tests.TestNotApplicable(
1384
1383
                "pycurl doesn't redirect silently annymore")
1385
1384
        super(TestHTTPSilentRedirections, self).setUp()
1386
 
        install_redirected_request(self)
 
1385
        self.setup_redirected_request()
 
1386
        self.addCleanup(self.cleanup_redirected_request)
1387
1387
        self.build_tree_contents([('a','a'),
1388
1388
                                  ('1/',),
1389
1389
                                  ('1/a', 'redirected once'),
1399
1399
 
1400
1400
        self.old_transport = self._transport(self.old_server.get_url())
1401
1401
 
 
1402
    def setup_redirected_request(self):
 
1403
        self.original_class = _urllib2_wrappers.Request
 
1404
        _urllib2_wrappers.Request = RedirectedRequest
 
1405
 
 
1406
    def cleanup_redirected_request(self):
 
1407
        _urllib2_wrappers.Request = self.original_class
 
1408
 
1402
1409
    def create_transport_secondary_server(self):
1403
1410
        """Create the secondary server, redirections are defined in the tests"""
1404
1411
        return http_utils.HTTPServerRedirecting(
1408
1415
        t = self.old_transport
1409
1416
 
1410
1417
        req = RedirectedRequest('GET', t.abspath('a'))
 
1418
        req.follow_redirections = True
1411
1419
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1412
1420
                                       self.new_server.port)
1413
1421
        self.old_server.redirections = \
1414
1422
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1415
 
        self.assertEqual('redirected once',t._perform(req).read())
 
1423
        self.assertEquals('redirected once',t._perform(req).read())
1416
1424
 
1417
1425
    def test_five_redirections(self):
1418
1426
        t = self.old_transport
1419
1427
 
1420
1428
        req = RedirectedRequest('GET', t.abspath('a'))
 
1429
        req.follow_redirections = True
1421
1430
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1422
1431
                                       self.old_server.port)
1423
1432
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1429
1438
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1430
1439
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1431
1440
            ]
1432
 
        self.assertEqual('redirected 5 times',t._perform(req).read())
 
1441
        self.assertEquals('redirected 5 times',t._perform(req).read())
1433
1442
 
1434
1443
 
1435
1444
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1448
1457
        t = self._transport(self.new_server.get_url())
1449
1458
 
1450
1459
        # We use None for redirected so that we fail if redirected
1451
 
        self.assertEqual('0123456789',
1452
 
                         transport.do_catching_redirections(
 
1460
        self.assertEquals('0123456789',
 
1461
                          transport.do_catching_redirections(
1453
1462
                self.get_a, t, None).read())
1454
1463
 
1455
1464
    def test_one_redirection(self):
1460
1469
            dir, file = urlutils.split(exception.target)
1461
1470
            return self._transport(dir)
1462
1471
 
1463
 
        self.assertEqual('0123456789',
1464
 
                         transport.do_catching_redirections(
 
1472
        self.assertEquals('0123456789',
 
1473
                          transport.do_catching_redirections(
1465
1474
                self.get_a, self.old_transport, redirected).read())
1466
 
        self.assertEqual(1, self.redirections)
 
1475
        self.assertEquals(1, self.redirections)
1467
1476
 
1468
1477
    def test_redirection_loop(self):
1469
1478
 
1497
1506
        return self._auth_server(protocol_version=self._protocol_version)
1498
1507
 
1499
1508
    def _testing_pycurl(self):
1500
 
        # TODO: This is duplicated for lots of the classes in this file
1501
 
        return (features.pycurl.available()
1502
 
                and self._transport == PyCurlTransport)
 
1509
        return pycurl_present and self._transport == PyCurlTransport
1503
1510
 
1504
1511
    def get_user_url(self, user, password):
1505
1512
        """Build an url embedding user and password"""
1570
1577
        self.assertEqual('', ui.ui_factory.stdin.readline())
1571
1578
        stderr.seek(0)
1572
1579
        expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1573
 
        self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1574
 
        self.assertEqual('', stdout.getvalue())
 
1580
        self.assertEquals(expected_prompt, stderr.read(len(expected_prompt)))
 
1581
        self.assertEquals('', stdout.getvalue())
1575
1582
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1576
1583
                                    stderr.readline())
1577
1584
 
1592
1599
        self.assertEqual('', ui.ui_factory.stdin.readline())
1593
1600
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1594
1601
                                    stderr.getvalue())
1595
 
        self.assertEqual('', stdout.getvalue())
 
1602
        self.assertEquals('', stdout.getvalue())
1596
1603
        # And we shouldn't prompt again for a different request
1597
1604
        # against the same transport.
1598
1605
        self.assertEqual('contents of b\n',t.get('b').read())
1608
1615
                              % (scheme.upper(),
1609
1616
                                 user, self.server.host, self.server.port,
1610
1617
                                 self.server.auth_realm)))
1611
 
        self.assertEqual(expected_prompt, actual_prompt)
 
1618
        self.assertEquals(expected_prompt, actual_prompt)
1612
1619
 
1613
1620
    def _expected_username_prompt(self, scheme):
1614
1621
        return (self._username_prompt_prefix
1628
1635
        self.server.add_user(user, password)
1629
1636
        t = self.get_user_transport(user, None)
1630
1637
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1631
 
                                            stderr=tests.StringIOWrapper())
 
1638
                                            stdout=tests.StringIOWrapper())
1632
1639
        # Create a minimal config file with the right password
1633
1640
        conf = config.AuthenticationConfig()
1634
1641
        conf._get_config().update(
1846
1853
                             'http://www.example.com/foo/subdir')
1847
1854
        self.assertIsInstance(r, type(t))
1848
1855
        # Both transports share the some connection
1849
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1856
        self.assertEquals(t._get_connection(), r._get_connection())
1850
1857
 
1851
1858
    def test_redirected_to_self_with_slash(self):
1852
1859
        t = self._transport('http://www.example.com/foo')
1856
1863
        # Both transports share the some connection (one can argue that we
1857
1864
        # should return the exact same transport here, but that seems
1858
1865
        # overkill).
1859
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1866
        self.assertEquals(t._get_connection(), r._get_connection())
1860
1867
 
1861
1868
    def test_redirected_to_host(self):
1862
1869
        t = self._transport('http://www.example.com/foo')
1881
1888
        r = t._redirected_to('http://www.example.com/foo',
1882
1889
                             'https://foo.example.com/foo')
1883
1890
        self.assertIsInstance(r, type(t))
1884
 
        self.assertEqual(t._user, r._user)
 
1891
        self.assertEquals(t._user, r._user)
1885
1892
 
1886
1893
 
1887
1894
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1946
1953
        pass
1947
1954
 
1948
1955
 
1949
 
class TestActivityMixin(object):
 
1956
class TestActivity(tests.TestCase):
1950
1957
    """Test socket activity reporting.
1951
1958
 
1952
1959
    We use a special purpose server to control the bytes sent and received and
1956
1963
    def setUp(self):
1957
1964
        tests.TestCase.setUp(self)
1958
1965
        self.server = self._activity_server(self._protocol_version)
1959
 
        self.server.start_server()
 
1966
        self.server.setUp()
1960
1967
        self.activities = {}
1961
1968
        def report_activity(t, bytes, direction):
1962
1969
            count = self.activities.get(direction, 0)
1971
1978
 
1972
1979
    def tearDown(self):
1973
1980
        self._transport._report_activity = self.orig_report_activity
1974
 
        self.server.stop_server()
 
1981
        self.server.tearDown()
1975
1982
        tests.TestCase.tearDown(self)
1976
1983
 
1977
1984
    def get_transport(self):
2090
2097
        code, f = t._post('abc def end-of-body\n')
2091
2098
        self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2092
2099
        self.assertActivitiesMatch()
2093
 
 
2094
 
 
2095
 
class TestActivity(tests.TestCase, TestActivityMixin):
2096
 
 
2097
 
    def setUp(self):
2098
 
        tests.TestCase.setUp(self)
2099
 
        self.server = self._activity_server(self._protocol_version)
2100
 
        self.server.start_server()
2101
 
        self.activities = {}
2102
 
        def report_activity(t, bytes, direction):
2103
 
            count = self.activities.get(direction, 0)
2104
 
            count += bytes
2105
 
            self.activities[direction] = count
2106
 
 
2107
 
        # We override at class level because constructors may propagate the
2108
 
        # bound method and render instance overriding ineffective (an
2109
 
        # alternative would be to define a specific ui factory instead...)
2110
 
        self.orig_report_activity = self._transport._report_activity
2111
 
        self._transport._report_activity = report_activity
2112
 
 
2113
 
    def tearDown(self):
2114
 
        self._transport._report_activity = self.orig_report_activity
2115
 
        self.server.stop_server()
2116
 
        tests.TestCase.tearDown(self)
2117
 
 
2118
 
 
2119
 
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2120
 
 
2121
 
    def setUp(self):
2122
 
        tests.TestCase.setUp(self)
2123
 
        # Unlike TestActivity, we are really testing ReportingFileSocket and
2124
 
        # ReportingSocket, so we don't need all the parametrization. Since
2125
 
        # ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2126
 
        # test them through their use by the transport than directly (that's a
2127
 
        # bit less clean but far more simpler and effective).
2128
 
        self.server = ActivityHTTPServer('HTTP/1.1')
2129
 
        self._transport=_urllib.HttpTransport_urllib
2130
 
 
2131
 
        self.server.start_server()
2132
 
 
2133
 
        # We override at class level because constructors may propagate the
2134
 
        # bound method and render instance overriding ineffective (an
2135
 
        # alternative would be to define a specific ui factory instead...)
2136
 
        self.orig_report_activity = self._transport._report_activity
2137
 
        self._transport._report_activity = None
2138
 
 
2139
 
    def tearDown(self):
2140
 
        self._transport._report_activity = self.orig_report_activity
2141
 
        self.server.stop_server()
2142
 
        tests.TestCase.tearDown(self)
2143
 
 
2144
 
    def assertActivitiesMatch(self):
2145
 
        # Nothing to check here
2146
 
        pass
2147
 
 
2148
 
 
2149
 
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2150
 
    """Test authentication on the redirected http server."""
2151
 
 
2152
 
    _auth_header = 'Authorization'
2153
 
    _password_prompt_prefix = ''
2154
 
    _username_prompt_prefix = ''
2155
 
    _auth_server = http_utils.HTTPBasicAuthServer
2156
 
    _transport = _urllib.HttpTransport_urllib
2157
 
 
2158
 
    def create_transport_readonly_server(self):
2159
 
        return self._auth_server()
2160
 
 
2161
 
    def create_transport_secondary_server(self):
2162
 
        """Create the secondary server redirecting to the primary server"""
2163
 
        new = self.get_readonly_server()
2164
 
 
2165
 
        redirecting = http_utils.HTTPServerRedirecting()
2166
 
        redirecting.redirect_to(new.host, new.port)
2167
 
        return redirecting
2168
 
 
2169
 
    def setUp(self):
2170
 
        super(TestAuthOnRedirected, self).setUp()
2171
 
        self.build_tree_contents([('a','a'),
2172
 
                                  ('1/',),
2173
 
                                  ('1/a', 'redirected once'),
2174
 
                                  ],)
2175
 
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2176
 
                                       self.new_server.port)
2177
 
        self.old_server.redirections = [
2178
 
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
2179
 
        self.old_transport = self._transport(self.old_server.get_url())
2180
 
        self.new_server.add_user('joe', 'foo')
2181
 
 
2182
 
    def get_a(self, transport):
2183
 
        return transport.get('a')
2184
 
 
2185
 
    def test_auth_on_redirected_via_do_catching_redirections(self):
2186
 
        self.redirections = 0
2187
 
 
2188
 
        def redirected(transport, exception, redirection_notice):
2189
 
            self.redirections += 1
2190
 
            dir, file = urlutils.split(exception.target)
2191
 
            return self._transport(dir)
2192
 
 
2193
 
        stdout = tests.StringIOWrapper()
2194
 
        stderr = tests.StringIOWrapper()
2195
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2196
 
                                            stdout=stdout, stderr=stderr)
2197
 
        self.assertEqual('redirected once',
2198
 
                         transport.do_catching_redirections(
2199
 
                self.get_a, self.old_transport, redirected).read())
2200
 
        self.assertEqual(1, self.redirections)
2201
 
        # stdin should be empty
2202
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
2203
 
        # stdout should be empty, stderr will contains the prompts
2204
 
        self.assertEqual('', stdout.getvalue())
2205
 
 
2206
 
    def test_auth_on_redirected_via_following_redirections(self):
2207
 
        self.new_server.add_user('joe', 'foo')
2208
 
        stdout = tests.StringIOWrapper()
2209
 
        stderr = tests.StringIOWrapper()
2210
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2211
 
                                            stdout=stdout, stderr=stderr)
2212
 
        t = self.old_transport
2213
 
        req = RedirectedRequest('GET', t.abspath('a'))
2214
 
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2215
 
                                       self.new_server.port)
2216
 
        self.old_server.redirections = [
2217
 
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
2218
 
        self.assertEqual('redirected once',t._perform(req).read())
2219
 
        # stdin should be empty
2220
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
2221
 
        # stdout should be empty, stderr will contains the prompts
2222
 
        self.assertEqual('', stdout.getvalue())
2223
 
 
2224