/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

Rework test_script a little bit.


Don't allow someone to request a stdin request to echo.
Echo never reads from stdin, it just echos its arguments.
You use 'cat' if you want to read from stdin.

A few other fixes because the tests were using filenames
that are actually illegal on Windows, rather than just
nonexistant.


Change the exception handling for commands so that
unknown errors don't get silently squashed and then
turn into hard-to-debug errors later.

test_script now passes on Windows.

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,)),)
218
220
    def get_url(self):
219
221
        return '%s://%s:%s/' % (self.scheme, self.host, self.port)
220
222
 
221
 
    def start_server(self):
 
223
    def setUp(self):
222
224
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
223
225
        self._sock.bind(('127.0.0.1', 0))
224
226
        self.host, self.port = self._sock.getsockname()
247
249
            # The client may have already closed the socket.
248
250
            pass
249
251
 
250
 
    def stop_server(self):
 
252
    def tearDown(self):
251
253
        try:
252
254
            self._sock.close()
253
255
        except socket.error:
267
269
 
268
270
    def test_empty_header(self):
269
271
        scheme, remainder = self.parse_header('')
270
 
        self.assertEqual('', scheme)
 
272
        self.assertEquals('', scheme)
271
273
        self.assertIs(None, remainder)
272
274
 
273
275
    def test_negotiate_header(self):
274
276
        scheme, remainder = self.parse_header('Negotiate')
275
 
        self.assertEqual('negotiate', scheme)
 
277
        self.assertEquals('negotiate', scheme)
276
278
        self.assertIs(None, remainder)
277
279
 
278
280
    def test_basic_header(self):
279
281
        scheme, remainder = self.parse_header(
280
282
            'Basic realm="Thou should not pass"')
281
 
        self.assertEqual('basic', scheme)
282
 
        self.assertEqual('realm="Thou should not pass"', remainder)
 
283
        self.assertEquals('basic', scheme)
 
284
        self.assertEquals('realm="Thou should not pass"', remainder)
283
285
 
284
286
    def test_basic_extract_realm(self):
285
287
        scheme, remainder = self.parse_header(
287
289
            _urllib2_wrappers.BasicAuthHandler)
288
290
        match, realm = self.auth_handler.extract_realm(remainder)
289
291
        self.assertTrue(match is not None)
290
 
        self.assertEqual('Thou should not pass', realm)
 
292
        self.assertEquals('Thou should not pass', realm)
291
293
 
292
294
    def test_digest_header(self):
293
295
        scheme, remainder = self.parse_header(
294
296
            'Digest realm="Thou should not pass"')
295
 
        self.assertEqual('digest', scheme)
296
 
        self.assertEqual('realm="Thou should not pass"', remainder)
 
297
        self.assertEquals('digest', scheme)
 
298
        self.assertEquals('realm="Thou should not pass"', remainder)
297
299
 
298
300
 
299
301
class TestHTTPServer(tests.TestCase):
306
308
 
307
309
        server = http_server.HttpServer(BogusRequestHandler)
308
310
        try:
309
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
311
            self.assertRaises(httplib.UnknownProtocol, server.setUp)
310
312
        except:
311
 
            server.stop_server()
 
313
            server.tearDown()
312
314
            self.fail('HTTP Server creation did not raise UnknownProtocol')
313
315
 
314
316
    def test_force_invalid_protocol(self):
315
317
        server = http_server.HttpServer(protocol_version='HTTP/0.1')
316
318
        try:
317
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
319
            self.assertRaises(httplib.UnknownProtocol, server.setUp)
318
320
        except:
319
 
            server.stop_server()
 
321
            server.tearDown()
320
322
            self.fail('HTTP Server creation did not raise UnknownProtocol')
321
323
 
322
324
    def test_server_start_and_stop(self):
323
325
        server = http_server.HttpServer()
324
 
        server.start_server()
 
326
        server.setUp()
325
327
        try:
326
328
            self.assertTrue(server._http_running)
327
329
        finally:
328
 
            server.stop_server()
 
330
            server.tearDown()
329
331
        self.assertFalse(server._http_running)
330
332
 
331
333
    def test_create_http_server_one_zero(self):
374
376
    """Test case to inherit from if pycurl is present"""
375
377
 
376
378
    def _get_pycurl_maybe(self):
377
 
        self.requireFeature(features.pycurl)
378
 
        return PyCurlTransport
 
379
        try:
 
380
            from bzrlib.transport.http._pycurl import PyCurlTransport
 
381
            return PyCurlTransport
 
382
        except errors.DependencyNotPresent:
 
