/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: Canonical.com Patch Queue Manager
  • Date: 2009-12-21 06:03:07 UTC
  • mfrom: (4665.7.3 serve-init)
  • Revision ID: pqm@pqm.ubuntu.com-20091221060307-uvja3vdy1o6dzzy0
(mbp) example debian init script

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
204
204
    It records the bytes sent to it, and replies with a 200.
205
205
    """
206
206
 
207
 
    def __init__(self, expect_body_tail=None):
 
207
    def __init__(self, expect_body_tail=None, scheme=''):
208
208
        """Constructor.
209
209
 
210
210
        :type expect_body_tail: str
215
215
        self.host = None
216
216
        self.port = None
217
217
        self.received_bytes = ''
 
218
        self.scheme = scheme
 
219
 
 
220
    def get_url(self):
 
221
        return '%s://%s:%s/' % (self.scheme, self.host, self.port)
218
222
 
219
223
    def setUp(self):
220
224
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
265
269
 
266
270
    def test_empty_header(self):
267
271
        scheme, remainder = self.parse_header('')
268
 
        self.assertEquals('', scheme)
 
272
        self.assertEqual('', scheme)
269
273
        self.assertIs(None, remainder)
270
274
 
271
275
    def test_negotiate_header(self):
272
276
        scheme, remainder = self.parse_header('Negotiate')
273
 
        self.assertEquals('negotiate', scheme)
 
277
        self.assertEqual('negotiate', scheme)
274
278
        self.assertIs(None, remainder)
275
279
 
276
280
    def test_basic_header(self):
277
281
        scheme, remainder = self.parse_header(
278
282
            'Basic realm="Thou should not pass"')
279
 
        self.assertEquals('basic', scheme)
280
 
        self.assertEquals('realm="Thou should not pass"', remainder)
 
283
        self.assertEqual('basic', scheme)
 
284
        self.assertEqual('realm="Thou should not pass"', remainder)
281
285
 
282
286
    def test_basic_extract_realm(self):
283
287
        scheme, remainder = self.parse_header(
285
289
            _urllib2_wrappers.BasicAuthHandler)
286
290
        match, realm = self.auth_handler.extract_realm(remainder)
287
291
        self.assertTrue(match is not None)
288
 
        self.assertEquals('Thou should not pass', realm)
 
292
        self.assertEqual('Thou should not pass', realm)
289
293
 
290
294
    def test_digest_header(self):
291
295
        scheme, remainder = self.parse_header(
292
296
            'Digest realm="Thou should not pass"')
293
 
        self.assertEquals('digest', scheme)
294
 
        self.assertEquals('realm="Thou should not pass"', remainder)
 
297
        self.assertEqual('digest', scheme)
 
298
        self.assertEqual('realm="Thou should not pass"', remainder)
295
299
 
296
300
 
297
301
class TestHTTPServer(tests.TestCase):
389
393
    def test_url_parsing(self):
390
394
        f = FakeManager()
391
395
        url = http.extract_auth('http://example.com', f)
392
 
        self.assertEquals('http://example.com', url)
393
 
        self.assertEquals(0, len(f.credentials))
 
396
        self.assertEqual('http://example.com', url)
 
397
        self.assertEqual(0, len(f.credentials))
394
398
        url = http.extract_auth(
395
399
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
396
 
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
397
 
        self.assertEquals(1, len(f.credentials))
398
 
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
399
 
                          f.credentials[0])
 
