← 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: 2011-04-18 04:55:00 UTC
  • mfrom: (5784.2.1 754188-apport-test)
  • Revision ID: pqm@pqm.ubuntu.com-20110418045500-ce6lkgyiq7f47q43
(mbp) Rewrite test_report_bug_legacy away from using doctest (see bug
 764188) (Martin Pool)

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-2011 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
32
32
import bzrlib
33
33
from bzrlib import (
34
34
    bzrdir,
 
35
    cethread,
35
36
    config,
36
37
    errors,
37
38
    osutils,
178
179
        self._sock.bind(('127.0.0.1', 0))
179
180
        self.host, self.port = self._sock.getsockname()
180
181
        self._ready = threading.Event()
181
 
        self._thread = test_server.ThreadWithException(
182
 
            event=self._ready, target=self._accept_read_and_reply)
 
182
        self._thread = test_server.TestThread(
 
183
            sync_event=self._ready, target=self._accept_read_and_reply)
183
184
        self._thread.start()
184
185
        if 'threads' in tests.selftest_debug_flags:
185
186
            sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
254
255
        self.assertEqual('realm="Thou should not pass"', remainder)
255
256
 
256
257
 
 
258
class TestHTTPRangeParsing(tests.TestCase):
 
259
 
 
260
    def setUp(self):
 
261
        super(TestHTTPRangeParsing, self).setUp()
 
262
        # We focus on range  parsing here and ignore everything else
 
263
        class RequestHandler(http_server.TestingHTTPRequestHandler):
 
264
            def setup(self): pass
 
265
            def handle(self): pass
 
266
            def finish(self): pass
 
267
 
 
268
        self.req_handler = RequestHandler(None, None, None)
 
269
 
 
270
    def assertRanges(self, ranges, header, file_size):
 
271
        self.assertEquals(ranges,
 
272
                          self.req_handler._parse_ranges(header, file_size))
 
273
 
 
274
    def test_simple_range(self):
 
275
        self.assertRanges([(0,2)], 'bytes=0-2', 12)
 
276
 
 
277
    def test_tail(self):
 
278
        self.assertRanges([(8, 11)], 'bytes=-4', 12)
 
279
 
 
280
    def test_tail_bigger_than_file(self):
 
281
        self.assertRanges([(0, 11)], 'bytes=-99', 12)
 
282
 
 
283
    def test_range_without_end(self):
 
284
        self.assertRanges([(4, 11)], 'bytes=4-', 12)
 
285
 
 
286
    def test_invalid_ranges(self):
 
287
        self.assertRanges(None, 'bytes=12-22', 12)
 
288
        self.assertRanges(None, 'bytes=1-3,12-22', 12)
 
289
        self.assertRanges(None, 'bytes=-', 12)
 
290
 
 
291
 
257
292
class TestHTTPServer(tests.TestCase):
258
293
    """Test the HTTP servers implementations."""
259
294
 
354
389
 
355
390
    def test_abs_url(self):
356
391
        """Construction of absolute http URLs"""
357
 
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
 
392
        t = self._transport('http://example.com/bzr/bzr.dev/')
358
393
        eq = self.assertEqualDiff
359
 
        eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
360
 
        eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
361
 
        eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
 
394
        eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
 
395
        eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
 
396
        eq(t.abspath('.bzr'), 'http://example.com/bzr/bzr.dev/.bzr')
362
397
        eq(t.abspath('.bzr/1//2/./3'),
363
 
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
 
398
           'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
364
399
 
365
400
    def test_invalid_http_urls(self):
366
401
        """Trap invalid construction of urls"""
367
 
        self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
 
402
        self._transport('http://example.com/bzr/bzr.dev/')
368
403
        self.assertRaises(errors.InvalidURL,
369
404
                          self._transport,
370
 
                          'http://http://bazaar-vcs.org/bzr/bzr.dev/')
 
405
                          'http://http://example.com/bzr/bzr.dev/')
371
406
 
372
407
    def test_http_root_urls(self):
373
408
        """Construction of URLs from server root"""
374
 
        t = self._transport('http://bzr.ozlabs.org/')
 
409
        t = self._transport('http://example.com/')
375
410
        eq = self.assertEqualDiff
376
411
        eq(t.abspath('.bzr/tree-version'),
377
 
           'http://bzr.ozlabs.org/.bzr/tree-version')
 
412
           'http://example.com/.bzr/tree-version')
378
413
 
379
414
    def test_http_impl_urls(self):
380
415
        """There are servers which ask for particular clients to connect"""