383
            raise tests.TestSkipped('pycurl not present')
379
384
 
380
385
    _transport = property(_get_pycurl_maybe)
381
386
 
388
393
    def test_url_parsing(self):
389
394
        f = FakeManager()
390
395
        url = http.extract_auth('http://example.com', f)
391
 
        self.assertEqual('http://example.com', url)
392
 
        self.assertEqual(0, len(f.credentials))
 
396
        self.assertEquals('http://example.com', url)
 
397
        self.assertEquals(0, len(f.credentials))
393
398
        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])
 
399
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
 
400
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
 
401
        self.assertEquals(1, len(f.credentials))
 
402
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
 
403
                          f.credentials[0])
399
404
 
400
405
 
401
406
class TestHttpTransportUrls(tests.TestCase):
428
433
    def test_http_impl_urls(self):
429
434
        """There are servers which ask for particular clients to connect"""
430
435
        server = self._server()
431
 
        server.start_server()
 
436
        server.setUp()
432
437
        try:
433
438
            url = server.get_url()
434
439
            self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
435
440
        finally:
436
 
            server.stop_server()
 
441
            server.tearDown()
437
442
 
438
443
 
439
444
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
448
453
        https by supplying a fake version_info that do not
449
454
        support it.
450
455
        """
451
 
        self.requireFeature(features.pycurl)
452
 
        # Import the module locally now that we now it's available.
453
 
        pycurl = features.pycurl.module
 
456
        try:
 
457
            import pycurl
 
458
        except ImportError:
 
459
            raise tests.TestSkipped('pycurl not present')
454
460
 
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')
 
461
        version_info_orig = pycurl.version_info
 
462
        try:
 
463
            # Now that we have pycurl imported, we can fake its version_info
 
464
            # This was taken from a windows pycurl without SSL
 
465
            # (thanks to bialix)
 
466
            pycurl.version_info = lambda : (2,
 
467
                                            '7.13.2',
 
468
                                            462082,
 
469
                                            'i386-pc-win32',
 
470
                                            2576,
 
471
                                            None,
 
472
                                            0,
 
473
                                            None,
 
474
                                            ('ftp', 'gopher', 'telnet',
 
475
                                             'dict', 'ldap', 'http', 'file'),
 
476
                                            None,
 
477
                                            0,
 
478
                                            None)
 
479
            self.assertRaises(errors.DependencyNotPresent, self._transport,
 
480
                              'https://launchpad.net')
 
481
        finally:
 
482
            # Restore the right function
 
483
            pycurl.version_info = version_info_orig
473
484
 
474
485
 
475
486
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
592
603
                                      protocol_version=self._protocol_version)
593
604
 
594
605
    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)
 
606
        return pycurl_present and self._transport == PyCurlTransport
598
607
 
599
608
 
600
609
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
709
718
    _req_handler_class = BadProtocolRequestHandler
710
719
 
711
720
    def setUp(self):
712
 
        if self._testing_pycurl():
 
721
        if pycurl_present and self._transport == PyCurlTransport:
713
722
            raise tests.TestNotApplicable(
714
723
                "pycurl doesn't check the protocol version")
715
724
        super(TestBadProtocolServer, self).setUp()
759
768
        self.assertEqual(None, server.host)
760
769
        self.assertEqual(None, server.port)
761
770
 
762
 
    def test_setUp_and_stop(self):
 
771
    def test_setUp_and_tearDown(self):
763
772
        server = RecordingServer(expect_body_tail=None)
764
 
        server.start_server()
 
773
        server.setUp()
765
774
        try:
766
775
            self.assertNotEqual(None, server.host)
767
776
            self.assertNotEqual(None, server.port)
768
777
        finally:
769
 
            server.stop_server()
 
778
            server.tearDown()
770
779
        self.assertEqual(None, server.host)
771
780
        self.assertEqual(None, server.port)
772
781
 
1169
1178
        if self._testing_pycurl():
1170
1179
            # Oh my ! pycurl does not check for the port as part of
1171
1180
            # no_proxy :-( So we just test the host part
1172
 
            self.no_proxy_host = self.server.host
 
1181
            self.no_proxy_host = 'localhost'
1173
1182
        else:
1174
1183
            self.no_proxy_host = self.proxy_address
1175
1184
        # The secondary server is the proxy
1178
1187
        self._old_env = {}
1179
1188
 
1180
1189
    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)
 
1190
        return pycurl_present and self._transport == PyCurlTransport
1184
1191
 
1185
1192
    def create_transport_secondary_server(self):
1186
1193
        """Creates an http server that will serve files with
1359
1366
        self.follow_redirections = True
