← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

merge trunk

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
23
23
# TODO: Should be renamed to bzrlib.transport.http.tests?
24
24
# TODO: What about renaming to bzrlib.tests.transport.http ?
25
25
 
26
 
from cStringIO import StringIO
27
26
import httplib
28
 
import os
29
 
import select
30
27
import SimpleHTTPServer
31
28
import socket
32
29
import sys
35
32
import bzrlib
36
33
from bzrlib import (
37
34
    bzrdir,
 
35
    cethread,
38
36
    config,
39
37
    errors,
40
38
    osutils,
42
40
    tests,
43
41
    transport,
44
42
    ui,
45
 
    urlutils,
46
43
    )
47
44
from bzrlib.tests import (
48
45
    features,
50
47
    http_utils,
51
48
    test_server,
52
49
    )
 
50
from bzrlib.tests.scenarios import (
 
51
    load_tests_apply_scenarios,
 
52
    multiply_scenarios,
 
53
    )
53
54
from bzrlib.transport import (
54
55
    http,
55
56
    remote,
64
65
    from bzrlib.transport.http._pycurl import PyCurlTransport
65
66
 
66
67
 
67
 
def load_tests(standard_tests, module, loader):
68
 
    """Multiply tests for http clients and protocol versions."""
69
 
    result = loader.suiteClass()
70
 
 
71
 
    # one for each transport implementation
72
 
    t_tests, remaining_tests = tests.split_suite_by_condition(
73
 
        standard_tests, tests.condition_isinstance((
74
 
                TestHttpTransportRegistration,
75
 
                TestHttpTransportUrls,
76
 
                Test_redirected_to,
77
 
                )))
 
68
load_tests = load_tests_apply_scenarios
 
69
 
 
70
 
 
71
def vary_by_http_client_implementation():
 
72
    """Test the two libraries we can use, pycurl and urllib."""
78
73
    transport_scenarios = [
79
74
        ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
80
75
                        _server=http_server.HttpServer_urllib,
85
80
            ('pycurl', dict(_transport=PyCurlTransport,
86
81
                            _server=http_server.HttpServer_PyCurl,
87
82
                            _url_protocol='http+pycurl',)))
88
 
    tests.multiply_tests(t_tests, transport_scenarios, result)
89
 
 
90
 
    protocol_scenarios = [
91
 
            ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
92
 
            ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
93
 
            ]
94
 
 
95
 
    # some tests are parametrized by the protocol version only
96
 
    p_tests, remaining_tests = tests.split_suite_by_condition(
97
 
        remaining_tests, tests.condition_isinstance((
98
 
                TestAuthOnRedirected,
99
 
                )))
100
 
    tests.multiply_tests(p_tests, protocol_scenarios, result)
101
 
 
102
 
    # each implementation tested with each HTTP version
103
 
    tp_tests, remaining_tests = tests.split_suite_by_condition(
104
 
        remaining_tests, tests.condition_isinstance((
105
 
                SmartHTTPTunnellingTest,
106
 
                TestDoCatchRedirections,
107
 
                TestHTTPConnections,
108
 
                TestHTTPRedirections,
109
 
                TestHTTPSilentRedirections,
110
 
                TestLimitedRangeRequestServer,
111
 
                TestPost,
112
 
                TestProxyHttpServer,
113
 
                TestRanges,
114
 
                TestSpecificRequestHandler,
115
 
                )))
116
 
    tp_scenarios = tests.multiply_scenarios(transport_scenarios,
117
 
                                            protocol_scenarios)
118
 
    tests.multiply_tests(tp_tests, tp_scenarios, result)
119
 
 
120
 
    # proxy auth: each auth scheme on all http versions on all implementations.
121
 
    tppa_tests, remaining_tests = tests.split_suite_by_condition(
122
 
        remaining_tests, tests.condition_isinstance((
123
 
                TestProxyAuth,
124
 
                )))
125
 
    proxy_auth_scheme_scenarios = [
 
83
    return transport_scenarios
 
84
 
 
85
 
 
86
def vary_by_http_protocol_version():
 
87
    """Test on http/1.0 and 1.1"""
 
88
    return [
 
89
        ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
 
90
        ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
 
91
        ]
 
92
 
 
93
 
 
94
def vary_by_http_proxy_auth_scheme():
 
95
    return [
126
96
        ('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
127
97
        ('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
128
98
        ('basicdigest',
129
 
         dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
 
99
            dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
130
100
        ]
131
 
    tppa_scenarios = tests.multiply_scenarios(tp_scenarios,
132
 
                                              proxy_auth_scheme_scenarios)
133
 
    tests.multiply_tests(tppa_tests, tppa_scenarios, result)
134
 
 
135
 
    # auth: each auth scheme on all http versions on all implementations.
136
 
    tpa_tests, remaining_tests = tests.split_suite_by_condition(
137
 
        remaining_tests, tests.condition_isinstance((
138
 
                TestAuth,
139
 
                )))
140
 
    auth_scheme_scenarios = [
 
101
 
 
102
 
 
103
def vary_by_http_auth_scheme():
 
104
    return [
141
105
        ('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
142
106
        ('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
143
107
        ('basicdigest',
144
 
         dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
 
108
            dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
145
109
        ]
146
 
    tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
147
 
                                             auth_scheme_scenarios)
148
 
    tests.multiply_tests(tpa_tests, tpa_scenarios, result)
149
 
 
150
 
    # activity: on all http[s] versions on all implementations
151
 
    tpact_tests, remaining_tests = tests.split_suite_by_condition(
152
 
        remaining_tests, tests.condition_isinstance((
153
 
                TestActivity,
154
 
                )))
 
110
 
 
111
 
 
112
def vary_by_http_activity():
155
113
    activity_scenarios = [
156
114
        ('urllib,http', dict(_activity_server=ActivityHTTPServer,
157
 
                             _transport=_urllib.HttpTransport_urllib,)),
 
115
                            _transport=_urllib.HttpTransport_urllib,)),
158
116
        ]
159
117
    if tests.HTTPSServerFeature.available():
160
118
        activity_scenarios.append(
161
119
            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,
162
 
                                  _transport=_urllib.HttpTransport_urllib,)),)
 
120
                                _transport=_urllib.HttpTransport_urllib,)),)
163
121
    if features.pycurl.available():
164
122
        activity_scenarios.append(
165
123
            ('pycurl,http', dict(_activity_server=ActivityHTTPServer,
166
 
                                 _transport=PyCurlTransport,)),)
 
124
                                _transport=PyCurlTransport,)),)
