← 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: Andrew Bennetts
  • Date: 2010-09-24 02:19:53 UTC
  • mto: This revision was merged to the branch mainline in revision 5443.
  • Revision ID: andrew.bennetts@canonical.com-20100924021953-0cg1kjtifuvkbyzp
Fix typo in test method name.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
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
39
42
    tests,
40
43
    transport,
41
44
    ui,
 
45
    urlutils,
42
46
    )
43
47
from bzrlib.tests import (
44
48
    features,
46
50
    http_utils,
47
51
    test_server,
48
52
    )
49
 
from bzrlib.tests.scenarios import (
50
 
    load_tests_apply_scenarios,
51
 
    multiply_scenarios,
52
 
    )
53
53
from bzrlib.transport import (
54
54
    http,
55
55
    remote,
64
64
    from bzrlib.transport.http._pycurl import PyCurlTransport
65
65
 
66
66
 
67
 
load_tests = load_tests_apply_scenarios
68
 
 
69
 
 
70
 
def vary_by_http_client_implementation():
71
 
    """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
                )))
72
78
    transport_scenarios = [
73
79
        ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
74
80
                        _server=http_server.HttpServer_urllib,
79
85
            ('pycurl', dict(_transport=PyCurlTransport,
80
86
                            _server=http_server.HttpServer_PyCurl,
81
87
                            _url_protocol='http+pycurl',)))
82
 
    return transport_scenarios
83
 
 
84
 
 
85
 
def vary_by_http_protocol_version():
86
 
    """Test on http/1.0 and 1.1"""
87
 
    return [
88
 
        ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
89
 
        ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
90
 
        ]
91
 
 
92
 
 
93
 
def vary_by_http_proxy_auth_scheme():
94
 
    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 = [
95
126
        ('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
96
127
        ('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
97
128
        ('basicdigest',
98
 
            dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
 
129
         dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
99
130
        ]
100
 
 
101
 
 
102
 
def vary_by_http_auth_scheme():
103
 
    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 = [
104
141
        ('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
105
142
        ('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
106
143
        ('basicdigest',
107
 
            dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
 
144
         dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
108
145
        ]
109
 
 
110
 
 
111
 
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
                )))
112
155
    activity_scenarios = [
113
156
        ('urllib,http', dict(_activity_server=ActivityHTTPServer,
114
 
                            _transport=_urllib.HttpTransport_urllib,)),
 
157
                             _transport=_urllib.HttpTransport_urllib,)),
115
158
        ]
116
159
    if tests.HTTPSServerFeature.available():
117
160
        activity_scenarios.append(
118
161
            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,
119
 
                                _transport=_urllib.HttpTransport_urllib,)),)
 
162
                                  _transport=_urllib.HttpTransport_urllib,)),)
120
163
    if features.pycurl.available():
121
164
        activity_scenarios.append(
122
165
            ('pycurl,http', dict(_activity_server=ActivityHTTPServer,
123
 
                                _transport=PyCurlTransport,)),)
 
166
                                 _transport=PyCurlTransport,)),)
124
167
        if tests.HTTPSServerFeature.available():
125
168
            from bzrlib.tests import (
126
169
                ssl_certs,
138
181
 
139
182
            activity_scenarios.append(
140
183
                ('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
141
 
                                    _transport=HTTPS_pycurl_transport,)),)
142
 
    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
143
194
 
144
195
 
145
196
class FakeManager(object):
350
401
class TestHttpTransportUrls(tests.TestCase):
351
402
    """Test the http urls."""
352
403
 
353
 
    scenarios = vary_by_http_client_implementation()
354
 
 
355
404
    def test_abs_url(self):
356
405
        """Construction of absolute http URLs"""
357
 
        t = self._transport('http://example.com/bzr/bzr.dev/')
 
406
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
358
407
        eq = self.assertEqualDiff
359
 
        eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
360
 
        eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
361
 
        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')
362
411
        eq(t.abspath('.bzr/1//2/./3'),
363
 
           'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
 
412
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
364
413
 
365
414
    def test_invalid_http_urls(self):
366
415
        """Trap invalid construction of urls"""
367
 
        self._transport('http://example.com/bzr/bzr.dev/')
 
416
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
368
417
        self.assertRaises(errors.InvalidURL,
369
418
                          self._transport,
370
 
                          'http://http://example.com/bzr/bzr.dev/')
 
419
                          'http://http://bazaar-vcs.org/bzr/bzr.dev/')
371
420
 
372
421
    def test_http_root_urls(self):
373
422
        """Construction of URLs from server root"""
374
 
        t = self._transport('http://example.com/')
 
423
        t = self._transport('http://bzr.ozlabs.org/')
375
424
        eq = self.assertEqualDiff
376
425
        eq(t.abspath('.bzr/tree-version'),
377
 
           'http://example.com/.bzr/tree-version')
 
426
           'http://bzr.ozlabs.org/.bzr/tree-version')
378
427
 
379
428
    def test_http_impl_urls(self):
380
429
        """There are servers which ask for particular clients to connect"""
426
475
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
427
476
    """Test the http connections."""
428
477
 
429
 
    scenarios = multiply_scenarios(
430
 
        vary_by_http_client_implementation(), 
431
 
        vary_by_http_protocol_version(),
432
 
        )
433
 
 
434
478
    def setUp(self):
435
479
        http_utils.TestCaseWithWebserver.setUp(self)
436
480
        self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
481
525
class TestHttpTransportRegistration(tests.TestCase):
482
526
    """Test registrations of various http implementations"""
483
527
 
484
 
    scenarios = vary_by_http_client_implementation()
485
 
 
486
528
    def test_http_registered(self):
487
529
        t = transport.get_transport('%s://foo.com/' % self._url_protocol)
488
530
        self.assertIsInstance(t, transport.Transport)
491
533
 
492
534
class TestPost(tests.TestCase):
493
535
 
494
 
    scenarios = multiply_scenarios(
495
 
        vary_by_http_client_implementation(), 
496
 
        vary_by_http_protocol_version(),
497
 
        )
498
 
 
499
536
    def test_post_body_is_received(self):
500
537
        server = RecordingServer(expect_body_tail='end-of-body',
501
538
                                 scheme=self._url_protocol)
507
544
        self.assertTrue(
508
545
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
509
546
        self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
510
 
        self.assertTrue('content-type: application/octet-stream\r'
511
 
                        in server.received_bytes.lower())
512
547
        # The transport should not be assuming that the server can accept
513
548
        # chunked encoding the first time it connects, because HTTP/1.1, so we
514
549
        # check for the literal string.
550
585
    Daughter classes are expected to override _req_handler_class
551
586
    """
552
587
 
553
 
    scenarios = multiply_scenarios(
554
 
        vary_by_http_client_implementation(), 
555
 
        vary_by_http_protocol_version(),
556
 
        )
557
 
 
558
588
    # Provide a useful default
559
589
    _req_handler_class = http_server.TestingHTTPRequestHandler
560
590
 
811
841
        t = self.get_readonly_transport()
812
842
        # force transport to issue multiple requests
813
843
        t._get_max_size = 2
814
 
        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))))