1360
1367
 
1361
1368
 
1362
 
def install_redirected_request(test):
1363
 
    test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1364
 
 
1365
 
 
1366
1369
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1367
1370
    """Test redirections.
1368
1371
 
1378
1381
    """
1379
1382
 
1380
1383
    def setUp(self):
1381
 
        if (features.pycurl.available()
1382
 
            and self._transport == PyCurlTransport):
 
1384
        if pycurl_present and self._transport == PyCurlTransport:
1383
1385
            raise tests.TestNotApplicable(
1384
1386
                "pycurl doesn't redirect silently annymore")
1385
1387
        super(TestHTTPSilentRedirections, self).setUp()
1386
 
        install_redirected_request(self)
 
1388
        self.setup_redirected_request()
 
1389
        self.addCleanup(self.cleanup_redirected_request)
1387
1390
        self.build_tree_contents([('a','a'),
1388
1391
                                  ('1/',),
1389
1392
                                  ('1/a', 'redirected once'),
1399
1402
 
1400
1403
        self.old_transport = self._transport(self.old_server.get_url())
1401
1404
 
 
1405
    def setup_redirected_request(self):
 
1406
        self.original_class = _urllib2_wrappers.Request
 
1407
        _urllib2_wrappers.Request = RedirectedRequest
 
1408
 
 
1409
    def cleanup_redirected_request(self):
 
1410
        _urllib2_wrappers.Request = self.original_class
 
1411
 
1402
1412
    def create_transport_secondary_server(self):
1403
1413
        """Create the secondary server, redirections are defined in the tests"""
1404
1414
        return http_utils.HTTPServerRedirecting(
1408
1418
        t = self.old_transport
1409
1419
 
1410
1420
        req = RedirectedRequest('GET', t.abspath('a'))
 
1421
        req.follow_redirections = True
1411
1422
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1412
1423
                                       self.new_server.port)
1413
1424
        self.old_server.redirections = \
1414
1425
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1415
 
        self.assertEqual('redirected once',t._perform(req).read())
 
1426
        self.assertEquals('redirected once',t._perform(req).read())
1416
1427
 
1417
1428
    def test_five_redirections(self):
1418
1429
        t = self.old_transport
1419
1430
 
1420
1431
        req = RedirectedRequest('GET', t.abspath('a'))
 
1432
        req.follow_redirections = True
1421
1433
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1422
1434
                                       self.old_server.port)
1423
1435
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1429
1441
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1430
1442
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1431
1443
            ]
1432
 
        self.assertEqual('redirected 5 times',t._perform(req).read())
 
1444
        self.assertEquals('redirected 5 times',t._perform(req).read())
1433
1445
 
1434
1446
 
1435
1447
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1448
1460
        t = self._transport(self.new_server.get_url())
1449
1461
 
1450
1462
        # We use None for redirected so that we fail if redirected
1451
 
        self.assertEqual('0123456789',
1452
 
                         transport.do_catching_redirections(
 
1463
        self.assertEquals('0123456789',
 
1464
                          transport.do_catching_redirections(
1453
1465
                self.get_a, t, None).read())
1454
1466
 
1455
1467
    def test_one_redirection(self):
1460
1472
            dir, file = urlutils.split(exception.target)
1461
1473
            return self._transport(dir)
1462
1474
 
1463
 
        self.assertEqual('0123456789',
1464
 
                         transport.do_catching_redirections(
 
1475
        self.assertEquals('0123456789',
 
1476
                          transport.do_catching_redirections(
1465
1477
                self.get_a, self.old_transport, redirected).read())
1466
 
        self.assertEqual(1, self.redirections)
 
1478
        self.assertEquals(1, self.redirections)
1467
1479
 
1468
1480
    def test_redirection_loop(self):
1469
1481
 
1497
1509
        return self._auth_server(protocol_version=self._protocol_version)
1498
1510
 
1499
1511
    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)
 
1512
        return pycurl_present and self._transport == PyCurlTransport
1503
1513
 
1504
1514
    def get_user_url(self, user, password):
1505
1515
        """Build an url embedding user and password"""
1570
1580
        self.assertEqual('', ui.ui_factory.stdin.readline())
1571
1581
        stderr.seek(0)
1572
1582
        expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1573
 
        self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1574
 
        self.assertEqual('', stdout.getvalue())
 
1583
        self.assertEquals(expected_prompt, stderr.read(len(expected_prompt)))
 
1584
        self.assertEquals('', stdout.getvalue())
1575
1585
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1576
1586
                                    stderr.readline())
1577
1587
 
1592
1602
        self.assertEqual('', ui.ui_factory.stdin.readline())
1593
1603
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1594
1604
                                    stderr.getvalue())
1595
 
        self.assertEqual('', stdout.getvalue())
 
1605
        self.assertEquals('', stdout.getvalue())
1596
1606
        # And we shouldn't prompt again for a different request
1597
1607
        # against the same transport.
1598
1608
        self.assertEqual('contents of b\n',t.get('b').read())
1608
1618
                              % (scheme.upper(),
1609
1619
                                 user, self.server.host, self.server.port,
1610
1620
                                 self.server.auth_realm)))
