~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Vincent Ladeuil
  • Date: 2009-12-03 17:11:57 UTC
  • mto: (4862.2.2 integration2)
  • mto: This revision was merged to the branch mainline in revision 4866.
  • Revision ID: v.ladeuil+lp@free.fr-20091203171157-2lo1a9ej0fx3743b
Make sure all redirection code paths can handle authentication.

* bzrlib/transport/http/_urllib2_wrappers.py:
(AbstractAuthHandler.auth_required): Provide the necessary
attributes if the caller didn't.

* bzrlib/tests/test_http.py:
(install_redirected_request): Factor out the helper.
(TestHTTPSilentRedirections.setUp): Use the above helper.
(TestHTTPSilentRedirections.test_one_redirection,
TestHTTPSilentRedirections.test_five_redirections): Delete the
useless (hence misleading) code.
(TestAuth.test_no_prompt_for_password_when_using_auth_config):
Drive-by fix prompt go to stderr, stdout is not needed here.
(TestAuthOnRedirected): Ensure we handle authentication on both
code paths.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1366
1366
        self.follow_redirections = True
1367
1367
 
1368
1368
 
 
1369
def install_redirected_request(test):
 
1370
    test.original_class = _urllib2_wrappers.Request
 
1371
    def restore():
 
1372
        _urllib2_wrappers.Request = test.original_class
 
1373
    _urllib2_wrappers.Request = RedirectedRequest
 
1374
    test.addCleanup(restore)
 
1375
 
 
1376
 
1369
1377
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1370
1378
    """Test redirections.
1371
1379
 
1385
1393
            raise tests.TestNotApplicable(
1386
1394
                "pycurl doesn't redirect silently annymore")
1387
1395
        super(TestHTTPSilentRedirections, self).setUp()
1388
 
        self.setup_redirected_request()
1389
 
        self.addCleanup(self.cleanup_redirected_request)
 
1396
        install_redirected_request(self)
1390
1397
        self.build_tree_contents([('a','a'),
1391
1398
                                  ('1/',),
1392
1399
                                  ('1/a', 'redirected once'),
1402
1409
 
1403
1410
        self.old_transport = self._transport(self.old_server.get_url())
1404
1411
 
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
 
 
1412
1412
    def create_transport_secondary_server(self):
1413
1413
        """Create the secondary server, redirections are defined in the tests"""
1414
1414
        return http_utils.HTTPServerRedirecting(
1418
1418
        t = self.old_transport
1419
1419
 
1420
1420
        req = RedirectedRequest('GET', t.abspath('a'))
1421
 
        req.follow_redirections = True
1422
1421
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1423
1422
                                       self.new_server.port)
1424
1423
        self.old_server.redirections = \
1429
1428
        t = self.old_transport
1430
1429
 
1431
1430
        req = RedirectedRequest('GET', t.abspath('a'))
1432
 
        req.follow_redirections = True
1433
1431
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1434
1432
                                       self.old_server.port)
1435
1433
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1638
1636
        self.server.add_user(user, password)
1639
1637
        t = self.get_user_transport(user, None)
1640
1638
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1641
 
                                            stdout=tests.StringIOWrapper())
 
1639
                                            stderr=tests.StringIOWrapper())
1642
1640
        # Create a minimal config file with the right password
1643
1641
        conf = config.AuthenticationConfig()
1644
1642
        conf._get_config().update(
2154
2152
    def assertActivitiesMatch(self):
2155
2153
        # Nothing to check here
2156
2154
        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.assertEquals('redirected once',
 
2206
                          transport.do_catching_redirections(
 
2207
                self.get_a, self.old_transport, redirected).read())
 
2208
        self.assertEquals(1, self.redirections)
 
2209
        # stdin should be empty
 
2210
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
2211
        self.assertEquals('', stdout.getvalue())
 
2212
 
 
2213
    def test_auth_on_redirected_via_following_redirections(self):
 
2214
        self.new_server.add_user('joe', 'foo')
 
2215
        stdout = tests.StringIOWrapper()
 
2216
        stderr = tests.StringIOWrapper()
 
2217
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
 
2218
                                            stdout=stdout, stderr=stderr)
 
2219
        t = self.old_transport
 
2220
        req = RedirectedRequest('GET', t.abspath('a'))
 
2221
        new_prefix = 'http://%s:%s' % (self.new_server.host,
 
2222
                                       self.new_server.port)
 
2223
        self.old_server.redirections = [
 
2224
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
 
2225
        self.assertEquals('redirected once',t._perform(req).read())
 
2226
        # stdin should be empty
 
2227
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
2228
        self.assertEquals('', stdout.getvalue())
 
2229
 
 
2230