815
845
        # The server should have issued 3 requests
816
846
        self.assertEqual(3, server.GET_request_nb)
817
847
        self.assertEqual('0123456789', t.get_bytes('a'))
894
924
    def get_multiple_ranges(self, file, file_size, ranges):
895
925
        self.send_response(206)
896
926
        self.send_header('Accept-Ranges', 'bytes')
897
 
        # XXX: this is strange; the 'random' name below seems undefined and
898
 
        # yet the tests pass -- mbp 2010-10-11 bug 658773
899
927
        boundary = "%d" % random.randint(0,0x7FFFFFFF)
900
928
        self.send_header("Content-Type",
901
929
                         "multipart/byteranges; boundary=%s" % boundary)
963
991
                return
964
992
            self.send_range_content(file, start, end - start + 1)
965
993
            cur += 1
966
 
        # Final boundary
 
994
        # No final boundary
967
995
        self.wfile.write(boundary_line)
968
996
 
969
997
 
998
1026
        # that mode
999
1027
        self.assertEqual('single', t._range_hint)
1000
1028
 
1001
 
 
1002
1029
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1003
1030
    """Errors out when range specifiers exceed the limit"""
1004
1031
 
1028
1055
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1029
1056
    """Tests readv requests against a server erroring out on too much ranges."""
