← 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: Canonical.com Patch Queue Manager
  • Date: 2011-01-11 15:45:55 UTC
  • mfrom: (5588.1.1 bzr.dev)
  • Revision ID: pqm@pqm.ubuntu.com-20110111154555-xmu386vs0qoigyhl
(vila) Document the packaging helper scripts. (Max Bowsher)

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