400
        self.assertEqual('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
 
401
        self.assertEqual(1, len(f.credentials))
 
402
        self.assertEqual([None, 'www.bazaar-vcs.org', 'user', 'pass'],
 
403
                         f.credentials[0])
400
404
 
401
405
 
402
406
class TestHttpTransportUrls(tests.TestCase):
541
545
class TestPost(tests.TestCase):
542
546
 
543
547
    def test_post_body_is_received(self):
544
 
        server = RecordingServer(expect_body_tail='end-of-body')
 
548
        server = RecordingServer(expect_body_tail='end-of-body',
 
549
            scheme=self._qualified_prefix)
545
550
        self.start_server(server)
546
 
        scheme = self._qualified_prefix
547
 
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
 
551
        url = server.get_url()
548
552
        http_transport = self._transport(url)
549
553
        code, response = http_transport._post('abc def end-of-body')
550
554
        self.assertTrue(
776
780
        self.assertEqual(None, server.port)
777
781
 
778
782
    def test_send_receive_bytes(self):
779
 
        server = RecordingServer(expect_body_tail='c')
 
783
        server = RecordingServer(expect_body_tail='c', scheme='http')
780
784
        self.start_server(server)
781
785
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
782
786
        sock.connect((server.host, server.port))
1174
1178
        if self._testing_pycurl():
1175
1179
            # Oh my ! pycurl does not check for the port as part of
1176
1180
            # no_proxy :-( So we just test the host part
1177
 
            self.no_proxy_host = 'localhost'
 
1181
            self.no_proxy_host = self.server.host
1178
1182
        else:
1179
1183
            self.no_proxy_host = self.proxy_address
1180
1184
        # The secondary server is the proxy
1362
1366
        self.follow_redirections = True
1363
1367
 
1364
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
 
1365
1377
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1366
1378
    """Test redirections.
1367
1379
 
1381
1393
            raise tests.TestNotApplicable(
1382
1394
                "pycurl doesn't redirect silently annymore")
1383
1395
        super(TestHTTPSilentRedirections, self).setUp()
1384
 
        self.setup_redirected_request()
1385
 
        self.addCleanup(self.cleanup_redirected_request)
 
1396
        install_redirected_request(self)
1386
1397
        self.build_tree_contents([('a','a'),
1387
1398
                                  ('1/',),
1388
1399
                                  ('1/a', 'redirected once'),
1398
1409
 
1399
1410
        self.old_transport = self._transport(self.old_server.get_url())
1400
1411
 
1401
 
    def setup_redirected_request(self):
1402
 
        self.original_class = _urllib2_wrappers.Request
1403
 
        _urllib2_wrappers.Request = RedirectedRequest
1404
 
 
1405
 
    def cleanup_redirected_request(self):
1406
 
        _urllib2_wrappers.Request = self.original_class
1407
 
 
1408
1412
    def create_transport_secondary_server(self):
1409
1413
        """Create the secondary server, redirections are defined in the tests"""
1410
1414
        return http_utils.HTTPServerRedirecting(
1414
1418
        t = self.old_transport
1415
1419
 
1416
1420
        req = RedirectedRequest('GET', t.abspath('a'))
1417
 
        req.follow_redirections = True
1418
1421
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1419
1422
                                       self.new_server.port)
1420
1423
        self.old_server.redirections = \
1421
1424
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1422
 
        self.assertEquals('redirected once',t._perform(req).read())
 
1425
        self.assertEqual('redirected once',t._perform(req).read())
1423
1426
 
1424
1427
    def test_five_redirections(self):
1425
1428
        t = self.old_transport
1426
1429
 
1427
1430
        req = RedirectedRequest('GET', t.abspath('a'))
1428
 
        req.follow_redirections = True
1429
1431
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1430
1432
                                       self.old_server.port)
1431
1433
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1437
1439
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1438
1440
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1439
1441
            ]
1440
 
        self.assertEquals('redirected 5 times',t._perform(req).read())
 
1442
        self.assertEqual('redirected 5 times',t._perform(req).read())
1441
1443
 
1442
1444
 
1443
1445
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1456
1458
        t = self._transport(self.new_server.get_url())
1457
1459
 
1458
1460
        # We use None for redirected so that we fail if redirected
1459
 
        self.assertEquals('0123456789',
1460
 
                          transport.do_catching_redirections(
 
1461
        self.assertEqual('0123456789',
 
1462
                         transport.do_catching_redirections(
1461
1463
                self.get_a, t, None).read())
1462
1464
 
1463
1465
    def test_one_redirection(self):
1468
1470
            dir, file = urlutils.split(exception.target)
1469
1471
            return self._transport(dir)
1470
1472
 
1471
 
        self.assertEquals('0123456789',
1472
 
                          transport.do_catching_redirections(
 
1473
        self.assertEqual('0123456789',
 
1474
                         transport.do_catching_redirections(
1473
1475
                self.get_a, self.old_transport, redirected).read())
1474
 
        self.assertEquals(1, self.redirections)
 
1476
        self.assertEqual(1, self.redirections)
1475
1477
 
1476
1478
    def test_redirection_loop(self):
1477
1479
 
1576
1578
        self.assertEqual('', ui.ui_factory.stdin.readline())
1577
1579
        stderr.seek(0)
1578
1580
        expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1579
 
        self.assertEquals(expected_prompt, stderr.read(len(expected_prompt)))
1580
 
        self.assertEquals('', stdout.getvalue())
 
1581
        self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
 
1582
        self.assertEqual('', stdout.getvalue())
1581
1583
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1582
1584
                                    stderr.readline())
1583
1585
 
1598
1600
        self.assertEqual('', ui.ui_factory.stdin.readline())
1599
1601
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1600
1602
                                    stderr.getvalue())
1601
 
        self.assertEquals('', stdout.getvalue())
 
1603
        self.assertEqual('', stdout.getvalue())
1602
1604
        # And we shouldn't prompt again for a different request
1603
1605
        # against the same transport.
1604
1606
        self.assertEqual('contents of b\n',t.get('b').read())
1614
1616
                              % (scheme.upper(),
1615
1617
                                 user, self.server.host, self.server.port,
1616
1618
                                 self.server.auth_realm)))
1617
 
        self.assertEquals(expected_prompt, actual_prompt)
 
1619
        self.assertEqual(expected_prompt, actual_prompt)
1618
1620
 
1619
1621
    def _expected_username_prompt(self, scheme):
1620
1622
        return (self._username_prompt_prefix
1634
1636
        self.server.add_user(user, password)
1635
1637
        t = self.get_user_transport(user, None)
1636
1638
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1637
 
                                            stdout=tests.StringIOWrapper())
 
1639
                                            stderr=tests.StringIOWrapper())
1638
1640
        # Create a minimal config file with the right password
1639
1641
        conf = config.AuthenticationConfig()
1640
1642
        conf._get_config().update(
1852
1854
                             'http://www.example.com/foo/subdir')
1853
1855
        self.assertIsInstance(r, type(t))
1854
1856
        # Both transports share the some connection
1855
 
        self.assertEquals(t._get_connection(), r._get_connection())
 
1857
        self.assertEqual(t._get_connection(), r._get_connection())
1856
1858
 
1857
1859
    def test_redirected_to_self_with_slash(self):
1858
1860
        t = self._transport('http://www.example.com/foo')
1862
1864
        # Both transports share the some connection (one can argue that we
1863
1865
        # should return the exact same transport here, but that seems
1864
1866
        # overkill).
1865
 
        self.assertEquals(t._get_connection(), r._get_connection())
 
1867
        self.assertEqual(t._get_connection(), r._get_connection())
1866
1868
 
1867
1869
    def test_redirected_to_host(self):
1868
1870
        t = self._transport('http://www.example.com/foo')
1887
1889
        r = t._redirected_to('http://www.example.com/foo',
1888
1890
                             'https://foo.example.com/foo')
1889
1891
        self.assertIsInstance(r, type(t))
1890
 
        self.assertEquals(t._user, r._user)
 
1892
        self.assertEqual(t._user, r._user)
1891
1893
 
1892
1894
 
1893
1895
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1952
1954
        pass
1953
1955
 
1954
1956
 
1955
 
class TestActivity(tests.TestCase):
 
1957
class TestActivityMixin(object):
1956
1958
    """Test socket activity reporting.
1957
1959
 
1958
1960
    We use a special purpose server to control the bytes sent and received and
2096
2098
        code, f = t._post('abc def end-of-body\n')
2097
2099
        self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2098
2100
        self.assertActivitiesMatch()
 
2101
 
 
2102
 
 
2103
class TestActivity(tests.TestCase, TestActivityMixin):
 
2104
 
 
2105
    def setUp(self):
 
2106
        tests.TestCase.setUp(self)
 
2107
        self.server = self._activity_server(self._protocol_version)
 
2108
        self.server.setUp()
 
2109
        self.activities = {}
 
2110
        def report_activity(t, bytes, direction):
 
2111
            count = self.activities.get(direction, 0)
 
2112
            count += bytes
 
2113
            self.activities[direction] = count
 
2114
 
 
2115
        # We override at class level because constructors may propagate the
 
2116
        # bound method and render instance overriding ineffective (an
 
2117
        # alternative would be to define a specific ui factory instead...)
 
2118
        self.orig_report_activity = self._transport._report_activity
 
2119
        self._transport._report_activity = report_activity
 
2120
 
 
2121
    def tearDown(self):
 
2122
        self._transport._report_activity = self.orig_report_activity
 
2123
        self.server.tearDown()
 
2124
        tests.TestCase.tearDown(self)
 
2125
 
 
2126
 
 
2127
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
 
2128
 
 
2129
    def setUp(self):
 
2130
        tests.TestCase.setUp(self)
 
2131
        # Unlike TestActivity, we are really testing ReportingFileSocket and
 
2132
        # ReportingSocket, so we don't need all the parametrization. Since
 
2133
        # ReportingFileSocket and ReportingSocket are wrappers, it's easier to
 
2134
        # test them through their use by the transport than directly (that's a
 
2135
        # bit less clean but far more simpler and effective).
 
2136
        self.server = ActivityHTTPServer('HTTP/1.1')
 
2137
        self._transport=_urllib.HttpTransport_urllib
 
2138
 
 
2139
        self.server.setUp()
 
2140
 
 
2141
        # We override at class level because constructors may propagate the
 
2142
        # bound method and render instance overriding ineffective (an
 
2143
        # alternative would be to define a specific ui factory instead...)
 
2144
        self.orig_report_activity = self._transport._report_activity
 
2145
        self._transport._report_activity = None
 
2146
 
 
2147
    def tearDown(self):
 
2148
        self._transport._report_activity = self.orig_report_activity
 
2149
        self.server.tearDown()
 
2150
        tests.TestCase.tearDown(self)
 
2151
 
 
2152
    def assertActivitiesMatch(self):
 
2153
        # Nothing to check here
 
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.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