167
125
        if tests.HTTPSServerFeature.available():
168
126
            from bzrlib.tests import (
169
127
                ssl_certs,
181
139
 
182
140
            activity_scenarios.append(
183
141
                ('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
184
 
                                      _transport=HTTPS_pycurl_transport,)),)
185
 
 
186
 
    tpact_scenarios = tests.multiply_scenarios(activity_scenarios,
187
 
                                               protocol_scenarios)
188
 
    tests.multiply_tests(tpact_tests, tpact_scenarios, result)
189
 
 
190
 
    # No parametrization for the remaining tests
191
 
    result.addTests(remaining_tests)
192
 
 
193
 
    return result
 
142
                                    _transport=HTTPS_pycurl_transport,)),)
 
143
    return activity_scenarios
194
144
 
195
145
 
196
146
class FakeManager(object):
229
179
        self._sock.bind(('127.0.0.1', 0))
230
180
        self.host, self.port = self._sock.getsockname()
231
181
        self._ready = threading.Event()
232
 
        self._thread = test_server.ThreadWithException(
233
 
            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)
234
184
        self._thread.start()
235
185
        if 'threads' in tests.selftest_debug_flags:
236
186
            sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
305
255
        self.assertEqual('realm="Thou should not pass"', remainder)
306
256
 
307
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
 
308
292
class TestHTTPServer(tests.TestCase):
309
293
    """Test the HTTP servers implementations."""
310
294
 
401
385
class TestHttpTransportUrls(tests.TestCase):
402
386
    """Test the http urls."""
403
387
 
 
388
    scenarios = vary_by_http_client_implementation()
 
389
 
404
390
    def test_abs_url(self):
405
391
        """Construction of absolute http URLs"""
406
 
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
 
392
        t = self._transport('http://example.com/bzr/bzr.dev/')
407
393
        eq = self.assertEqualDiff
408
 
        eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
409
 
        eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
410
 
        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')
