← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-11-17 03:20:35 UTC
  • mfrom: (4792.4.3 456036)
  • Revision ID: pqm@pqm.ubuntu.com-20091117032035-s3sgtlixj1lrminn
(Gordon Tyler) Fix IndexError during 'bzr ignore /' (#456036)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
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
399
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
395
 
        self.assertEqual('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
396
 
        self.assertEqual(1, len(f.credentials))
397
 
        self.assertEqual([None, 'www.bazaar-vcs.org', 'user', 'pass'],
398
 
                         f.credentials[0])
 
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
461
        version_info_orig = pycurl.version_info
456
 
        def restore():
 
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
457
483
            pycurl.version_info = version_info_orig
458
 
        self.addCleanup(restore)
459
 
 
460
 
        # Fake the pycurl version_info This was taken from a windows pycurl
461
 
        # without SSL (thanks to bialix)
462
 
        pycurl.version_info = lambda : (2,
463
 
                                        '7.13.2',
464
 
                                        462082,
465
 
                                        'i386-pc-win32',
466
 
                                        2576,
467
 
                                        None,
468
 
                                        0,
469
 
                                        None,
470
 
                                        ('ftp', 'gopher', 'telnet',
471
 
                                         'dict', 'ldap', 'http', 'file'),
472
 
                                        None,
473
 
                                        0,
474
 
                                        None)
475
 
        self.assertRaises(errors.DependencyNotPresent, self._transport,
476
 
                          'https://launchpad.net')
477
484
 
478
485
 
479
486
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
596
603
                                      protocol_version=self._protocol_version)
597
604
 
598
605
    def _testing_pycurl(self):
599
 
        # TODO: This is duplicated for lots of the classes in this file
600
 
        return (features.pycurl.available()
601
 
                and self._transport == PyCurlTransport)
 
606
        return pycurl_present and self._transport == PyCurlTransport
602
607
 
603
608
 
604
609
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
713
718
    _req_handler_class = BadProtocolRequestHandler
714
719
 
715
720
    def setUp(self):
716
 
        if self._testing_pycurl():
 
721
        if pycurl_present and self._transport == PyCurlTransport:
717
722
            raise tests.TestNotApplicable(
718
723
                "pycurl doesn't check the protocol version")
719
724
        super(TestBadProtocolServer, self).setUp()
763
768
        self.assertEqual(None, server.host)
764
769
        self.assertEqual(None, server.port)
765
770
 
766
 
    def test_setUp_and_stop(self):
 
771
    def test_setUp_and_tearDown(self):
767
772
        server = RecordingServer(expect_body_tail=None)
768
 
        server.start_server()
 
773
        server.setUp()
769
774
        try:
770
775
            self.assertNotEqual(None, server.host)
771
776
            self.assertNotEqual(None, server.port)
772
777
        finally:
773
 
            server.stop_server()
 
778
            server.tearDown()
774
779
        self.assertEqual(None, server.host)
775
780
        self.assertEqual(None, server.port)
776
781
 
1173
1178
        if self._testing_pycurl():
1174
1179
            # Oh my ! pycurl does not check for the port as part of
1175
1180
            # no_proxy :-( So we just test the host part
1176
 
            self.no_proxy_host = self.server.host
 
1181
            self.no_proxy_host = 'localhost'
1177
1182
        else:
1178
1183
            self.no_proxy_host = self.proxy_address
1179
1184
        # The secondary server is the proxy
1182
1187
        self._old_env = {}
1183
1188
 
1184
1189
    def _testing_pycurl(self):
1185
 
        # TODO: This is duplicated for lots of the classes in this file
1186
 
        return (features.pycurl.available()
1187
 
                and self._transport == PyCurlTransport)
 
1190
        return pycurl_present and self._transport == PyCurlTransport
1188
1191
 
1189
1192
    def create_transport_secondary_server(self):
1190
1193
        """Creates an http server that will serve files with
1363
1366
        self.follow_redirections = True
1364
1367
 
1365
1368
 
1366
 
def install_redirected_request(test):
1367
 
    test.original_class = _urllib2_wrappers.Request
1368
 
    def restore():
1369
 
        _urllib2_wrappers.Request = test.original_class
1370
 
    _urllib2_wrappers.Request = RedirectedRequest
1371
 
    test.addCleanup(restore)
1372
 
 
1373
 
 
1374
1369
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1375
1370
    """Test redirections.
1376
1371
 
1386
1381
    """
1387
1382
 
1388
1383
    def setUp(self):
1389
 
        if (features.pycurl.available()
1390
 
            and self._transport == PyCurlTransport):
 
1384
        if pycurl_present and self._transport == PyCurlTransport:
1391
1385
            raise tests.TestNotApplicable(
1392
1386
                "pycurl doesn't redirect silently annymore")
1393
1387
        super(TestHTTPSilentRedirections, self).setUp()
1394
 
        install_redirected_request(self)
 
1388
        self.setup_redirected_request()
 
1389
        self.addCleanup(self.cleanup_redirected_request)
1395
1390
        self.build_tree_contents([('a','a'),
1396
1391
                                  ('1/',),
1397
1392
                                  ('1/a', 'redirected once'),
1407
1402
 
1408
1403
        self.old_transport = self._transport(self.old_server.get_url())
1409
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
 
1410
1412
    def create_transport_secondary_server(self):
1411
1413
        """Create the secondary server, redirections are defined in the tests"""
1412
1414
        return http_utils.HTTPServerRedirecting(
1416
1418
        t = self.old_transport
1417
1419
 
1418
1420
        req = RedirectedRequest('GET', t.abspath('a'))
 
1421
        req.follow_redirections = True
1419
1422
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1420
1423
                                       self.new_server.port)
1421
1424
        self.old_server.redirections = \
1422
1425
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1423
 
        self.assertEqual('redirected once',t._perform(req).read())
 
1426
        self.assertEquals('redirected once',t._perform(req).read())
1424
1427
 
1425
1428
    def test_five_redirections(self):
1426
1429
        t = self.old_transport
1427
1430
 
1428
1431
        req = RedirectedRequest('GET', t.abspath('a'))
 
1432
        req.follow_redirections = True
1429
1433
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1430
1434
                                       self.old_server.port)
1431
1435
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1437
1441
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1438
1442
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1439
1443
            ]
1440
 
        self.assertEqual('redirected 5 times',t._perform(req).read())
 
1444
        self.assertEquals('redirected 5 times',t._perform(req).read())
1441
1445
 
1442
1446
 
1443
1447
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1456
1460
        t = self._transport(self.new_server.get_url())
1457
1461
 
1458
1462
        # We use None for redirected so that we fail if redirected
1459
 
        self.assertEqual('0123456789',
1460
 
                         transport.do_catching_redirections(
 
1463
        self.assertEquals('0123456789',
 
1464
                          transport.do_catching_redirections(
1461
1465
                self.get_a, t, None).read())
1462
1466
 
1463
1467
    def test_one_redirection(self):
1468
1472
            dir, file = urlutils.split(exception.target)
1469
1473
            return self._transport(dir)
1470
1474
 
1471
 
        self.assertEqual('0123456789',
1472
 
                         transport.do_catching_redirections(
 
1475
        self.assertEquals('0123456789',
 
1476
                          transport.do_catching_redirections(
1473
1477
                self.get_a, self.old_transport, redirected).read())
1474
 
        self.assertEqual(1, self.redirections)
 
1478
        self.assertEquals(1, self.redirections)
1475
1479
 
1476
1480
    def test_redirection_loop(self):
1477
1481
 
1505
1509
        return self._auth_server(protocol_version=self._protocol_version)
1506
1510
 
1507
1511
    def _testing_pycurl(self):
1508
 
        # TODO: This is duplicated for lots of the classes in this file
1509
 
        return (features.pycurl.available()
1510
 
                and self._transport == PyCurlTransport)
 
1512
        return pycurl_present and self._transport == PyCurlTransport
1511
1513
 
1512
1514
    def get_user_url(self, user, password):
1513
1515
        """Build an url embedding user and password"""
1578
1580
        self.assertEqual('', ui.ui_factory.stdin.readline())
1579
1581
        stderr.seek(0)
1580
1582
        expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1581
 
        self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1582
 
        self.assertEqual('', stdout.getvalue())
 
1583
        self.assertEquals(expected_prompt, stderr.read(len(expected_prompt)))
 
1584
        self.assertEquals('', stdout.getvalue())
1583
1585
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1584
1586
                                    stderr.readline())
1585
1587
 
1600
1602
        self.assertEqual('', ui.ui_factory.stdin.readline())
1601
1603
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1602
1604
                                    stderr.getvalue())
1603
 
        self.assertEqual('', stdout.getvalue())
 
1605
        self.assertEquals('', stdout.getvalue())
1604
1606
        # And we shouldn't prompt again for a different request
1605
1607
        # against the same transport.
1606
1608
        self.assertEqual('contents of b\n',t.get('b').read())
1616
1618
                              % (scheme.upper(),
1617
1619
                                 user, self.server.host, self.server.port,
1618
1620
                                 self.server.auth_realm)))
1619
 
        self.assertEqual(expected_prompt, actual_prompt)
 
1621
        self.assertEquals(expected_prompt, actual_prompt)
1620
1622
 
1621
1623
    def _expected_username_prompt(self, scheme):
1622
1624
        return (self._username_prompt_prefix
1636
1638
        self.server.add_user(user, password)
1637
1639
        t = self.get_user_transport(user, None)
1638
1640
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1639
 
                                            stderr=tests.StringIOWrapper())
 
1641
                                            stdout=tests.StringIOWrapper())
1640
1642
        # Create a minimal config file with the right password
1641
1643
        conf = config.AuthenticationConfig()
1642
1644
        conf._get_config().update(
1854
1856
                             'http://www.example.com/foo/subdir')
1855
1857
        self.assertIsInstance(r, type(t))
1856
1858
        # Both transports share the some connection
1857
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1859
        self.assertEquals(t._get_connection(), r._get_connection())
1858
1860
 
1859
1861
    def test_redirected_to_self_with_slash(self):
1860
1862
        t = self._transport('http://www.example.com/foo')
1864
1866
        # Both transports share the some connection (one can argue that we
1865
1867
        # should return the exact same transport here, but that seems
1866
1868
        # overkill).
1867
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1869
        self.assertEquals(t._get_connection(), r._get_connection())
1868
1870
 
1869
1871
    def test_redirected_to_host(self):
1870
1872
        t = self._transport('http://www.example.com/foo')
1889
1891
        r = t._redirected_to('http://www.example.com/foo',
1890
1892
                             'https://foo.example.com/foo')
1891
1893
        self.assertIsInstance(r, type(t))
1892
 
        self.assertEqual(t._user, r._user)
 
1894
        self.assertEquals(t._user, r._user)
1893
1895
 
1894
1896
 
1895
1897
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1964
1966
    def setUp(self):
1965
1967
        tests.TestCase.setUp(self)
1966
1968
        self.server = self._activity_server(self._protocol_version)
1967
 
        self.server.start_server()
 
1969
        self.server.setUp()
1968
1970
        self.activities = {}
1969
1971
        def report_activity(t, bytes, direction):
1970
1972
            count = self.activities.get(direction, 0)
1979
1981
 
1980
1982
    def tearDown(self):
1981
1983
        self._transport._report_activity = self.orig_report_activity
1982
 
        self.server.stop_server()
 
1984
        self.server.tearDown()
1983
1985
        tests.TestCase.tearDown(self)
1984
1986
 
1985
1987
    def get_transport(self):
2105
2107
    def setUp(self):
2106
2108
        tests.TestCase.setUp(self)
2107
2109
        self.server = self._activity_server(self._protocol_version)
2108
 
        self.server.start_server()
 
2110
        self.server.setUp()
2109
2111
        self.activities = {}
2110
2112
        def report_activity(t, bytes, direction):
2111
2113
            count = self.activities.get(direction, 0)
2120
2122
 
2121
2123
    def tearDown(self):
2122
2124
        self._transport._report_activity = self.orig_report_activity
2123
 
        self.server.stop_server()
 
2125
        self.server.tearDown()
2124
2126
        tests.TestCase.tearDown(self)
2125
2127
 
2126
2128
 
2136
2138
        self.server = ActivityHTTPServer('HTTP/1.1')
2137
2139
        self._transport=_urllib.HttpTransport_urllib
2138
2140
 
2139
 
        self.server.start_server()
 
2141
        self.server.setUp()
2140
2142
 
2141
2143
        # We override at class level because constructors may propagate the
2142
2144
        # bound method and render instance overriding ineffective (an
2146
2148
 
2147
2149
    def tearDown(self):
2148
2150
        self._transport._report_activity = self.orig_report_activity
2149
 
        self.server.stop_server()
 
2151
        self.server.tearDown()
2150
2152
        tests.TestCase.tearDown(self)
2151
2153
 
2152
2154
    def assertActivitiesMatch(self):
2153
2155
        # Nothing to check here
2154
2156
        pass
2155
 
 
2156
 
 
2157
 
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2158
 
    """Test authentication on the redirected http server."""
2159
 
 
2160
 
    _auth_header = 'Authorization'
2161
 
    _password_prompt_prefix = ''
2162
 
    _username_prompt_prefix = ''
2163
 
    _auth_server = http_utils.HTTPBasicAuthServer
2164
 
    _transport = _urllib.HttpTransport_urllib
2165
 
 
2166
 
    def create_transport_readonly_server(self):
2167
 
        return self._auth_server()
2168
 
 
2169
 
    def create_transport_secondary_server(self):
2170
 
        """Create the secondary server redirecting to the primary server"""
2171
 
        new = self.get_readonly_server()
2172
 
 
2173
 
        redirecting = http_utils.HTTPServerRedirecting()
2174
 
        redirecting.redirect_to(new.host, new.port)
2175
 
        return redirecting
2176
 
 
2177
 
    def setUp(self):
2178
 
        super(TestAuthOnRedirected, self).setUp()
2179
 
        self.build_tree_contents([('a','a'),
2180
 
                                  ('1/',),
2181
 
                                  ('1/a', 'redirected once'),
2182
 
                                  ],)
2183
 
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2184
 
                                       self.new_server.port)
2185
 
        self.old_server.redirections = [
2186
 
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
2187
 
        self.old_transport = self._transport(self.old_server.get_url())
2188
 
        self.new_server.add_user('joe', 'foo')
2189
 
 
2190
 
    def get_a(self, transport):
2191
 
        return transport.get('a')
2192
 
 
2193
 
    def test_auth_on_redirected_via_do_catching_redirections(self):
2194
 
        self.redirections = 0
2195
 
 
2196
 
        def redirected(transport, exception, redirection_notice):
2197
 
            self.redirections += 1
2198
 
            dir, file = urlutils.split(exception.target)
2199
 
            return self._transport(dir)
2200
 
 
2201
 
        stdout = tests.StringIOWrapper()
2202
 
        stderr = tests.StringIOWrapper()
2203
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2204
 
                                            stdout=stdout, stderr=stderr)
2205
 
        self.assertEqual('redirected once',
2206
 
                         transport.do_catching_redirections(
2207
 
                self.get_a, self.old_transport, redirected).read())
2208
 
        self.assertEqual(1, self.redirections)
2209
 
        # stdin should be empty
2210
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
2211
 
        # stdout should be empty, stderr will contains the prompts
2212
 
        self.assertEqual('', stdout.getvalue())
2213
 
 
2214
 
    def test_auth_on_redirected_via_following_redirections(self):
2215
 
        self.new_server.add_user('joe', 'foo')
2216
 
        stdout = tests.StringIOWrapper()
2217
 
        stderr = tests.StringIOWrapper()
2218
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2219
 
                                            stdout=stdout, stderr=stderr)
2220
 
        t = self.old_transport
2221
 
        req = RedirectedRequest('GET', t.abspath('a'))
2222
 
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2223
 
                                       self.new_server.port)
2224
 
        self.old_server.redirections = [
2225
 
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
2226
 
        self.assertEqual('redirected once',t._perform(req).read())
2227
 
        # stdin should be empty
2228
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
2229
 
        # stdout should be empty, stderr will contains the prompts
2230
 
        self.assertEqual('', stdout.getvalue())
2231
 
 
2232