427
462
    """Test the http connections."""
428
463
 
429
464
    scenarios = multiply_scenarios(
430
 
        vary_by_http_client_implementation(), 
 
465
        vary_by_http_client_implementation(),
431
466
        vary_by_http_protocol_version(),
432
467
        )
433
468
 
492
527
class TestPost(tests.TestCase):
493
528
 
494
529
    scenarios = multiply_scenarios(
495
 
        vary_by_http_client_implementation(), 
 
530
        vary_by_http_client_implementation(),
496
531
        vary_by_http_protocol_version(),
497
532
        )
498
533
 
551
586
    """
552
587
 
553
588
    scenarios = multiply_scenarios(
554
 
        vary_by_http_client_implementation(), 
 
589
        vary_by_http_client_implementation(),
555
590
        vary_by_http_protocol_version(),
556
591
        )
557
592
 
1029
1064
    """Tests readv requests against a server erroring out on too much ranges."""
1030
1065
 
1031
1066
    scenarios = multiply_scenarios(
1032
 
        vary_by_http_client_implementation(), 
 
1067
        vary_by_http_client_implementation(),
1033
1068
        vary_by_http_protocol_version(),
1034
1069
        )
1035
1070
 
1073
1108
    Only the urllib implementation is tested here.
1074
1109
    """
1075
1110
 
1076
 
    def setUp(self):
1077
 
        tests.TestCase.setUp(self)
1078
 
        self._old_env = {}
1079
 
        self.addCleanup(self._restore_env)
1080
 
 
1081
 
    def _install_env(self, env):
1082
 
        for name, value in env.iteritems():
1083
 
            self._old_env[name] = osutils.set_or_unset_env(name, value)
1084
 
 
1085
 
    def _restore_env(self):
1086
 
        for name, value in self._old_env.iteritems():
1087
 
            osutils.set_or_unset_env(name, value)
1088
 
 
1089
1111
    def _proxied_request(self):
1090
1112
        handler = _urllib2_wrappers.ProxyHandler()
1091
 
        request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
 
1113
        request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
1092
1114
        handler.set_proxy(request, 'http')
1093
1115
        return request
1094
1116
 
 
1117
    def assertEvaluateProxyBypass(self, expected, host, no_proxy):
 
1118
        handler = _urllib2_wrappers.ProxyHandler()
 
1119
        self.assertEquals(expected,
 
1120
                          handler.evaluate_proxy_bypass(host, no_proxy))
 
1121
 
1095
1122
    def test_empty_user(self):
1096
 
        self._install_env({'http_proxy': 'http://bar.com'})
 
1123
        self.overrideEnv('http_proxy', 'http://bar.com')
 
1124
        request = self._proxied_request()
 
1125
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
 
1126
 
 
1127
    def test_user_with_at(self):
 
1128
        self.overrideEnv('http_proxy',
 
1129
                         'http://username@domain:password@proxy_host:1234')
1097
1130
        request = self._proxied_request()
1098
1131
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
1099
1132
 
1100
1133
    def test_invalid_proxy(self):
1101
1134
        """A proxy env variable without scheme"""
1102
 
        self._install_env({'http_proxy': 'host:1234'})
 
1135
        self.overrideEnv('http_proxy', 'host:1234')
1103
1136
        self.assertRaises(errors.InvalidURL, self._proxied_request)
1104
1137
 
 
1138
    def test_evaluate_proxy_bypass_true(self):
 
1139
        """The host is not proxied"""
 
1140
        self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
 
1141
        self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
 
1142
 
 
1143
    def test_evaluate_proxy_bypass_false(self):
 
1144
        """The host is proxied"""
 
1145
        self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
 
1146
 
 
1147
    def test_evaluate_proxy_bypass_unknown(self):
 
1148
        """The host is not explicitly proxied"""
 
1149
        self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
 
1150
        self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
 
1151
 
 
1152
    def test_evaluate_proxy_bypass_empty_entries(self):
 
1153
        """Ignore empty entries"""
 
1154
        self.assertEvaluateProxyBypass(None, 'example.com', '')
 
1155
        self.assertEvaluateProxyBypass(None, 'example.com', ',')
 
1156
        self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
 
1157
 
1105
1158
 
1106
1159
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1107
1160
    """Tests proxy server.
1113
1166
    """
1114
1167
 
1115
1168
    scenarios = multiply_scenarios(
1116
 
        vary_by_http_client_implementation(), 
 
1169
        vary_by_http_client_implementation(),
1117
1170
        vary_by_http_protocol_version(),
1118
1171
        )