411
397
        eq(t.abspath('.bzr/1//2/./3'),
412
 
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
 
398
           'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
413
399
 
414
400
    def test_invalid_http_urls(self):
415
401
        """Trap invalid construction of urls"""
416
 
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
 
402
        self._transport('http://example.com/bzr/bzr.dev/')
417
403
        self.assertRaises(errors.InvalidURL,
418
404
                          self._transport,
419
 
                          'http://http://bazaar-vcs.org/bzr/bzr.dev/')
 
405
                          'http://http://example.com/bzr/bzr.dev/')
420
406
 
421
407
    def test_http_root_urls(self):
422
408
        """Construction of URLs from server root"""
423
 
        t = self._transport('http://bzr.ozlabs.org/')
 
409
        t = self._transport('http://example.com/')
424
410
        eq = self.assertEqualDiff
425
411
        eq(t.abspath('.bzr/tree-version'),
426
 
           'http://bzr.ozlabs.org/.bzr/tree-version')
 
412
           'http://example.com/.bzr/tree-version')
427
413
 
428
414
    def test_http_impl_urls(self):
429
415
        """There are servers which ask for particular clients to connect"""
475
461
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
476
462
    """Test the http connections."""
477
463
 
 
464
    scenarios = multiply_scenarios(
 
465
        vary_by_http_client_implementation(),
 
466
        vary_by_http_protocol_version(),
 
467
        )
 
468
 
478
469
    def setUp(self):
479
470
        http_utils.TestCaseWithWebserver.setUp(self)
480
471
        self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
525
516
class TestHttpTransportRegistration(tests.TestCase):
526
517
    """Test registrations of various http implementations"""
527
518
 
 
519
    scenarios = vary_by_http_client_implementation()
 
520
 
528
521
    def test_http_registered(self):
529
522
        t = transport.get_transport('%s://foo.com/' % self._url_protocol)
530
523
        self.assertIsInstance(t, transport.Transport)
533
526
 
534
527
class TestPost(tests.TestCase):
535
528
 
 
529
    scenarios = multiply_scenarios(
 
530
        vary_by_http_client_implementation(),
 
531
        vary_by_http_protocol_version(),
 
532
        )
 
533
 
536
534
    def test_post_body_is_received(self):
537
535
        server = RecordingServer(expect_body_tail='end-of-body',
538
536
                                 scheme=self._url_protocol)
