← 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: 2011-02-11 16:15:39 UTC
  • mto: This revision was merged to the branch mainline in revision 5661.
  • Revision ID: v.ladeuil+lp@free.fr-20110211161539-26o5a28ihyemvuzg
Fix pqm failure on python2.4.

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