1030
1057
 
1031
 
    scenarios = multiply_scenarios(
1032
 
        vary_by_http_client_implementation(), 
1033
 
        vary_by_http_protocol_version(),
1034
 
        )
1035
 
 
1036
1058
    # Requests with more range specifiers will error out
1037
1059
    range_limit = 3
1038
1060
 
1073
1095
    Only the urllib implementation is tested here.
1074
1096
    """
1075
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
 
1076
1111
    def _proxied_request(self):
1077
1112
        handler = _urllib2_wrappers.ProxyHandler()
1078
1113
        request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
1080
1115
        return request
1081
1116
 
1082
1117
    def test_empty_user(self):
1083
 
        self.overrideEnv('http_proxy', 'http://bar.com')
 
1118
        self._install_env({'http_proxy': 'http://bar.com'})
1084
1119
        request = self._proxied_request()
1085
1120
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
1086
1121
 
1087
1122
    def test_invalid_proxy(self):
1088
1123
        """A proxy env variable without scheme"""
1089
 
        self.overrideEnv('http_proxy', 'host:1234')
 
1124
        self._install_env({'http_proxy': 'host:1234'})
1090
1125
        self.assertRaises(errors.InvalidURL, self._proxied_request)
1091
1126
 
1092
1127
 
1099
1134
    to the file names).
1100
1135
    """
1101
1136
 
1102
 
    scenarios = multiply_scenarios(
1103
 
        vary_by_http_client_implementation(), 
1104
 
        vary_by_http_protocol_version(),
1105
 
        )
1106
 
 
1107
1137
    # FIXME: We don't have an https server available, so we don't
1108
1138
    # test https connections. --vila toolongago
1109
1139
 
1123
1153
            self.no_proxy_host = self.server_host_port
1124
1154
        # The secondary server is the proxy
1125
1155
        self.proxy_url = self.get_secondary_url()
 
1156
        self._old_env = {}
1126
1157
 
1127
1158
    def _testing_pycurl(self):
1128
1159
        # TODO: This is duplicated for lots of the classes in this file
1129
1160
        return (features.pycurl.available()
1130
1161
                and self._transport == PyCurlTransport)
1131
1162
 
1132
 
    def assertProxied(self):
1133
 
        t = self.get_readonly_transport()
1134
 
        self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1135
 
 
1136
 
    def assertNotProxied(self):
1137
 
        t = self.get_readonly_transport()
1138
 
        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()
1139
1186
 
1140
1187
    def test_http_proxy(self):
1141
 
        self.overrideEnv('http_proxy', self.proxy_url)
1142
 
        self.assertProxied()
 
1188
        self.proxied_in_env({'http_proxy': self.proxy_url})
1143
1189
 
1144
1190
    def test_HTTP_PROXY(self):
1145
1191
        if self._testing_pycurl():
1148
1194
            # about. Should we ?)
1149
1195
            raise tests.TestNotApplicable(
1150
1196
                'pycurl does not check HTTP_PROXY for security reasons')
1151
 
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
1152
 
        self.assertProxied()
 
1197
        self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
1153
1198
 
1154
1199
    def test_all_proxy(self):
1155
 
        self.overrideEnv('all_proxy', self.proxy_url)
1156
 
        self.assertProxied()
 
1200
        self.proxied_in_env({'all_proxy': self.proxy_url})
1157
1201
 
