← 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: Vincent Ladeuil
  • Date: 2010-10-08 10:50:51 UTC
  • mfrom: (5050.17.31 2.2.2-dev)
  • mto: This revision was merged to the branch mainline in revision 5474.
  • Revision ID: v.ladeuil+lp@free.fr-20101008105051-xd4knkrohzclffic
Merge 2.2 into trunk including fixes for bug #651706 and bug #646133

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,))
351
401
class TestHttpTransportUrls(tests.TestCase):
352
402
    """Test the http urls."""
353
403
 
354
 
    scenarios = vary_by_http_client_implementation()
355
 
 
356
404
    def test_abs_url(self):
357
405
        """Construction of absolute http URLs"""
358
 
        t = self._transport('http://example.com/bzr/bzr.dev/')
 
406
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
359
407
        eq = self.assertEqualDiff
360
 
        eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
361
 
        eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
362
 
        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')
363
411
        eq(t.abspath('.bzr/1//2/./3'),
364
 
           'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
 
412
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
365
413
 
366
414
    def test_invalid_http_urls(self):
367
415
        """Trap invalid construction of urls"""
368
 
        self._transport('http://example.com/bzr/bzr.dev/')
 
416
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
369
417
        self.assertRaises(errors.InvalidURL,
370
418
                          self._transport,
371
 
                          'http://http://example.com/bzr/bzr.dev/')
 
419
                          'http://http://bazaar-vcs.org/bzr/bzr.dev/')
372
420
 
373
421
    def test_http_root_urls(self):
374
422
        """Construction of URLs from server root"""
375
 
        t = self._transport('http://example.com/')
 
423
        t = self._transport('http://bzr.ozlabs.org/')
376
424
        eq = self.assertEqualDiff
377
425
        eq(t.abspath('.bzr/tree-version'),
378
 
           'http://example.com/.bzr/tree-version')
 
426
           'http://bzr.ozlabs.org/.bzr/tree-version')
379
427
 
380
428
    def test_http_impl_urls(self):
381
429
        """There are servers which ask for particular clients to connect"""
427
475
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
428
476
    """Test the http connections."""
429
477
 
430
 
    scenarios = multiply_scenarios(
431
 
        vary_by_http_client_implementation(), 
432
 
        vary_by_http_protocol_version(),
433
 
        )
434
 
 
435
478
    def setUp(self):
436
479
        http_utils.TestCaseWithWebserver.setUp(self)
437
480
        self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
482
525
class TestHttpTransportRegistration(tests.TestCase):
483
526
    """Test registrations of various http implementations"""
484
527
 
485
 
    scenarios = vary_by_http_client_implementation()
486
 
 
487
528
    def test_http_registered(self):
488
529
        t = transport.get_transport('%s://foo.com/' % self._url_protocol)
489
530
        self.assertIsInstance(t, transport.Transport)
492
533
 
493
534
class TestPost(tests.TestCase):
494
535
 
495
 
    scenarios = multiply_scenarios(
496
 
        vary_by_http_client_implementation(), 
497
 
        vary_by_http_protocol_version(),
498
 
        )
499
 
 
500
536
    def test_post_body_is_received(self):
501
537
        server = RecordingServer(expect_body_tail='end-of-body',
502
538
                                 scheme=self._url_protocol)