544
542
        self.assertTrue(
545
543
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
546
544
        self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
 
545
        self.assertTrue('content-type: application/octet-stream\r'
 
546
                        in server.received_bytes.lower())
547
547
        # The transport should not be assuming that the server can accept
548
548
        # chunked encoding the first time it connects, because HTTP/1.1, so we
549
549
        # check for the literal string.
585
585
    Daughter classes are expected to override _req_handler_class
586
586
    """
587
587
 
 
588
    scenarios = multiply_scenarios(
 
589
        vary_by_http_client_implementation(),
 
590
        vary_by_http_protocol_version(),
 
591
        )
 
592
 
588
593
    # Provide a useful default
589
594
    _req_handler_class = http_server.TestingHTTPRequestHandler
590
595
 
841
846
        t = self.get_readonly_transport()
842
847
        # force transport to issue multiple requests
843
848
        t._get_max_size = 2
844
 
        l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
 
849
        list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
845
850
        # The server should have issued 3 requests
846
851
        self.assertEqual(3, server.GET_request_nb)
847
852
        self.assertEqual('0123456789', t.get_bytes('a'))
924
929
    def get_multiple_ranges(self, file, file_size, ranges):
925
930
        self.send_response(206)
926
931
        self.send_header('Accept-Ranges', 'bytes')
 
932
        # XXX: this is strange; the 'random' name below seems undefined and
 
933
        # yet the tests pass -- mbp 2010-10-11 bug 658773
927
934
        boundary = "%d" % random.randint(0,0x7FFFFFFF)
928
935
        self.send_header("Content-Type",
929
936
                         "multipart/byteranges; boundary=%s" % boundary)
991
998
                return
992
999
            self.send_range_content(file, start, end - start + 1)
993
1000
            cur += 1
994
 
        # No final boundary
 
1001
        # Final boundary
995
1002
        self.wfile.write(boundary_line)
996
1003
 
997
1004
 
1026
1033
        # that mode
1027
1034
        self.assertEqual('single', t._range_hint)
1028
1035
 
 
1036
 
1029
1037
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1030
1038
    """Errors out when range specifiers exceed the limit"""
1031
1039
 
1055
1063
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1056
1064
    """Tests readv requests against a server erroring out on too much ranges."""
1057
1065
 
 
1066
    scenarios = multiply_scenarios(
 
1067
        vary_by_http_client_implementation(),
 
1068
        vary_by_http_protocol_version(),
 
1069
        )
 
1070
 
1058
1071
    # Requests with more range specifiers will error out
1059
1072
    range_limit = 3
1060
1073
 
1095
1108
    Only the urllib implementation is tested here.
1096
1109
    """
1097
1110
 
1098
 
    def setUp(self):
1099
 
        tests.TestCase.setUp(self)
1100
 
        self._old_env = {}
1101
 
        self.addCleanup(self._restore_env)
1102
 
 
1103
 
    def _install_env(self, env):
1104
 
        for name, value in env.iteritems():
1105
 
            self._old_env[name] = osutils.set_or_unset_env(name, value)
1106
 
 
1107
 
    def _restore_env(self):
1108
 
        for name, value in self._old_env.iteritems():
1109
 
            osutils.set_or_unset_env(name, value)
1110
 
 
1111
1111
    def _proxied_request(self):
1112
1112
        handler = _urllib2_wrappers.ProxyHandler()
1113
 
        request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
 
1113
        request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
1114
1114
        handler.set_proxy(request, 'http')
1115
1115
        return request
1116
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
 
1117
1122
    def test_empty_user(self):
1118
 
        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')
1119
1130
        request = self._proxied_request()
1120
1131
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
1121
1132
 
1122
1133
    def test_invalid_proxy(self):
1123
1134
        """A proxy env variable without scheme"""
1124
 
        self._install_env({'http_proxy': 'host:1234'})
 
1135
        self.overrideEnv('http_proxy', 'host:1234')
1125
1136
        self.assertRaises(errors.InvalidURL, self._proxied_request)
1126
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
 
1127
1158
 
1128
1159
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1129
1160
    """Tests proxy server.
1134
1165
    to the file names).
1135
1166
    """
1136
1167
 
 
1168
    scenarios = multiply_scenarios(
 
1169
        vary_by_http_client_implementation(),
 
1170
        vary_by_http_protocol_version(),
 
1171
        )
 
1172
 
1137
1173
    # FIXME: We don't have an https server available, so we don't
1138
1174
    # test https connections. --vila toolongago
1139
1175
 
1153
1189
            self.no_proxy_host = self.server_host_port
1154
1190
        # The secondary server is the proxy
1155
1191
        self.proxy_url = self.get_secondary_url()
1156
 
        self._old_env = {}
1157
1192
 
1158
1193
    def _testing_pycurl(self):
1159
1194
        # TODO: This is duplicated for lots of the classes in this file
1160
1195
        return (features.pycurl.available()
1161
1196
                and self._transport == PyCurlTransport)
1162
1197
 
1163
 
    def _install_env(self, env):
1164
 
        for name, value in env.iteritems():
1165
 
            self._old_env[name] = osutils.set_or_unset_env(name, value)
1166
 
 
1167
 
    def _restore_env(self):
1168
 
        for name, value in self._old_env.iteritems():
1169
 
            osutils.set_or_unset_env(name, value)
1170
 
 
1171
 
    def proxied_in_env(self, env):
1172
 
        self._install_env(env)
1173
 
        t = self.get_readonly_transport()
1174
 
        try:
1175
 
            self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1176
 
        finally:
1177
 
            self._restore_env()
1178
 
 
1179
 
    def not_proxied_in_env(self, env):
1180
 
        self._install_env(env)
1181
 
        t = self.get_readonly_transport()
1182
 
        try:
1183
 
            self.assertEqual('contents of foo\n', t.get('foo').read())
1184
 
        finally:
1185
 
            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())
1186
1205
 
1187
1206
    def test_http_proxy(self):
1188
 
        self.proxied_in_env({'http_proxy': self.proxy_url})
 
1207
        self.overrideEnv('http_proxy', self.proxy_url)
 
1208
        self.assertProxied()
1189
1209
 
1190
1210
    def test_HTTP_PROXY(self):
1191
1211
        if self._testing_pycurl():