1158
1202
    def test_ALL_PROXY(self):
1159
 
        self.overrideEnv('ALL_PROXY', self.proxy_url)
1160
 
        self.assertProxied()
 
1203
        self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1161
1204
 
1162
1205
    def test_http_proxy_with_no_proxy(self):
1163
 
        self.overrideEnv('no_proxy', self.no_proxy_host)
1164
 
        self.overrideEnv('http_proxy', self.proxy_url)
1165
 
        self.assertNotProxied()
 
1206
        self.not_proxied_in_env({'http_proxy': self.proxy_url,
 
1207
                                 'no_proxy': self.no_proxy_host})
1166
1208
 
1167
1209
    def test_HTTP_PROXY_with_NO_PROXY(self):
1168
1210
        if self._testing_pycurl():
1169
1211
            raise tests.TestNotApplicable(
1170
1212
                'pycurl does not check HTTP_PROXY for security reasons')
1171
 
        self.overrideEnv('NO_PROXY', self.no_proxy_host)
1172
 
        self.overrideEnv('HTTP_PROXY', self.proxy_url)
1173
 
        self.assertNotProxied()
 
1213
        self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
 
1214
                                 'NO_PROXY': self.no_proxy_host})
1174
1215
 
1175
1216
    def test_all_proxy_with_no_proxy(self):
1176
 
        self.overrideEnv('no_proxy', self.no_proxy_host)
1177
 
        self.overrideEnv('all_proxy', self.proxy_url)
1178
 
        self.assertNotProxied()
 
1217
        self.not_proxied_in_env({'all_proxy': self.proxy_url,
 
1218
                                 'no_proxy': self.no_proxy_host})
1179
1219
 
1180
1220
    def test_ALL_PROXY_with_NO_PROXY(self):
1181
 
        self.overrideEnv('NO_PROXY', self.no_proxy_host)
1182
 
        self.overrideEnv('ALL_PROXY', self.proxy_url)
1183
 
        self.assertNotProxied()
 
1221
        self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
 
1222
                                 'NO_PROXY': self.no_proxy_host})
1184
1223
 
1185
1224
    def test_http_proxy_without_scheme(self):
1186
 
        self.overrideEnv('http_proxy', self.server_host_port)
1187
1225
        if self._testing_pycurl():
1188
1226
            # pycurl *ignores* invalid proxy env variables. If that ever change
1189
1227
            # in the future, this test will fail indicating that pycurl do not
1190
1228
            # ignore anymore such variables.
1191
 
            self.assertNotProxied()
 
1229
            self.not_proxied_in_env({'http_proxy': self.server_host_port})
1192
1230
        else:
1193
 
            self.assertRaises(errors.InvalidURL, self.assertProxied)
 
1231
            self.assertRaises(errors.InvalidURL,
 
1232
                              self.proxied_in_env,
 
1233
                              {'http_proxy': self.server_host_port})
1194
1234
 
1195
1235
 
1196
1236
class TestRanges(http_utils.TestCaseWithWebserver):
1197
1237
    """Test the Range header in GET methods."""
1198
1238
 
1199
 
    scenarios = multiply_scenarios(
1200
 
        vary_by_http_client_implementation(), 
1201
 
        vary_by_http_protocol_version(),
1202
 
        )
1203
 
 
1204
1239
    def setUp(self):
1205
1240
        http_utils.TestCaseWithWebserver.setUp(self)
1206
1241
        self.build_tree_contents([('a', '0123456789')],)
1246
1281
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1247
1282
    """Test redirection between http servers."""
1248
1283
 
1249
 
    scenarios = multiply_scenarios(
1250
 
        vary_by_http_client_implementation(), 
1251
 
        vary_by_http_protocol_version(),
1252
 
        )
1253
 
 
1254
1284
    def setUp(self):
1255
1285
        super(TestHTTPRedirections, self).setUp()