508
544
        self.assertTrue(
509
545
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
510
546
        self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
511
 
        self.assertTrue('content-type: application/octet-stream\r'
512
 
                        in server.received_bytes.lower())
513
547
        # The transport should not be assuming that the server can accept
514
548
        # chunked encoding the first time it connects, because HTTP/1.1, so we
515
549
        # check for the literal string.
551
585
    Daughter classes are expected to override _req_handler_class
552
586
    """
553
587
 
554
 
    scenarios = multiply_scenarios(
555
 
        vary_by_http_client_implementation(), 
556
 
        vary_by_http_protocol_version(),
557
 
        )
558
 
 
559
588
    # Provide a useful default
560
589
    _req_handler_class = http_server.TestingHTTPRequestHandler
561
590
 
812
841
        t = self.get_readonly_transport()
813
842
        # force transport to issue multiple requests
814
843
        t._get_max_size = 2
815
 
        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))))
816
845
        # The server should have issued 3 requests
817
846
        self.assertEqual(3, server.GET_request_nb)
818
847
        self.assertEqual('0123456789', t.get_bytes('a'))
895
924
    def get_multiple_ranges(self, file, file_size, ranges):
896
925
        self.send_response(206)
897
926
        self.send_header('Accept-Ranges', 'bytes')
898
 
        # XXX: this is strange; the 'random' name below seems undefined and
899
 
        # yet the tests pass -- mbp 2010-10-11 bug 658773
900
927
        boundary = "%d" % random.randint(0,0x7FFFFFFF)
901
928
        self.send_header("Content-Type",
902
929
                         "multipart/byteranges; boundary=%s" % boundary)
964
991
                return
965
992
            self.send_range_content(file, start, end - start + 1)
966
993
            cur += 1
967
 
        # Final boundary
 
994
        # No final boundary
968
995
        self.wfile.write(boundary_line)
969
996
 
970
997
 
999
1026
        # that mode
1000
1027
        self.assertEqual('single', t._range_hint)
1001
1028
 
1002
 
 
1003
1029
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1004
1030
    """Errors out when range specifiers exceed the limit"""
1005
1031
 
1029
1055
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1030
1056
    """Tests readv requests against a server erroring out on too much ranges."""
1031
1057
 
1032
 
    scenarios = multiply_scenarios(
1033
 
        vary_by_http_client_implementation(), 
1034
 
        vary_by_http_protocol_version(),
1035
 
        )
1036
 
 
1037
1058
    # Requests with more range specifiers will error out
1038
1059
    range_limit = 3
1039
1060
 
1074
1095
    Only the urllib implementation is tested here.
1075
1096
    """
1076
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
 
1077
1111
    def _proxied_request(self):
1078
1112
        handler = _urllib2_wrappers.ProxyHandler()
1079
1113
        request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
1080
1114
        handler.set_proxy(request, 'http')
1081
1115
        return request
1082
1116
 
1083
 
    def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1084
 
        handler = _urllib2_wrappers.ProxyHandler()
1085
 
        self.assertEquals(expected,
1086
 
                          handler.evaluate_proxy_bypass(host, no_proxy))
1087
 
 
1088
1117
    def test_empty_user(self):
1089
 
        self.overrideEnv('http_proxy', 'http://bar.com')
 
1118
        self._install_env({'http_proxy': 'http://bar.com'})
1090
1119
        request = self._proxied_request()
1091
1120
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
1092
1121
 
1093
1122
    def test_invalid_proxy(self):
1094
1123
        """A proxy env variable without scheme"""
1095
 
        self.overrideEnv('http_proxy', 'host:1234')
 
1124
        self._install_env({'http_proxy': 'host:1234'})
1096
1125
        self.assertRaises(errors.InvalidURL, self._proxied_request)
1097
1126
 
1098
 
    def test_evaluate_proxy_bypass_true(self):
1099
 
        """The host is not proxied"""
1100
 
        self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
1101
 
        self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
1102
 
 
1103
 
    def test_evaluate_proxy_bypass_false(self):
1104
 
        """The host is proxied"""
1105
 
        self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
1106
 
 
1107
 
    def test_evaluate_proxy_bypass_unknown(self):
1108
 
        """The host is not explicitly proxied"""
1109
 
        self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
1110
 
        self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
1111
 
 
1112
 
    def test_evaluate_proxy_bypass_empty_entries(self):
1113
 
        """Ignore empty entries"""
1114
 
        self.assertEvaluateProxyBypass(None, 'example.com', '')
1115
 
        self.assertEvaluateProxyBypass(None, 'example.com', ',')
1116
 
        self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
1117
 
 
1118
1127
 
1119
1128
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1120
1129
    """Tests proxy server.
1125
1134
    to the file names).
1126
1135
    """
1127
1136
 
1128
 
    scenarios = multiply_scenarios(
1129
 
        vary_by_http_client_implementation(), 
1130
 
        vary_by_http_protocol_version(),
1131
 
        )
1132
 
 
1133
1137
    # FIXME: We don't have an https server available, so we don't
1134
1138
    # test https connections. --vila toolongago
1135
1139
 
1149
1153
            self.no_proxy_host = self.server_host_port
1150
1154
        # The secondary server is the proxy
1151
1155
        self.proxy_url = self.get_secondary_url()
 
1156
        self._old_env = {}
1152
1157
 
1153
1158
    def _testing_pycurl(self):
1154
1159
        # TODO: This is duplicated for lots of the classes in this file
1155
1160
        return (features.pycurl.available()
1156
1161
                and self._transport == PyCurlTransport)
1157
1162
 
1158
 
    def assertProxied(self):
1159
 
        t = self.get_readonly_transport()
1160
 
        self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1161
 
 
1162
 
    def assertNotProxied(self):
1163
 
        t = self.get_readonly_transport()
1164
 
        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()
1165
1186
 
1166
1187
    def test_http_proxy(self):
1167
 
        self.overrideEnv('http_proxy', self.proxy_url)
1168
 
        self.assertProxied()
 
1188
        self.proxied_in_env({'http_proxy': self.proxy_url})
1169
1189
 
1170
1190
    def test_HTTP_PROXY(self):
1171
1191
        if self._testing_pycurl():
1174
1194
            # about. Should we ?)
1175
1195
            raise tests.TestNotApplicable(
1176
1196
                'pycurl does not check HTTP_PROXY for security reasons')
1177
 
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
1178
 
        self.assertProxied()
 
1197
        self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
1179
1198
 
1180
1199
    def test_all_proxy(self):
1181
 
        self.overrideEnv('all_proxy', self.proxy_url)
1182
 
        self.assertProxied()
 
1200
        self.proxied_in_env({'all_proxy': self.proxy_url})
1183
1201
 
1184
1202
    def test_ALL_PROXY(self):
1185
 
        self.overrideEnv('ALL_PROXY', self.proxy_url)
1186
 
        self.assertProxied()
 
1203
        self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1187
1204
 
1188
1205
    def test_http_proxy_with_no_proxy(self):
1189
 
        self.overrideEnv('no_proxy', self.no_proxy_host)
1190
 
        self.overrideEnv('http_proxy', self.proxy_url)
1191
 
        self.assertNotProxied()
 
1206
        self.not_proxied_in_env({'http_proxy': self.proxy_url,
 
1207
                                 'no_proxy': self.no_proxy_host})
1192
1208
 
1193
1209
    def test_HTTP_PROXY_with_NO_PROXY(self):
1194
1210
        if self._testing_pycurl():
1195
1211
            raise tests.TestNotApplicable(
1196
1212
                'pycurl does not check HTTP_PROXY for security reasons')
1197
 
        self.overrideEnv('NO_PROXY', self.no_proxy_host)
1198
 
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
1199
 
        self.assertNotProxied()
 
1213
        self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
 
1214
                                 'NO_PROXY': self.no_proxy_host})