1119
1172
 
1136
1189
            self.no_proxy_host = self.server_host_port
1137
1190
        # The secondary server is the proxy
1138
1191
        self.proxy_url = self.get_secondary_url()
1139
 
        self._old_env = {}
1140
1192
 
1141
1193
    def _testing_pycurl(self):
1142
1194
        # TODO: This is duplicated for lots of the classes in this file
1143
1195
        return (features.pycurl.available()
1144
1196
                and self._transport == PyCurlTransport)
1145
1197
 
1146
 
    def _install_env(self, env):
1147
 
        for name, value in env.iteritems():
1148
 
            self._old_env[name] = osutils.set_or_unset_env(name, value)
1149
 
 
1150
 
    def _restore_env(self):
1151
 
        for name, value in self._old_env.iteritems():
1152
 
            osutils.set_or_unset_env(name, value)
1153
 
 
1154
 
    def proxied_in_env(self, env):
1155
 
        self._install_env(env)
1156
 
        t = self.get_readonly_transport()
1157
 
        try:
1158
 
            self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1159
 
        finally:
1160
 
            self._restore_env()
1161
 
 
1162
 
    def not_proxied_in_env(self, env):
1163
 
        self._install_env(env)
1164
 
        t = self.get_readonly_transport()
1165
 
        try:
1166
 
            self.assertEqual('contents of foo\n', t.get('foo').read())
1167
 
        finally:
1168
 
            self._restore_env()
 
1198
    def assertProxied(self):
 
1199
        t = self.get_readonly_transport()
 
1200
        self.assertEqual('proxied contents of foo\n', t.get('foo').read())
 
1201
 
 
1202
    def assertNotProxied(self):
 
1203
        t = self.get_readonly_transport()
 
1204
        self.assertEqual('contents of foo\n', t.get('foo').read())
1169
1205
 
1170
1206
    def test_http_proxy(self):
1171
 
        self.proxied_in_env({'http_proxy': self.proxy_url})
 
1207
        self.overrideEnv('http_proxy', self.proxy_url)
 
1208
        self.assertProxied()
1172
1209
 
1173
1210
    def test_HTTP_PROXY(self):
1174
1211
        if self._testing_pycurl():
1177
1214
            # about. Should we ?)
1178
1215
            raise tests.TestNotApplicable(
1179
1216
                'pycurl does not check HTTP_PROXY for security reasons')
1180
 
        self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
 
1217
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
 
1218
        self.assertProxied()
1181
1219
 
1182
1220
    def test_all_proxy(self):
1183
 
        self.proxied_in_env({'all_proxy': self.proxy_url})
 
1221
        self.overrideEnv('all_proxy', self.proxy_url)
 
1222
        self.assertProxied()
1184
1223
 
1185
1224
    def test_ALL_PROXY(self):
1186
 
        self.proxied_in_env({'ALL_PROXY': self.proxy_url})
 
1225
        self.overrideEnv('ALL_PROXY', self.proxy_url)
 
1226
        self.assertProxied()
1187
1227
 
1188
1228
    def test_http_proxy_with_no_proxy(self):
1189
 
        self.not_proxied_in_env({'http_proxy': self.proxy_url,
1190
 
                                 'no_proxy': self.no_proxy_host})
 
1229
        self.overrideEnv('no_proxy', self.no_proxy_host)
 
1230
        self.overrideEnv('http_proxy', self.proxy_url)
 
1231
        self.assertNotProxied()
1191
1232
 
1192
1233
    def test_HTTP_PROXY_with_NO_PROXY(self):
1193
1234
        if self._testing_pycurl():
1194
1235
            raise tests.TestNotApplicable(
1195
1236
                'pycurl does not check HTTP_PROXY for security reasons')
1196
 
        self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
1197
 
                                 'NO_PROXY': self.no_proxy_host})
 
1237
        self.overrideEnv('NO_PROXY', self.no_proxy_host)
 
1238
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
 
1239
        self.assertNotProxied()
1198
1240
 
1199
1241
    def test_all_proxy_with_no_proxy(self):
1200
 
        self.not_proxied_in_env({'all_proxy': self.proxy_url,
1201
 
                                 'no_proxy': self.no_proxy_host})
 
1242
        self.overrideEnv('no_proxy', self.no_proxy_host)
 
1243
        self.overrideEnv('all_proxy', self.proxy_url)
 
1244
        self.assertNotProxied()
1202
1245
 
