← 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: John Arbash Meinel
  • Date: 2010-11-05 20:54:32 UTC
  • mfrom: (5526 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5527.
  • Revision ID: john@arbash-meinel.com-20101105205432-rmyozu8sthyhmri8
Merge bzr.dev to resolve bzr-2.3.txt (aka NEWS)

Show diffs side-by-side

added added

removed removed

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