1200
1215
 
1201
1216
    def test_all_proxy_with_no_proxy(self):
1202
 
        self.overrideEnv('no_proxy', self.no_proxy_host)
1203
 
        self.overrideEnv('all_proxy', self.proxy_url)
1204
 
        self.assertNotProxied()
 
1217
        self.not_proxied_in_env({'all_proxy': self.proxy_url,
 
1218
                                 'no_proxy': self.no_proxy_host})
1205
1219
 
1206
1220
    def test_ALL_PROXY_with_NO_PROXY(self):
1207
 
        self.overrideEnv('NO_PROXY', self.no_proxy_host)
1208
 
        self.overrideEnv('ALL_PROXY', self.proxy_url)
1209
 
        self.assertNotProxied()
 
1221
        self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
 
1222
                                 'NO_PROXY': self.no_proxy_host})
1210
1223
 
1211
1224
    def test_http_proxy_without_scheme(self):
1212
 
        self.overrideEnv('http_proxy', self.server_host_port)
1213
1225
        if self._testing_pycurl():
1214
1226
            # pycurl *ignores* invalid proxy env variables. If that ever change
1215
1227
            # in the future, this test will fail indicating that pycurl do not
1216
1228
            # ignore anymore such variables.
1217
 
            self.assertNotProxied()
 
1229
            self.not_proxied_in_env({'http_proxy': self.server_host_port})
1218
1230
        else:
1219
 
            self.assertRaises(errors.InvalidURL, self.assertProxied)
 
1231
            self.assertRaises(errors.InvalidURL,
 
1232
                              self.proxied_in_env,
 
1233
                              {'http_proxy': self.server_host_port})
1220
1234
 
1221
1235
 
1222
1236
class TestRanges(http_utils.TestCaseWithWebserver):
1223
1237
    """Test the Range header in GET methods."""
1224
1238
 