1611
 
        self.assertEqual(expected_prompt, actual_prompt)
 
1621
        self.assertEquals(expected_prompt, actual_prompt)
1612
1622
 
1613
1623
    def _expected_username_prompt(self, scheme):
1614
1624
        return (self._username_prompt_prefix
1628
1638
        self.server.add_user(user, password)
1629
1639
        t = self.get_user_transport(user, None)
1630
1640
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1631
 
                                            stderr=tests.StringIOWrapper())
 
1641
                                            stdout=tests.StringIOWrapper())
1632
1642
        # Create a minimal config file with the right password
1633
1643
        conf = config.AuthenticationConfig()
1634
1644
        conf._get_config().update(
1846
1856
                             'http://www.example.com/foo/subdir')
1847
1857
        self.assertIsInstance(r, type(t))
1848
1858
        # Both transports share the some connection
1849
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1859
        self.assertEquals(t._get_connection(), r._get_connection())
1850
1860
 
1851
1861
    def test_redirected_to_self_with_slash(self):
1852
1862
        t = self._transport('http://www.example.com/foo')
1856
1866
        # Both transports share the some connection (one can argue that we
1857
1867
        # should return the exact same transport here, but that seems
1858
1868
        # overkill).
1859
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1869
        self.assertEquals(t._get_connection(), r._get_connection())
1860
1870
 
1861
1871
    def test_redirected_to_host(self):
1862
1872
        t = self._transport('http://www.example.com/foo')
1881
1891
        r = t._redirected_to('http://www.example.com/foo',
1882
1892
                             'https://foo.example.com/foo')
1883
1893
        self.assertIsInstance(r, type(t))
1884
 
        self.assertEqual(t._user, r._user)
 
1894
        self.assertEquals(t._user, r._user)
1885
1895
 
1886
1896
 
1887
1897
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1956
1966
    def setUp(self):
1957
1967
        tests.TestCase.setUp(self)
1958
1968
        self.server = self._activity_server(self._protocol_version)
1959
 
        self.server.start_server()
 
1969
        self.server.setUp()
1960
1970
        self.activities = {}
1961
1971
        def report_activity(t, bytes, direction):
1962
1972
            count = self.activities.get(direction, 0)
1971
1981
 
1972
1982
    def tearDown(self):
1973
1983
        self._transport._report_activity = self.orig_report_activity
1974
 
        self.server.stop_server()
 
1984
        self.server.tearDown()
1975
1985
        tests.TestCase.tearDown(self)
1976
1986
 
1977
1987
    def get_transport(self):
2097
2107
    def setUp(self):
2098
2108
        tests.TestCase.setUp(self)
2099
2109
        self.server = self._activity_server(self._protocol_version)
2100
 
        self.server.start_server()
 
2110
        self.server.setUp()
2101
2111
        self.activities = {}
2102
2112
        def report_activity(t, bytes, direction):
2103
2113
            count = self.activities.get(direction, 0)
2112
2122
 
2113
2123
    def tearDown(self):
2114
2124
        self._transport._report_activity = self.orig_report_activity
2115
 
        self.server.stop_server()
 
2125
        self.server.tearDown()
2116
2126
        tests.TestCase.tearDown(self)
2117
2127
 
2118
2128
 
2128
2138
        self.server = ActivityHTTPServer('HTTP/1.1')
2129
2139
        self._transport=_urllib.HttpTransport_urllib
2130
2140
 
2131
 
        self.server.start_server()
 
2141
        self.server.setUp()
2132
2142
 
2133
2143
        # We override at class level because constructors may propagate the
2134
2144
        # bound method and render instance overriding ineffective (an
2138
2148
 
2139
2149
    def tearDown(self):
2140
2150
        self._transport._report_activity = self.orig_report_activity
2141
 
        self.server.stop_server()
 
2151
        self.server.tearDown()
2142
2152
        tests.TestCase.tearDown(self)
2143
2153
 
2144
2154
    def assertActivitiesMatch(self):
2145
2155
        # Nothing to check here
2146
2156
        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