1203
1246
    def test_ALL_PROXY_with_NO_PROXY(self):
1204
 
        self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
1205
 
                                 'NO_PROXY': self.no_proxy_host})
 
1247
        self.overrideEnv('NO_PROXY', self.no_proxy_host)
 
1248
        self.overrideEnv('ALL_PROXY', self.proxy_url)
 
1249
        self.assertNotProxied()
1206
1250
 
1207
1251
    def test_http_proxy_without_scheme(self):
 
1252
        self.overrideEnv('http_proxy', self.server_host_port)
1208
1253
        if self._testing_pycurl():
1209
1254
            # pycurl *ignores* invalid proxy env variables. If that ever change
1210
1255
            # in the future, this test will fail indicating that pycurl do not
1211
1256
            # ignore anymore such variables.
1212
 
            self.not_proxied_in_env({'http_proxy': self.server_host_port})
 
1257
            self.assertNotProxied()
1213
1258
        else:
1214
 
            self.assertRaises(errors.InvalidURL,
1215
 
                              self.proxied_in_env,
1216
 
                              {'http_proxy': self.server_host_port})
 
1259
            self.assertRaises(errors.InvalidURL, self.assertProxied)
1217
1260
 
1218
1261
 
1219
1262
class TestRanges(http_utils.TestCaseWithWebserver):
1220
1263
    """Test the Range header in GET methods."""
1221
1264
 
1222
1265
    scenarios = multiply_scenarios(
1223
 
        vary_by_http_client_implementation(), 
 
1266
        vary_by_http_client_implementation(),
1224
1267
        vary_by_http_protocol_version(),
1225
1268
        )
1226
1269
 
1270
1313
    """Test redirection between http servers."""
1271
1314
 
1272
1315
    scenarios = multiply_scenarios(
1273
 
        vary_by_http_client_implementation(), 
 
1316
        vary_by_http_client_implementation(),
1274
1317
        vary_by_http_protocol_version(),
1275
1318
        )
1276
1319
 
1343
1386
    """
1344
1387
 
1345
1388
    scenarios = multiply_scenarios(
1346
 
        vary_by_http_client_implementation(), 
 
1389
        vary_by_http_client_implementation(),
1347
1390
        vary_by_http_protocol_version(),
1348
1391
        )
1349
1392
 
1398
1441
    """Test transport.do_catching_redirections."""
1399
1442
 
1400
1443
    scenarios = multiply_scenarios(
1401
 
        vary_by_http_client_implementation(), 
 
1444
        vary_by_http_client_implementation(),
1402
1445
        vary_by_http_protocol_version(),
1403
1446
        )
1404
1447
 
1707
1750
 
1708
1751
    def setUp(self):
1709
1752
        super(TestProxyAuth, self).setUp()
1710
 
        self._old_env = {}
1711
 
        self.addCleanup(self._restore_env)
1712
1753
        # Override the contents to avoid false positives
1713
1754
        self.build_tree_contents([('a', 'not proxied contents of a\n'),
1714
1755
                                  ('b', 'not proxied contents of b\n'),
1717
1758
                                  ])
1718
1759
 
1719
1760
    def get_user_transport(self, user, password):
1720
 
        self._install_env({'all_proxy': self.get_user_url(user, password)})
 
1761
        self.overrideEnv('all_proxy', self.get_user_url(user, password))
1721
1762
        return TestAuth.get_user_transport(self, user, password)
1722
1763
 
1723
 
    def _install_env(self, env):
1724
 
        for name, value in env.iteritems():
1725
 
            self._old_env[name] = osutils.set_or_unset_env(name, value)
1726
 
 
1727
 
    def _restore_env(self):
1728
 
        for name, value in self._old_env.iteritems():
1729
 
            osutils.set_or_unset_env(name, value)
1730
 
 
1731
1764
    def test_empty_pass(self):
1732
1765
        if self._testing_pycurl():
1733
1766
            import pycurl
1763
1796
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1764
1797
 
1765
1798
    scenarios = multiply_scenarios(
1766
 
        vary_by_http_client_implementation(), 
 
1799
        vary_by_http_client_implementation(),
1767
1800
        vary_by_http_protocol_version(),
1768
1801
        )
1769
1802
 
1770
1803
    def setUp(self):
1771
1804
        super(SmartHTTPTunnellingTest, self).setUp()
1772
1805
        # We use the VFS layer as part of HTTP tunnelling tests.
1773
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1806
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1774
1807
        self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1775
1808
        self.http_server = self.get_readonly_server()
1776
1809