1194
1214
            # about. Should we ?)
1195
1215
            raise tests.TestNotApplicable(
1196
1216
                'pycurl does not check HTTP_PROXY for security reasons')
1197
 
        self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
 
1217
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
 
1218
        self.assertProxied()
1198
1219
 
1199
1220
    def test_all_proxy(self):
1200
 
        self.proxied_in_env({'all_proxy': self.proxy_url})
 
1221
        self.overrideEnv('all_proxy', self.proxy_url)
 
1222
        self.assertProxied()
1201
1223
 
1202
1224
    def test_ALL_PROXY(self):
1203
 
        self.proxied_in_env({'ALL_PROXY': self.proxy_url})
 
1225
        self.overrideEnv('ALL_PROXY', self.proxy_url)
 
1226
        self.assertProxied()
1204
1227
 
1205
1228
    def test_http_proxy_with_no_proxy(self):
1206
 
        self.not_proxied_in_env({'http_proxy': self.proxy_url,
1207
 
                                 '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()
1208
1232
 
1209
1233
    def test_HTTP_PROXY_with_NO_PROXY(self):
1210
1234
        if self._testing_pycurl():
1211
1235
            raise tests.TestNotApplicable(
1212
1236
                'pycurl does not check HTTP_PROXY for security reasons')
1213
 
        self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
1214
 
                                 '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()
1215
1240
 
1216
1241
    def test_all_proxy_with_no_proxy(self):
1217
 
        self.not_proxied_in_env({'all_proxy': self.proxy_url,
1218
 
                                 '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()
1219
1245
 
1220
1246
    def test_ALL_PROXY_with_NO_PROXY(self):
1221
 
        self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
1222
 
                                 '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()
1223
1250
 
1224
1251
    def test_http_proxy_without_scheme(self):
 
1252
        self.overrideEnv('http_proxy', self.server_host_port)
1225
1253
        if self._testing_pycurl():
1226
1254
            # pycurl *ignores* invalid proxy env variables. If that ever change
1227
1255
            # in the future, this test will fail indicating that pycurl do not
1228
1256
            # ignore anymore such variables.
1229
 
            self.not_proxied_in_env({'http_proxy': self.server_host_port})
 
1257
            self.assertNotProxied()
1230
1258
        else:
1231
 
            self.assertRaises(errors.InvalidURL,
1232
 
                              self.proxied_in_env,
1233
 
                              {'http_proxy': self.server_host_port})
 
1259
            self.assertRaises(errors.InvalidURL, self.assertProxied)
1234
1260
 
1235
1261
 
1236
1262
class TestRanges(http_utils.TestCaseWithWebserver):
1237
1263
    """Test the Range header in GET methods."""
1238
1264
 
 
1265
    scenarios = multiply_scenarios(
 
1266
        vary_by_http_client_implementation(),
 
1267
        vary_by_http_protocol_version(),
 
1268
        )
 
1269
 
1239
1270
    def setUp(self):
1240
1271
        http_utils.TestCaseWithWebserver.setUp(self)
1241
1272
        self.build_tree_contents([('a', '0123456789')],)
1281
1312
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1282
1313
    """Test redirection between http servers."""
1283
1314
 
 
1315
    scenarios = multiply_scenarios(
 
1316
        vary_by_http_client_implementation(),
 
1317
        vary_by_http_protocol_version(),
 
1318
        )
 
1319
 
1284
1320
    def setUp(self):
1285
1321
        super(TestHTTPRedirections, self).setUp()
1286
1322
        self.build_tree_contents([('a', '0123456789'),
1349
1385
    -- vila 20070212
1350
1386
    """
1351
1387
 
 
1388
    scenarios = multiply_scenarios(
 
1389
        vary_by_http_client_implementation(),
 
1390
        vary_by_http_protocol_version(),
 
1391
        )
 
1392
 
1352
1393
    def setUp(self):
1353
1394
        if (features.pycurl.available()
1354
1395
            and self._transport == PyCurlTransport):
1399
1440
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1400
1441
    """Test transport.do_catching_redirections."""
1401
1442
 
 
1443
    scenarios = multiply_scenarios(
 
1444
        vary_by_http_client_implementation(),
 
1445
        vary_by_http_protocol_version(),
 
1446
        )
 
1447
 
1402
1448
    def setUp(self):
1403
1449
        super(TestDoCatchRedirections, self).setUp()
1404
1450
        self.build_tree_contents([('a', '0123456789'),],)
1446
1492
class TestAuth(http_utils.TestCaseWithWebserver):
1447
1493
    """Test authentication scheme"""
1448
1494
 
 
1495
    scenarios = multiply_scenarios(
 
1496
        vary_by_http_client_implementation(),
 
1497
        vary_by_http_protocol_version(),
 
1498
        vary_by_http_auth_scheme(),
 
1499
        )
 
1500
 
1449
1501
    _auth_header = 'Authorization'
1450
1502
    _password_prompt_prefix = ''
1451
1503
    _username_prompt_prefix = ''
1598
1650
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1599
1651
                                            stderr=tests.StringIOWrapper())
1600
1652
        # Create a minimal config file with the right password
1601
 
        conf = config.AuthenticationConfig()
1602
 
        conf._get_config().update(
1603
 
            {'httptest': {'scheme': 'http', 'port': self.server.port,
1604
 
                          'user': user, 'password': password}})
1605
 
        conf._save()
 
1653
        _setup_authentication_config(
 
1654
            scheme='http', 
 
1655
            port=self.server.port,
 
1656
            user=user,
 
1657
            password=password)
1606
1658
        # Issue a request to the server to connect
1607
1659
        self.assertEqual('contents of a\n',t.get('a').read())
1608
1660
        # stdin should have  been left untouched
1610
1662
        # Only one 'Authentication Required' error should occur
1611
1663
        self.assertEqual(1, self.server.auth_required_errors)
1612
1664
 
1613
 
    def test_user_from_auth_conf(self):
1614
 
        if self._testing_pycurl():
1615
 
            raise tests.TestNotApplicable(
1616
 
                'pycurl does not support authentication.conf')
1617
 
        user = 'joe'
1618
 
        password = 'foo'
1619
 
        self.server.add_user(user, password)
1620
 
        # Create a minimal config file with the right password
1621
 
        conf = config.AuthenticationConfig()
1622
 
        conf._get_config().update(
1623
 
            {'httptest': {'scheme': 'http', 'port': self.server.port,
1624
 
                          'user': user, 'password': password}})
1625
 
        conf._save()
1626
 
        t = self.get_user_transport(None, None)
1627
 
        # Issue a request to the server to connect
1628
 
        self.assertEqual('contents of a\n', t.get('a').read())
1629
 
        # Only one 'Authentication Required' error should occur
1630
 
        self.assertEqual(1, self.server.auth_required_errors)
1631
 
 
1632
1665
    def test_changing_nonce(self):
1633
1666
        if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1634
1667
                                     http_utils.ProxyDigestAuthServer):
1650
1683
        # initial 'who are you' and a second 'who are you' with the new nonce)
1651
1684
        self.assertEqual(2, self.server.auth_required_errors)
1652
1685
 
 
1686
    def test_user_from_auth_conf(self):
 
1687
        if self._testing_pycurl():
 
1688
            raise tests.TestNotApplicable(
 
1689
                'pycurl does not support authentication.conf')
 
1690
        user = 'joe'
 
1691
        password = 'foo'
 
1692
        self.server.add_user(user, password)
 
1693
        _setup_authentication_config(
 
1694
            scheme='http', 
 
1695
            port=self.server.port,
 
1696
            user=user,
 
1697
            password=password)
 
1698
        t = self.get_user_transport(None, None)
 
1699
        # Issue a request to the server to connect
 
1700
        self.assertEqual('contents of a\n', t.get('a').read())
 
1701
        # Only one 'Authentication Required' error should occur
 
1702
        self.assertEqual(1, self.server.auth_required_errors)
 
1703
 
 
1704
 
 
1705
def _setup_authentication_config(**kwargs):
 
1706
    conf = config.AuthenticationConfig()
 
1707
    conf._get_config().update({'httptest': kwargs})
 
1708
    conf._save()
 
1709
 
 
1710
 
 
1711
 
 
1712
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
 
1713
    """Unit tests for glue by which urllib2 asks us for authentication"""
 
1714
 
 
1715
    def test_get_user_password_without_port(self):
 
1716
        """We cope if urllib2 doesn't tell us the port.
 
1717
 
 
1718
        See https://bugs.launchpad.net/bzr/+bug/654684
 
1719
        """
 
1720
        user = 'joe'
 
1721
        password = 'foo'
 
1722
        _setup_authentication_config(
 
1723
            scheme='http', 
 
1724
            host='localhost',
 
1725
            user=user,
 
1726
            password=password)
 
1727
        handler = _urllib2_wrappers.HTTPAuthHandler()
 
1728
        got_pass = handler.get_user_password(dict(
 
1729
            user='joe',
 
1730
            protocol='http',
 
1731
            host='localhost',
 
1732
            path='/',
 
1733
            realm='Realm',
 
1734
            ))
 
1735
        self.assertEquals((user, password), got_pass)
1653
1736
 
1654
1737
 
1655
1738
class TestProxyAuth(TestAuth):
1656
1739
    """Test proxy authentication schemes."""
1657
1740
 
 
1741
    scenarios = multiply_scenarios(
 
1742
        vary_by_http_client_implementation(),
 
1743
        vary_by_http_protocol_version(),
 
1744
        vary_by_http_proxy_auth_scheme(),
 
1745
        )
 
1746
 
1658
1747
    _auth_header = 'Proxy-authorization'
1659
1748
    _password_prompt_prefix = 'Proxy '
1660
1749
    _username_prompt_prefix = 'Proxy '
1661
1750
 
1662
1751
    def setUp(self):
1663
1752
        super(TestProxyAuth, self).setUp()
1664
 
        self._old_env = {}
1665
 
        self.addCleanup(self._restore_env)
1666
1753
        # Override the contents to avoid false positives
1667
1754
        self.build_tree_contents([('a', 'not proxied contents of a\n'),
1668
1755
                                  ('b', 'not proxied contents of b\n'),
1671
1758
                                  ])
1672
1759
 
1673
1760
    def get_user_transport(self, user, password):
1674
 
        self._install_env({'all_proxy': self.get_user_url(user, password)})
 
1761
        self.overrideEnv('all_proxy', self.get_user_url(user, password))
1675
1762
        return TestAuth.get_user_transport(self, user, password)
1676
1763
 
1677
 
    def _install_env(self, env):
1678
 
        for name, value in env.iteritems():
1679
 
            self._old_env[name] = osutils.set_or_unset_env(name, value)
1680
 
 
1681
 
    def _restore_env(self):
1682
 
        for name, value in self._old_env.iteritems():
1683
 
            osutils.set_or_unset_env(name, value)
1684
 
 
1685
1764
    def test_empty_pass(self):
1686
1765
        if self._testing_pycurl():
1687
1766
            import pycurl
1716
1795
 
1717
1796
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1718
1797
 
 
1798
    scenarios = multiply_scenarios(
 
1799
        vary_by_http_client_implementation(),
 
1800
        vary_by_http_protocol_version(),
 
1801
        )
 
1802
 
1719
1803
    def setUp(self):
1720
1804
        super(SmartHTTPTunnellingTest, self).setUp()
1721
1805
        # We use the VFS layer as part of HTTP tunnelling tests.
1722
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1806
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1723
1807
        self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1724
1808
        self.http_server = self.get_readonly_server()
1725
1809
 
1810
1894
 
1811
1895
class Test_redirected_to(tests.TestCase):
1812
1896
 
 
1897
    scenarios = vary_by_http_client_implementation()
 
1898
 
1813
1899
    def test_redirected_to_subdir(self):
1814
1900
        t = self._transport('http://www.example.com/foo')
1815
1901
        r = t._redirected_to('http://www.example.com/foo',
2061
2147
 
2062
2148
class TestActivity(tests.TestCase, TestActivityMixin):
2063
2149
 
 
2150
    scenarios = multiply_scenarios(
 
2151
        vary_by_http_activity(),
 
2152
        vary_by_http_protocol_version(),
 
2153
        )
 
2154
 
2064
2155
    def setUp(self):
2065
2156
        TestActivityMixin.setUp(self)
2066
2157
 
2087
2178
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2088
2179
    """Test authentication on the redirected http server."""
2089
2180
 
 
2181
    scenarios = vary_by_http_protocol_version()
 
2182
 
2090
2183
    _auth_header = 'Authorization'
2091
2184
    _password_prompt_prefix = ''
2092
2185
    _username_prompt_prefix = ''