1225
 
    scenarios = multiply_scenarios(
1226
 
        vary_by_http_client_implementation(), 
1227
 
        vary_by_http_protocol_version(),
1228
 
        )
1229
 
 
1230
1239
    def setUp(self):
1231
1240
        http_utils.TestCaseWithWebserver.setUp(self)
1232
1241
        self.build_tree_contents([('a', '0123456789')],)
1272
1281
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1273
1282
    """Test redirection between http servers."""
1274
1283
 
1275
 
    scenarios = multiply_scenarios(
1276
 
        vary_by_http_client_implementation(), 
1277
 
        vary_by_http_protocol_version(),
1278
 
        )
1279
 
 
1280
1284
    def setUp(self):
1281
1285
        super(TestHTTPRedirections, self).setUp()
1282
1286
        self.build_tree_contents([('a', '0123456789'),
1345
1349
    -- vila 20070212
1346
1350
    """
1347
1351
 
1348
 
    scenarios = multiply_scenarios(
1349
 
        vary_by_http_client_implementation(), 
1350
 
        vary_by_http_protocol_version(),
1351
 
        )
1352
 
 
1353
1352
    def setUp(self):
1354
1353
        if (features.pycurl.available()
1355
1354
            and self._transport == PyCurlTransport):
1400
1399
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1401
1400
    """Test transport.do_catching_redirections."""
1402
1401
 
1403
 
    scenarios = multiply_scenarios(
1404
 
        vary_by_http_client_implementation(), 
1405
 
        vary_by_http_protocol_version(),
1406
 
        )
1407
 
 
1408
1402
    def setUp(self):
1409
1403
        super(TestDoCatchRedirections, self).setUp()
1410
1404
        self.build_tree_contents([('a', '0123456789'),],)
1452
1446
class TestAuth(http_utils.TestCaseWithWebserver):
1453
1447
    """Test authentication scheme"""
1454
1448
 
1455
 
    scenarios = multiply_scenarios(
1456
 
        vary_by_http_client_implementation(),
1457
 
        vary_by_http_protocol_version(),
1458
 
        vary_by_http_auth_scheme(),
1459
 
        )
1460
 
 
1461
1449
    _auth_header = 'Authorization'
1462
1450
    _password_prompt_prefix = ''
1463
1451
    _username_prompt_prefix = ''
1610
1598
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1611
1599
                                            stderr=tests.StringIOWrapper())
1612
1600
        # Create a minimal config file with the right password
1613
 
        _setup_authentication_config(
1614
 
            scheme='http', 
1615
 
            port=self.server.port,
1616
 
            user=user,
1617
 
            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()
1618
1606
        # Issue a request to the server to connect
1619
1607
        self.assertEqual('contents of a\n',t.get('a').read())
1620
1608
        # stdin should have  been left untouched
1622
1610
        # Only one 'Authentication Required' error should occur
1623
1611
        self.assertEqual(1, self.server.auth_required_errors)
1624
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
 
1625
1632
    def test_changing_nonce(self):
1626
1633
        if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1627
1634
                                     http_utils.ProxyDigestAuthServer):
1643
1650
        # initial 'who are you' and a second 'who are you' with the new nonce)
1644
1651
        self.assertEqual(2, self.server.auth_required_errors)
1645
1652
 
1646
 
    def test_user_from_auth_conf(self):
1647
 
        if self._testing_pycurl():
1648
 
            raise tests.TestNotApplicable(
1649
 
                'pycurl does not support authentication.conf')
1650
 
        user = 'joe'
1651
 
        password = 'foo'
1652
 
        self.server.add_user(user, password)
1653
 
        _setup_authentication_config(
1654
 
            scheme='http', 
1655
 
            port=self.server.port,
1656
 
            user=user,
1657
 
            password=password)
1658
 
        t = self.get_user_transport(None, None)
1659
 
        # Issue a request to the server to connect
1660
 
        self.assertEqual('contents of a\n', t.get('a').read())
1661
 
        # Only one 'Authentication Required' error should occur
1662
 
        self.assertEqual(1, self.server.auth_required_errors)
1663
 
 
1664
 
 
1665
 
def _setup_authentication_config(**kwargs):
1666
 
    conf = config.AuthenticationConfig()
1667
 
    conf._get_config().update({'httptest': kwargs})
1668
 
    conf._save()
1669
 
 
1670
 
 
1671
 
 
1672
 
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1673
 
    """Unit tests for glue by which urllib2 asks us for authentication"""
1674
 
 
1675
 
    def test_get_user_password_without_port(self):
1676
 
        """We cope if urllib2 doesn't tell us the port.
1677
 
 
1678
 
        See https://bugs.launchpad.net/bzr/+bug/654684
1679
 
        """
1680
 
        user = 'joe'
1681
 
        password = 'foo'
1682
 
        _setup_authentication_config(
1683
 
            scheme='http', 
1684
 
            host='localhost',
1685
 
            user=user,
1686
 
            password=password)
1687
 
        handler = _urllib2_wrappers.HTTPAuthHandler()
1688
 
        got_pass = handler.get_user_password(dict(
1689
 
            user='joe',
1690
 
            protocol='http',
1691
 
            host='localhost',
1692
 
            path='/',
1693
 
            realm='Realm',
1694
 
            ))
1695
 
        self.assertEquals((user, password), got_pass)
1696
1653
 
1697
1654
 
1698
1655
class TestProxyAuth(TestAuth):
1699
1656
    """Test proxy authentication schemes."""
1700
1657
 
1701
 
    scenarios = multiply_scenarios(
1702
 
        vary_by_http_client_implementation(),
1703
 
        vary_by_http_protocol_version(),
1704
 
        vary_by_http_proxy_auth_scheme(),
1705
 
        )
1706
 
 
1707
1658
    _auth_header = 'Proxy-authorization'
1708
1659
    _password_prompt_prefix = 'Proxy '
1709
1660
    _username_prompt_prefix = 'Proxy '
1710
1661
 
1711
1662
    def setUp(self):
1712
1663
        super(TestProxyAuth, self).setUp()
 
1664
        self._old_env = {}
 
1665
        self.addCleanup(self._restore_env)
1713
1666
        # Override the contents to avoid false positives
1714
1667
        self.build_tree_contents([('a', 'not proxied contents of a\n'),
1715
1668
                                  ('b', 'not proxied contents of b\n'),
1718
1671
                                  ])
1719
1672
 
1720
1673
    def get_user_transport(self, user, password):
1721
 
        self.overrideEnv('all_proxy', self.get_user_url(user, password))
 
1674
        self._install_env({'all_proxy': self.get_user_url(user, password)})
1722
1675
        return TestAuth.get_user_transport(self, user, password)
1723
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
 
1724
1685
    def test_empty_pass(self):
1725
1686
        if self._testing_pycurl():
1726
1687
            import pycurl
1755
1716
 
1756
1717
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1757
1718
 
1758
 
    scenarios = multiply_scenarios(
1759
 
        vary_by_http_client_implementation(), 
1760
 
        vary_by_http_protocol_version(),
1761
 
        )
1762
 
 
1763
1719
    def setUp(self):
1764
1720
        super(SmartHTTPTunnellingTest, self).setUp()
1765
1721
        # We use the VFS layer as part of HTTP tunnelling tests.
1766
 
        self.overrideEnv('BZR_NO_SMART_VFS', None)
 
1722
        self._captureVar('BZR_NO_SMART_VFS', None)
1767
1723
        self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1768
1724
        self.http_server = self.get_readonly_server()
1769
1725
 
1854
1810
 
1855
1811
class Test_redirected_to(tests.TestCase):
1856
1812
 
1857
 
    scenarios = vary_by_http_client_implementation()
1858
 
 
1859
1813
    def test_redirected_to_subdir(self):
1860
1814
        t = self._transport('http://www.example.com/foo')
1861
1815
        r = t._redirected_to('http://www.example.com/foo',
2107
2061
 
2108
2062
class TestActivity(tests.TestCase, TestActivityMixin):
2109
2063
 
2110
 
    scenarios = multiply_scenarios(
2111
 
        vary_by_http_activity(),
2112
 
        vary_by_http_protocol_version(),
2113
 
        )
2114
 
 
2115
2064
    def setUp(self):
2116
2065
        TestActivityMixin.setUp(self)
2117
2066
 
2138
2087
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2139
2088
    """Test authentication on the redirected http server."""
2140
2089
 
2141
 
    scenarios = vary_by_http_protocol_version()
2142
 
 
2143
2090
    _auth_header = 'Authorization'
2144
2091
    _password_prompt_prefix = ''
2145
2092
    _username_prompt_prefix = ''