1256
1286
        self.build_tree_contents([('a', '0123456789'),
1319
1349
    -- vila 20070212
1320
1350
    """
1321
1351
 
1322
 
    scenarios = multiply_scenarios(
1323
 
        vary_by_http_client_implementation(), 
1324
 
        vary_by_http_protocol_version(),
1325
 
        )
1326
 
 
1327
1352
    def setUp(self):
1328
1353
        if (features.pycurl.available()
1329
1354
            and self._transport == PyCurlTransport):
1374
1399
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1375
1400
    """Test transport.do_catching_redirections."""
1376
1401
 
1377
 
    scenarios = multiply_scenarios(
1378
 
        vary_by_http_client_implementation(), 
1379
 
        vary_by_http_protocol_version(),
1380
 
        )
1381
 
 
1382
1402
    def setUp(self):
1383
1403
        super(TestDoCatchRedirections, self).setUp()
1384
1404
        self.build_tree_contents([('a', '0123456789'),],)
1426
1446
class TestAuth(http_utils.TestCaseWithWebserver):
1427
1447
    """Test authentication scheme"""
1428
1448
 
1429
 
    scenarios = multiply_scenarios(
1430
 
        vary_by_http_client_implementation(),
1431
 
        vary_by_http_protocol_version(),
1432
 
        vary_by_http_auth_scheme(),
1433
 
        )
1434
 
 
1435
1449
    _auth_header = 'Authorization'
1436
1450
    _password_prompt_prefix = ''
1437
1451
    _username_prompt_prefix = ''
1584
1598
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1585
1599
                                            stderr=tests.StringIOWrapper())
1586
1600
        # Create a minimal config file with the right password
1587
 
        _setup_authentication_config(
1588
 
            scheme='http', 
1589
 
            port=self.server.port,
1590
 
            user=user,
1591
 
            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()
1592
1606
        # Issue a request to the server to connect
1593
1607
        self.assertEqual('contents of a\n',t.get('a').read())
1594
1608
        # stdin should have  been left untouched
1596
1610
        # Only one 'Authentication Required' error should occur
1597
1611
        self.assertEqual(1, self.server.auth_required_errors)
1598
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
 
1599
1632
    def test_changing_nonce(self):
1600
1633
        if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1601
1634
                                     http_utils.ProxyDigestAuthServer):
1617
1650
        # initial 'who are you' and a second 'who are you' with the new nonce)
1618
1651
        self.assertEqual(2, self.server.auth_required_errors)
1619
1652
 
1620
 
    def test_user_from_auth_conf(self):
1621
 
        if self._testing_pycurl():
1622
 
            raise tests.TestNotApplicable(
1623
 
                'pycurl does not support authentication.conf')
1624
 
        user = 'joe'
1625
 
        password = 'foo'
1626
 
        self.server.add_user(user, password)
1627
 
        _setup_authentication_config(
1628
 
            scheme='http', 
1629
 
            port=self.server.port,
1630
 
            user=user,
1631
 
            password=password)
1632
 
        t = self.get_user_transport(None, None)
1633
 
        # Issue a request to the server to connect
1634
 
        self.assertEqual('contents of a\n', t.get('a').read())
1635
 
        # Only one 'Authentication Required' error should occur
1636
 
        self.assertEqual(1, self.server.auth_required_errors)
1637
 
 
1638
 
 
1639
 
def _setup_authentication_config(**kwargs):
1640
 
    conf = config.AuthenticationConfig()
1641
 
    conf._get_config().update({'httptest': kwargs})
1642
 
    conf._save()
1643
 
 
1644
 
 
1645
 
 
1646
 
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1647
 
    """Unit tests for glue by which urllib2 asks us for authentication"""
1648
 
 
1649
 
    def test_get_user_password_without_port(self):
1650
 
        """We cope if urllib2 doesn't tell us the port.
1651
 
 
1652
 
        See https://bugs.launchpad.net/bzr/+bug/654684
1653
 
        """
1654
 
        user = 'joe'
1655
 
        password = 'foo'
1656
 
        _setup_authentication_config(
1657
 
            scheme='http', 
1658
 
            host='localhost',
1659
 
            user=user,
1660
 
            password=password)
1661
 
        handler = _urllib2_wrappers.HTTPAuthHandler()
1662
 
        got_pass = handler.get_user_password(dict(
1663
 
            user='joe',
1664
 
            protocol='http',
1665
 
            host='localhost',
1666
 
            path='/',
1667
 
            realm='Realm',
1668
 
            ))
1669
 
        self.assertEquals((user, password), got_pass)
1670
1653
 
1671
1654
 
1672
1655
class TestProxyAuth(TestAuth):
1673
1656
    """Test proxy authentication schemes."""
1674
1657
 
1675
 
    scenarios = multiply_scenarios(
1676
 
        vary_by_http_client_implementation(),
1677
 
        vary_by_http_protocol_version(),
1678
 
        vary_by_http_proxy_auth_scheme(),
1679
 
        )
1680
 
 
1681
1658
    _auth_header = 'Proxy-authorization'
1682
1659
    _password_prompt_prefix = 'Proxy '
1683
1660
    _username_prompt_prefix = 'Proxy '
1684
1661
 
1685
1662
    def setUp(self):
1686
1663
        super(TestProxyAuth, self).setUp()
 
1664
        self._old_env = {}
 
1665
        self.addCleanup(self._restore_env)
1687
1666
        # Override the contents to avoid false positives
1688
1667
        self.build_tree_contents([('a', 'not proxied contents of a\n'),
1689
1668
                                  ('b', 'not proxied contents of b\n'),
1692
1671
                                  ])
1693
1672
 
1694
1673
    def get_user_transport(self, user, password):
1695
 
        self.overrideEnv('all_proxy', self.get_user_url(user, password))
 
1674
        self._install_env({'all_proxy': self.get_user_url(user, password)})
1696
1675
        return TestAuth.get_user_transport(self, user, password)
1697
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
 
1698
1685
    def test_empty_pass(self):
1699
1686
        if self._testing_pycurl():
1700
1687
            import pycurl
1729
1716
 
1730
1717
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1731
1718
 
1732
 
    scenarios = multiply_scenarios(
1733
 
        vary_by_http_client_implementation(), 
1734
 
        vary_by_http_protocol_version(),
1735
 
        )
1736
 
 
1737
1719
    def setUp(self):
1738
1720
        super(SmartHTTPTunnellingTest, self).setUp()
1739
1721
        # We use the VFS layer as part of HTTP tunnelling tests.
1740
 
        self.overrideEnv('BZR_NO_SMART_VFS', None)
 
1722
        self._captureVar('BZR_NO_SMART_VFS', None)
1741
1723
        self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1742
1724
        self.http_server = self.get_readonly_server()
1743
1725
 
1828
1810
 
1829
1811
class Test_redirected_to(tests.TestCase):
1830
1812
 
1831
 
    scenarios = vary_by_http_client_implementation()
1832
 
 
1833
1813
    def test_redirected_to_subdir(self):
1834
1814
        t = self._transport('http://www.example.com/foo')
1835
1815
        r = t._redirected_to('http://www.example.com/foo',
2081
2061
 
2082
2062
class TestActivity(tests.TestCase, TestActivityMixin):
2083
2063
 
2084
 
    scenarios = multiply_scenarios(
2085
 
        vary_by_http_activity(),
2086
 
        vary_by_http_protocol_version(),
2087
 
        )
2088
 
 
2089
2064
    def setUp(self):
2090
2065
        TestActivityMixin.setUp(self)
2091
2066
 
2112
2087
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2113
2088
    """Test authentication on the redirected http server."""
2114
2089
 
2115
 
    scenarios = vary_by_http_protocol_version()
2116
 
 
2117
2090
    _auth_header = 'Authorization'
2118
2091
    _password_prompt_prefix = ''
2119
2092
    _username_prompt_prefix = ''