← 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-09-25 20:08:01 UTC
  • mfrom: (5444 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5445.
  • Revision ID: john@arbash-meinel.com-20100925200801-7uf0ux3uwxo9i3x0
Merge bzr.dev 5444 to resolve some small text conflicts.

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
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
26
27
import httplib
 
28
import os
 
29
import select
27
30
import SimpleHTTPServer
28
31
import socket
29
32
import sys
32
35
import bzrlib
33
36
from bzrlib import (
34
37
    bzrdir,
35
 
    cethread,
36
38
    config,
37
39
    errors,
38
40
    osutils,
40
42
    tests,
41
43
    transport,
42
44
    ui,
 
45
    urlutils,
43
46
    )
44
47
from bzrlib.tests import (
45
48
    features,
47
50
    http_utils,
48
51
    test_server,
49
52
    )
50
 
from bzrlib.tests.scenarios import (
51
 
    load_tests_apply_scenarios,
52
 
    multiply_scenarios,
53
 
    )
54
53
from bzrlib.transport import (
55
54
    http,
56
55
    remote,
65
64
    from bzrlib.transport.http._pycurl import PyCurlTransport
66
65
 
67
66
 
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."""
 
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
                )))
73
78
    transport_scenarios = [
74
79
        ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
75
80
                        _server=http_server.HttpServer_urllib,
80
85
            ('pycurl', dict(_transport=PyCurlTransport,
81
86
                            _server=http_server.HttpServer_PyCurl,
82
87
                            _url_protocol='http+pycurl',)))
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 [
 
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 = [
96
126
        ('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
97
127
        ('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
98
128
        ('basicdigest',
99
 
            dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
 
129
         dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
100
130
        ]
101
 
 
102
 
 
103
 
def vary_by_http_auth_scheme():
104
 
    return [
 
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 = [
105
141
        ('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
106
142
        ('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
107
143
        ('basicdigest',
108
 
            dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
 
144
         dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
109
145
        ]
110
 
 
111
 
 
112
 
def vary_by_http_activity():
 
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
                )))
113
155
    activity_scenarios = [
114
156
        ('urllib,http', dict(_activity_server=ActivityHTTPServer,
115
 
                            _transport=_urllib.HttpTransport_urllib,)),
 
157
                             _transport=_urllib.HttpTransport_urllib,)),
116
158
        ]
117
159
    if tests.HTTPSServerFeature.available():
118
160
        activity_scenarios.append(
119
161
            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,
120
 
                                _transport=_urllib.HttpTransport_urllib,)),)
 
162
                                  _transport=_urllib.HttpTransport_urllib,)),)
121
163
    if features.pycurl.available():
122
164
        activity_scenarios.append(
123
165
            ('pycurl,http', dict(_activity_server=ActivityHTTPServer,
124
 
                                _transport=PyCurlTransport,)),)
 
166
                                 _transport=PyCurlTransport,)),)
125
167
        if tests.HTTPSServerFeature.available():
126
168
            from bzrlib.tests import (
127
169
                ssl_certs,
139
181
 
140
182
            activity_scenarios.append(
141
183
                ('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
142
 
                                    _transport=HTTPS_pycurl_transport,)),)
143
 
    return activity_scenarios
 
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
144
194
 
145
195
 
146
196
class FakeManager(object):
179
229
        self._sock.bind(('127.0.0.1', 0))
180
230
        self.host, self.port = self._sock.getsockname()
181
231
        self._ready = threading.Event()
182
 
        self._thread = test_server.TestThread(
183
 
            sync_event=self._ready, target=self._accept_read_and_reply)
 
232
        self._thread = test_server.ThreadWithException(
 
233
            event=self._ready, target=self._accept_read_and_reply)
184
234
        self._thread.start()
185
235
        if 'threads' in tests.selftest_debug_flags:
186
236
            sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
255
305
        self.assertEqual('realm="Thou should not pass"', remainder)
256
306
 
257
307
 
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
308
class TestHTTPServer(tests.TestCase):
293
309
    """Test the HTTP servers implementations."""
294
310
 
385
401
class TestHttpTransportUrls(tests.TestCase):
386
402
    """Test the http urls."""
387
403
 
388
 
    scenarios = vary_by_http_client_implementation()
389
 
 
390
404
    def test_abs_url(self):
391
405
        """Construction of absolute http URLs"""
392
 
        t = self._transport('http://example.com/bzr/bzr.dev/')
 
406
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
393
407
        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')
 
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')
397
411
        eq(t.abspath('.bzr/1//2/./3'),
398
 
           'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
 
412
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
399
413
 
400
414
    def test_invalid_http_urls(self):
401
415
        """Trap invalid construction of urls"""
402
 
        self._transport('http://example.com/bzr/bzr.dev/')
 
416
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
403
417
        self.assertRaises(errors.InvalidURL,
404
418
                          self._transport,
405
 
                          'http://http://example.com/bzr/bzr.dev/')
 
419
                          'http://http://bazaar-vcs.org/bzr/bzr.dev/')
406
420
 
407
421
    def test_http_root_urls(self):
408
422
        """Construction of URLs from server root"""
409
 
        t = self._transport('http://example.com/')
 
423
        t = self._transport('http://bzr.ozlabs.org/')
410
424
        eq = self.assertEqualDiff
411
425
        eq(t.abspath('.bzr/tree-version'),
412
 
           'http://example.com/.bzr/tree-version')
 
426
           'http://bzr.ozlabs.org/.bzr/tree-version')
413
427
 
414
428
    def test_http_impl_urls(self):
415
429
        """There are servers which ask for particular clients to connect"""
461
475
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
462
476
    """Test the http connections."""
463
477
 
464
 
    scenarios = multiply_scenarios(
465
 
        vary_by_http_client_implementation(),
466
 
        vary_by_http_protocol_version(),
467
 
        )
468
 
 
469
478
    def setUp(self):
470
479
        http_utils.TestCaseWithWebserver.setUp(self)
471
480
        self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
516
525
class TestHttpTransportRegistration(tests.TestCase):
517
526
    """Test registrations of various http implementations"""
518
527
 
519
 
    scenarios = vary_by_http_client_implementation()
520
 
 
521
528
    def test_http_registered(self):
522
529
        t = transport.get_transport('%s://foo.com/' % self._url_protocol)
523
530
        self.assertIsInstance(t, transport.Transport)
526
533
 
527
534
class TestPost(tests.TestCase):
528
535
 
529
 
    scenarios = multiply_scenarios(
530
 
        vary_by_http_client_implementation(),
531
 
        vary_by_http_protocol_version(),
532
 
        )
533
 
 
534
536
    def test_post_body_is_received(self):
535
537
        server = RecordingServer(expect_body_tail='end-of-body',
536
538
                                 scheme=self._url_protocol)
542
544
        self.assertTrue(
543
545
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
544
546
        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
 
 
593
588
    # Provide a useful default
594
589
    _req_handler_class = http_server.TestingHTTPRequestHandler
595
590
 
846
841
        t = self.get_readonly_transport()
847
842
        # force transport to issue multiple requests
848
843
        t._get_max_size = 2
849
 
        list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
 
844
        l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
850
845
        # The server should have issued 3 requests
851
846
        self.assertEqual(3, server.GET_request_nb)
852
847
        self.assertEqual('0123456789', t.get_bytes('a'))
929
924
    def get_multiple_ranges(self, file, file_size, ranges):
930
925
        self.send_response(206)
931
926
        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
934
927
        boundary = "%d" % random.randint(0,0x7FFFFFFF)
935
928
        self.send_header("Content-Type",
936
929
                         "multipart/byteranges; boundary=%s" % boundary)
998
991
                return
999
992
            self.send_range_content(file, start, end - start + 1)
1000
993
            cur += 1
1001
 
        # Final boundary
 
994
        # No final boundary
1002
995
        self.wfile.write(boundary_line)
1003
996
 
1004
997
 
1033
1026
        # that mode
1034
1027
        self.assertEqual('single', t._range_hint)
1035
1028
 
1036
 
 
1037
1029
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1038
1030
    """Errors out when range specifiers exceed the limit"""
1039
1031
 
1063
1055
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1064
1056
    """Tests readv requests against a server erroring out on too much ranges."""
1065
1057
 
1066
 
    scenarios = multiply_scenarios(
1067
 
        vary_by_http_client_implementation(),
1068
 
        vary_by_http_protocol_version(),
1069
 
        )
1070
 
 
1071
1058
    # Requests with more range specifiers will error out
1072
1059
    range_limit = 3
1073
1060
 
1108
1095
    Only the urllib implementation is tested here.
1109
1096
    """
1110
1097
 
 
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
 
 
1122
1117
    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')
 
1118
        self._install_env({'http_proxy': 'http://bar.com'})
1130
1119
        request = self._proxied_request()
1131
1120
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
1132
1121
 
1133
1122
    def test_invalid_proxy(self):
1134
1123
        """A proxy env variable without scheme"""
1135
 
        self.overrideEnv('http_proxy', 'host:1234')
 
1124
        self._install_env({'http_proxy': 'host:1234'})
1136
1125
        self.assertRaises(errors.InvalidURL, self._proxied_request)
1137
1126
 
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
1127
 
1159
1128
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1160
1129
    """Tests proxy server.
1165
1134
    to the file names).
1166
1135
    """
1167
1136
 
1168
 
    scenarios = multiply_scenarios(
1169
 
        vary_by_http_client_implementation(),
1170
 
        vary_by_http_protocol_version(),
1171
 
        )
1172
 
 
1173
1137
    # FIXME: We don't have an https server available, so we don't
1174
1138
    # test https connections. --vila toolongago
1175
1139
 
1189
1153
            self.no_proxy_host = self.server_host_port
1190
1154
        # The secondary server is the proxy
1191
1155
        self.proxy_url = self.get_secondary_url()
 
1156
        self._old_env = {}
1192
1157
 
1193
1158
    def _testing_pycurl(self):
1194
1159
        # TODO: This is duplicated for lots of the classes in this file
1195
1160
        return (features.pycurl.available()
1196
1161
                and self._transport == PyCurlTransport)
1197
1162
 
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())
 
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()
1205
1186
 
1206
1187
    def test_http_proxy(self):
1207
 
        self.overrideEnv('http_proxy', self.proxy_url)
1208
 
        self.assertProxied()
 
1188
        self.proxied_in_env({'http_proxy': self.proxy_url})
1209
1189
 
1210
1190
    def test_HTTP_PROXY(self):
1211
1191
        if self._testing_pycurl():
1214
1194
            # about. Should we ?)
1215
1195
            raise tests.TestNotApplicable(
1216
1196
                'pycurl does not check HTTP_PROXY for security reasons')
1217
 
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
1218
 
        self.assertProxied()
 
1197
        self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
1219
1198
 
1220
1199
    def test_all_proxy(self):
1221
 
        self.overrideEnv('all_proxy', self.proxy_url)
1222
 
        self.assertProxied()
 
1200
        self.proxied_in_env({'all_proxy': self.proxy_url})
1223
1201
 
1224
1202
    def test_ALL_PROXY(self):
1225
 
        self.overrideEnv('ALL_PROXY', self.proxy_url)
1226
 
        self.assertProxied()
 
1203
        self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1227
1204
 
1228
1205
    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()
 
1206
        self.not_proxied_in_env({'http_proxy': self.proxy_url,
 
1207
                                 'no_proxy': self.no_proxy_host})
1232
1208
 
1233
1209
    def test_HTTP_PROXY_with_NO_PROXY(self):
1234
1210
        if self._testing_pycurl():
1235
1211
            raise tests.TestNotApplicable(
1236
1212
                '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()
 
1213
        self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
 
1214
                                 'NO_PROXY': self.no_proxy_host})
1240
1215
 
1241
1216
    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()
 
1217
        self.not_proxied_in_env({'all_proxy': self.proxy_url,
 
1218
                                 'no_proxy': self.no_proxy_host})
1245
1219
 
1246
1220
    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()
 
1221
        self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
 
1222
                                 'NO_PROXY': self.no_proxy_host})
1250
1223
 
1251
1224
    def test_http_proxy_without_scheme(self):
1252
 
        self.overrideEnv('http_proxy', self.server_host_port)
1253
1225
        if self._testing_pycurl():
1254
1226
            # pycurl *ignores* invalid proxy env variables. If that ever change
1255
1227
            # in the future, this test will fail indicating that pycurl do not
1256
1228
            # ignore anymore such variables.
1257
 
            self.assertNotProxied()
 
1229
            self.not_proxied_in_env({'http_proxy': self.server_host_port})
1258
1230
        else:
1259
 
            self.assertRaises(errors.InvalidURL, self.assertProxied)
 
1231
            self.assertRaises(errors.InvalidURL,
 
1232
                              self.proxied_in_env,
 
1233
                              {'http_proxy': self.server_host_port})
1260
1234
 
1261
1235
 
1262
1236
class TestRanges(http_utils.TestCaseWithWebserver):
1263
1237
    """Test the Range header in GET methods."""
1264
1238
 
1265
 
    scenarios = multiply_scenarios(
1266
 
        vary_by_http_client_implementation(),
1267
 
        vary_by_http_protocol_version(),
1268
 
        )
1269
 
 
1270
1239
    def setUp(self):
1271
1240
        http_utils.TestCaseWithWebserver.setUp(self)
1272
1241
        self.build_tree_contents([('a', '0123456789')],)
1312
1281
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1313
1282
    """Test redirection between http servers."""
1314
1283
 
1315
 
    scenarios = multiply_scenarios(
1316
 
        vary_by_http_client_implementation(),
1317
 
        vary_by_http_protocol_version(),
1318
 
        )
1319
 
 
1320
1284
    def setUp(self):
1321
1285
        super(TestHTTPRedirections, self).setUp()
1322
1286
        self.build_tree_contents([('a', '0123456789'),
1385
1349
    -- vila 20070212
1386
1350
    """
1387
1351
 
1388
 
    scenarios = multiply_scenarios(
1389
 
        vary_by_http_client_implementation(),
1390
 
        vary_by_http_protocol_version(),
1391
 
        )
1392
 
 
1393
1352
    def setUp(self):
1394
1353
        if (features.pycurl.available()
1395
1354
            and self._transport == PyCurlTransport):
1440
1399
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1441
1400
    """Test transport.do_catching_redirections."""
1442
1401
 
1443
 
    scenarios = multiply_scenarios(
1444
 
        vary_by_http_client_implementation(),
1445
 
        vary_by_http_protocol_version(),
1446
 
        )
1447
 
 
1448
1402
    def setUp(self):
1449
1403
        super(TestDoCatchRedirections, self).setUp()
1450
1404
        self.build_tree_contents([('a', '0123456789'),],)
1492
1446
class TestAuth(http_utils.TestCaseWithWebserver):
1493
1447
    """Test authentication scheme"""
1494
1448
 
1495
 
    scenarios = multiply_scenarios(
1496
 
        vary_by_http_client_implementation(),
1497
 
        vary_by_http_protocol_version(),
1498
 
        vary_by_http_auth_scheme(),
1499
 
        )
1500
 
 
1501
1449
    _auth_header = 'Authorization'
1502
1450
    _password_prompt_prefix = ''
1503
1451
    _username_prompt_prefix = ''
1650
1598
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1651
1599
                                            stderr=tests.StringIOWrapper())
1652
1600
        # Create a minimal config file with the right password
1653
 
        _setup_authentication_config(
1654
 
            scheme='http', 
1655
 
            port=self.server.port,
1656
 
            user=user,
1657
 
            password=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()
1658
1606
        # Issue a request to the server to connect
1659
1607
        self.assertEqual('contents of a\n',t.get('a').read())
1660
1608
        # stdin should have  been left untouched
1662
1610
        # Only one 'Authentication Required' error should occur
1663
1611
        self.assertEqual(1, self.server.auth_required_errors)
1664
1612
 
 
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
 
1665
1632
    def test_changing_nonce(self):
1666
1633
        if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1667
1634
                                     http_utils.ProxyDigestAuthServer):
1683
1650
        # initial 'who are you' and a second 'who are you' with the new nonce)
1684
1651
        self.assertEqual(2, self.server.auth_required_errors)
1685
1652
 
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)
1736
1653
 
1737
1654
 
1738
1655
class TestProxyAuth(TestAuth):
1739
1656
    """Test proxy authentication schemes."""
1740
1657
 
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
 
 
1747
1658
    _auth_header = 'Proxy-authorization'
1748
1659
    _password_prompt_prefix = 'Proxy '
1749
1660
    _username_prompt_prefix = 'Proxy '
1750
1661
 
1751
1662
    def setUp(self):
1752
1663
        super(TestProxyAuth, self).setUp()
 
1664
        self._old_env = {}
 
1665
        self.addCleanup(self._restore_env)
1753
1666
        # Override the contents to avoid false positives
1754
1667
        self.build_tree_contents([('a', 'not proxied contents of a\n'),
1755
1668
                                  ('b', 'not proxied contents of b\n'),
1758
1671
                                  ])
1759
1672
 
1760
1673
    def get_user_transport(self, user, password):
1761
 
        self.overrideEnv('all_proxy', self.get_user_url(user, password))
 
1674
        self._install_env({'all_proxy': self.get_user_url(user, password)})
1762
1675
        return TestAuth.get_user_transport(self, user, password)
1763
1676
 
 
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
 
1764
1685
    def test_empty_pass(self):
1765
1686
        if self._testing_pycurl():
1766
1687
            import pycurl
1795
1716
 
1796
1717
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1797
1718
 
1798
 
    scenarios = multiply_scenarios(
1799
 
        vary_by_http_client_implementation(),
1800
 
        vary_by_http_protocol_version(),
1801
 
        )
1802
 
 
1803
1719
    def setUp(self):
1804
1720
        super(SmartHTTPTunnellingTest, self).setUp()
1805
1721
        # We use the VFS layer as part of HTTP tunnelling tests.
1806
 
        self.overrideEnv('BZR_NO_SMART_VFS', None)
 
1722
        self._captureVar('BZR_NO_SMART_VFS', None)
1807
1723
        self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1808
1724
        self.http_server = self.get_readonly_server()
1809
1725
 
1894
1810
 
1895
1811
class Test_redirected_to(tests.TestCase):
1896
1812
 
1897
 
    scenarios = vary_by_http_client_implementation()
1898
 
 
1899
1813
    def test_redirected_to_subdir(self):
1900
1814
        t = self._transport('http://www.example.com/foo')
1901
1815
        r = t._redirected_to('http://www.example.com/foo',
2147
2061
 
2148
2062
class TestActivity(tests.TestCase, TestActivityMixin):
2149
2063
 
2150
 
    scenarios = multiply_scenarios(
2151
 
        vary_by_http_activity(),
2152
 
        vary_by_http_protocol_version(),
2153
 
        )
2154
 
 
2155
2064
    def setUp(self):
2156
2065
        TestActivityMixin.setUp(self)
2157
2066
 
2178
2087
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2179
2088
    """Test authentication on the redirected http server."""
2180
2089
 
2181
 
    scenarios = vary_by_http_protocol_version()
2182
 
 
2183
2090
    _auth_header = 'Authorization'
2184
2091
    _password_prompt_prefix = ''
2185
2092
    _username_prompt_prefix = ''