/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: John Arbash Meinel
  • Date: 2008-07-09 21:42:24 UTC
  • mto: This revision was merged to the branch mainline in revision 3543.
  • Revision ID: john@arbash-meinel.com-20080709214224-r75k87r6a01pfc3h
Restore a real weave merge to 'bzr merge --weave'.

To do so efficiently, we only add the simple LCAs to the final weave
object, unless we run into complexities with the merge graph.
This gives the same effective result as adding all the texts,
with the advantage of not having to extract all of them.

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, 2006 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for HTTP implementations.
18
18
 
34
34
 
35
35
import bzrlib
36
36
from bzrlib import (
37
 
    bzrdir,
38
37
    config,
39
38
    errors,
40
39
    osutils,
41
 
    remote as _mod_remote,
42
40
    tests,
43
41
    transport,
44
42
    ui,
45
43
    urlutils,
46
44
    )
47
 
from bzrlib.symbol_versioning import (
48
 
    deprecated_in,
49
 
    )
50
45
from bzrlib.tests import (
51
 
    features,
52
46
    http_server,
53
47
    http_utils,
54
48
    )
62
56
    )
63
57
 
64
58
 
65
 
if features.pycurl.available():
 
59
try:
66
60
    from bzrlib.transport.http._pycurl import PyCurlTransport
67
 
 
 
61
    pycurl_present = True
 
62
except errors.DependencyNotPresent:
 
63
    pycurl_present = False
 
64
 
 
65
 
 
66
class TransportAdapter(tests.TestScenarioApplier):
 
67
    """Generate the same test for each transport implementation."""
 
68
 
 
69
    def __init__(self):
 
70
        transport_scenarios = [
 
71
            ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
 
72
                            _server=http_server.HttpServer_urllib,
 
73
                            _qualified_prefix='http+urllib',)),
 
74
            ]
 
75
        if pycurl_present:
 
76
            transport_scenarios.append(
 
77
                ('pycurl', dict(_transport=PyCurlTransport,
 
78
                                _server=http_server.HttpServer_PyCurl,
 
79
                                _qualified_prefix='http+pycurl',)))
 
80
        self.scenarios = transport_scenarios
 
81
 
 
82
 
 
83
class TransportProtocolAdapter(TransportAdapter):
 
84
    """Generate the same test for each protocol implementation.
 
85
 
 
86
    In addition to the transport adaptatation that we inherit from.
 
87
    """
 
88
 
 
89
    def __init__(self):
 
90
        super(TransportProtocolAdapter, self).__init__()
 
91
        protocol_scenarios = [
 
92
            ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
 
93
            ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
 
94
            ]
 
95
        self.scenarios = tests.multiply_scenarios(self.scenarios,
 
96
                                                  protocol_scenarios)
 
97
 
 
98
 
 
99
class TransportProtocolAuthenticationAdapter(TransportProtocolAdapter):
 
100
    """Generate the same test for each authentication scheme implementation.
 
101
 
 
102
    In addition to the protocol adaptatation that we inherit from.
 
103
    """
 
104
 
 
105
    def __init__(self):
 
106
        super(TransportProtocolAuthenticationAdapter, self).__init__()
 
107
        auth_scheme_scenarios = [
 
108
            ('basic', dict(_auth_scheme='basic')),
 
109
            ('digest', dict(_auth_scheme='digest')),
 
110
            ]
 
111
 
 
112
        self.scenarios = tests.multiply_scenarios(self.scenarios,
 
113
                                                  auth_scheme_scenarios)
68
114
 
69
115
def load_tests(standard_tests, module, loader):
70
116
    """Multiply tests for http clients and protocol versions."""
71
 
    result = loader.suiteClass()
72
 
 
73
 
    # one for each transport implementation
74
 
    t_tests, remaining_tests = tests.split_suite_by_condition(
75
 
        standard_tests, tests.condition_isinstance((
76
 
                TestHttpTransportRegistration,
 
117
    # one for each transport
 
118
    t_adapter = TransportAdapter()
 
119
    t_classes= (TestHttpTransportRegistration,
77
120
                TestHttpTransportUrls,
78
 
                Test_redirected_to,
79
 
                )))
80
 
    transport_scenarios = [
81
 
        ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
82
 
                        _server=http_server.HttpServer_urllib,
83
 
                        _qualified_prefix='http+urllib',)),
84
 
        ]
85
 
    if features.pycurl.available():
86
 
        transport_scenarios.append(
87
 
            ('pycurl', dict(_transport=PyCurlTransport,
88
 
                            _server=http_server.HttpServer_PyCurl,
89
 
                            _qualified_prefix='http+pycurl',)))
90
 
    tests.multiply_tests(t_tests, transport_scenarios, result)
91
 
 
92
 
    # each implementation tested with each HTTP version
93
 
    tp_tests, remaining_tests = tests.split_suite_by_condition(
94
 
        remaining_tests, tests.condition_isinstance((
95
 
                SmartHTTPTunnellingTest,
96
 
                TestDoCatchRedirections,
97
 
                TestHTTPConnections,
98
 
                TestHTTPRedirections,
99
 
                TestHTTPSilentRedirections,
100
 
                TestLimitedRangeRequestServer,
101
 
                TestPost,
102
 
                TestProxyHttpServer,
103
 
                TestRanges,
104
 
                TestSpecificRequestHandler,
105
 
                )))
106
 
    protocol_scenarios = [
107
 
            ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
108
 
            ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
109
 
            ]
110
 
    tp_scenarios = tests.multiply_scenarios(transport_scenarios,
111
 
                                            protocol_scenarios)
112
 
    tests.multiply_tests(tp_tests, tp_scenarios, result)
113
 
 
114
 
    # proxy auth: each auth scheme on all http versions on all implementations.
115
 
    tppa_tests, remaining_tests = tests.split_suite_by_condition(
116
 
        remaining_tests, tests.condition_isinstance((
117
 
                TestProxyAuth,
118
 
                )))
119
 
    proxy_auth_scheme_scenarios = [
120
 
        ('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
121
 
        ('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
122
 
        ('basicdigest',
123
 
         dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
124
 
        ]
125
 
    tppa_scenarios = tests.multiply_scenarios(tp_scenarios,
126
 
                                              proxy_auth_scheme_scenarios)
127
 
    tests.multiply_tests(tppa_tests, tppa_scenarios, result)
128
 
 
129
 
    # auth: each auth scheme on all http versions on all implementations.
130
 
    tpa_tests, remaining_tests = tests.split_suite_by_condition(
131
 
        remaining_tests, tests.condition_isinstance((
132
 
                TestAuth,
133
 
                )))
134
 
    auth_scheme_scenarios = [
135
 
        ('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
136
 
        ('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
137
 
        ('basicdigest',
138
 
         dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
139
 
        ]
140
 
    tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
141
 
                                             auth_scheme_scenarios)
142
 
    tests.multiply_tests(tpa_tests, tpa_scenarios, result)
143
 
 
144
 
    # activity: on all http[s] versions on all implementations
145
 
    tpact_tests, remaining_tests = tests.split_suite_by_condition(
146
 
        remaining_tests, tests.condition_isinstance((
147
 
                TestActivity,
148
 
                )))
149
 
    activity_scenarios = [
150
 
        ('urllib,http', dict(_activity_server=ActivityHTTPServer,
151
 
                             _transport=_urllib.HttpTransport_urllib,)),
152
 
        ]
153
 
    if tests.HTTPSServerFeature.available():
154
 
        activity_scenarios.append(
155
 
            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,
156
 
                                  _transport=_urllib.HttpTransport_urllib,)),)
157
 
    if features.pycurl.available():
158
 
        activity_scenarios.append(
159
 
            ('pycurl,http', dict(_activity_server=ActivityHTTPServer,
160
 
                                 _transport=PyCurlTransport,)),)
161
 
        if tests.HTTPSServerFeature.available():
162
 
            from bzrlib.tests import (
163
 
                ssl_certs,
164
121
                )
165
 
            # FIXME: Until we have a better way to handle self-signed
166
 
            # certificates (like allowing them in a test specific
167
 
            # authentication.conf for example), we need some specialized pycurl
168
 
            # transport for tests.
169
 
            class HTTPS_pycurl_transport(PyCurlTransport):
170
 
 
171
 
                def __init__(self, base, _from_transport=None):
172
 
                    super(HTTPS_pycurl_transport, self).__init__(
173
 
                        base, _from_transport)
174
 
                    self.cabundle = str(ssl_certs.build_path('ca.crt'))
175
 
 
176
 
            activity_scenarios.append(
177
 
                ('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
178
 
                                      _transport=HTTPS_pycurl_transport,)),)
179
 
 
180
 
    tpact_scenarios = tests.multiply_scenarios(activity_scenarios,
181
 
                                               protocol_scenarios)
182
 
    tests.multiply_tests(tpact_tests, tpact_scenarios, result)
183
 
 
184
 
    # No parametrization for the remaining tests
185
 
    result.addTests(remaining_tests)
186
 
 
 
122
    is_testing_for_transports = tests.condition_isinstance(t_classes)
 
123
 
 
124
    # multiplied by one for each protocol version
 
125
    tp_adapter = TransportProtocolAdapter()
 
126
    tp_classes= (SmartHTTPTunnellingTest,
 
127
                 TestDoCatchRedirections,
 
128
                 TestHTTPConnections,
 
129
                 TestHTTPRedirections,
 
130
                 TestHTTPSilentRedirections,
 
131
                 TestLimitedRangeRequestServer,
 
132
                 TestPost,
 
133
                 TestProxyHttpServer,
 
134
                 TestRanges,
 
135
                 TestSpecificRequestHandler,
 
136
                 )
 
137
    is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)
 
138
 
 
139
    # multiplied by one for each authentication scheme
 
140
    tpa_adapter = TransportProtocolAuthenticationAdapter()
 
141
    tpa_classes = (TestAuth,
 
142
                   )
 
143
    is_also_testing_for_authentication = tests.condition_isinstance(
 
144
        tpa_classes)
 
145
 
 
146
    result = loader.suiteClass()
 
147
    for test_class in tests.iter_suite_tests(standard_tests):
 
148
        # Each test class is either standalone or testing for some combination
 
149
        # of transport, protocol version, authentication scheme. Use the right
 
150
        # adpater (or none) depending on the class.
 
151
        if is_testing_for_transports(test_class):
 
152
            result.addTests(t_adapter.adapt(test_class))
 
153
        elif is_also_testing_for_protocols(test_class):
 
154
            result.addTests(tp_adapter.adapt(test_class))
 
155
        elif is_also_testing_for_authentication(test_class):
 
156
            result.addTests(tpa_adapter.adapt(test_class))
 
157
        else:
 
158
            result.addTest(test_class)
187
159
    return result
188
160
 
189
161
 
198
170
 
199
171
class RecordingServer(object):
200
172
    """A fake HTTP server.
201
 
 
 
173
    
202
174
    It records the bytes sent to it, and replies with a 200.
203
175
    """
204
176
 
205
 
    def __init__(self, expect_body_tail=None, scheme=''):
 
177
    def __init__(self, expect_body_tail=None):
206
178
        """Constructor.
207
179
 
208
180
        :type expect_body_tail: str
213
185
        self.host = None
214
186
        self.port = None
215
187
        self.received_bytes = ''
216
 
        self.scheme = scheme
217
 
 
218
 
    def get_url(self):
219
 
        return '%s://%s:%s/' % (self.scheme, self.host, self.port)
220
 
 
221
 
    def start_server(self):
 
188
 
 
189
    def setUp(self):
222
190
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
223
191
        self._sock.bind(('127.0.0.1', 0))
224
192
        self.host, self.port = self._sock.getsockname()
247
215
            # The client may have already closed the socket.
248
216
            pass
249
217
 
250
 
    def stop_server(self):
 
218
    def tearDown(self):
251
219
        try:
252
220
            self._sock.close()
253
221
        except socket.error:
257
225
        self.port = None
258
226
 
259
227
 
260
 
class TestAuthHeader(tests.TestCase):
261
 
 
262
 
    def parse_header(self, header, auth_handler_class=None):
263
 
        if auth_handler_class is None:
264
 
            auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
265
 
        self.auth_handler =  auth_handler_class()
266
 
        return self.auth_handler._parse_auth_header(header)
267
 
 
268
 
    def test_empty_header(self):
269
 
        scheme, remainder = self.parse_header('')
270
 
        self.assertEqual('', scheme)
271
 
        self.assertIs(None, remainder)
272
 
 
273
 
    def test_negotiate_header(self):
274
 
        scheme, remainder = self.parse_header('Negotiate')
275
 
        self.assertEqual('negotiate', scheme)
276
 
        self.assertIs(None, remainder)
277
 
 
278
 
    def test_basic_header(self):
279
 
        scheme, remainder = self.parse_header(
280
 
            'Basic realm="Thou should not pass"')
281
 
        self.assertEqual('basic', scheme)
282
 
        self.assertEqual('realm="Thou should not pass"', remainder)
283
 
 
284
 
    def test_basic_extract_realm(self):
285
 
        scheme, remainder = self.parse_header(
286
 
            'Basic realm="Thou should not pass"',
287
 
            _urllib2_wrappers.BasicAuthHandler)
288
 
        match, realm = self.auth_handler.extract_realm(remainder)
289
 
        self.assertTrue(match is not None)
290
 
        self.assertEqual('Thou should not pass', realm)
291
 
 
292
 
    def test_digest_header(self):
293
 
        scheme, remainder = self.parse_header(
294
 
            'Digest realm="Thou should not pass"')
295
 
        self.assertEqual('digest', scheme)
296
 
        self.assertEqual('realm="Thou should not pass"', remainder)
297
 
 
298
 
 
299
228
class TestHTTPServer(tests.TestCase):
300
229
    """Test the HTTP servers implementations."""
301
230
 
306
235
 
307
236
        server = http_server.HttpServer(BogusRequestHandler)
308
237
        try:
309
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
238
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
310
239
        except:
311
 
            server.stop_server()
 
240
            server.tearDown()
312
241
            self.fail('HTTP Server creation did not raise UnknownProtocol')
313
242
 
314
243
    def test_force_invalid_protocol(self):
315
244
        server = http_server.HttpServer(protocol_version='HTTP/0.1')
316
245
        try:
317
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
246
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
318
247
        except:
319
 
            server.stop_server()
 
248
            server.tearDown()
320
249
            self.fail('HTTP Server creation did not raise UnknownProtocol')
321
250
 
322
251
    def test_server_start_and_stop(self):
323
252
        server = http_server.HttpServer()
324
 
        server.start_server()
325
 
        try:
326
 
            self.assertTrue(server._http_running)
327
 
        finally:
328
 
            server.stop_server()
 
253
        server.setUp()
 
254
        self.assertTrue(server._http_running)
 
255
        server.tearDown()
329
256
        self.assertFalse(server._http_running)
330
257
 
331
258
    def test_create_http_server_one_zero(self):
334
261
            protocol_version = 'HTTP/1.0'
335
262
 
336
263
        server = http_server.HttpServer(RequestHandlerOneZero)
337
 
        self.start_server(server)
 
264
        server.setUp()
 
265
        self.addCleanup(server.tearDown)
338
266
        self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
339
267
 
340
268
    def test_create_http_server_one_one(self):
343
271
            protocol_version = 'HTTP/1.1'
344
272
 
345
273
        server = http_server.HttpServer(RequestHandlerOneOne)
346
 
        self.start_server(server)
 
274
        server.setUp()
 
275
        self.addCleanup(server.tearDown)
347
276
        self.assertIsInstance(server._httpd,
348
277
                              http_server.TestingThreadingHTTPServer)
349
278
 
354
283
 
355
284
        server = http_server.HttpServer(RequestHandlerOneZero,
356
285
                                        protocol_version='HTTP/1.1')
357
 
        self.start_server(server)
 
286
        server.setUp()
 
287
        self.addCleanup(server.tearDown)
358
288
        self.assertIsInstance(server._httpd,
359
289
                              http_server.TestingThreadingHTTPServer)
360
290
 
365
295
 
366
296
        server = http_server.HttpServer(RequestHandlerOneOne,
367
297
                                        protocol_version='HTTP/1.0')
368
 
        self.start_server(server)
 
298
        server.setUp()
 
299
        self.addCleanup(server.tearDown)
369
300
        self.assertIsInstance(server._httpd,
370
301
                              http_server.TestingHTTPServer)
371
302
 
374
305
    """Test case to inherit from if pycurl is present"""
375
306
 
376
307
    def _get_pycurl_maybe(self):
377
 
        self.requireFeature(features.pycurl)
378
 
        return PyCurlTransport
 
308
        try:
 
309
            from bzrlib.transport.http._pycurl import PyCurlTransport
 
310
            return PyCurlTransport
 
311
        except errors.DependencyNotPresent:
 
312
            raise tests.TestSkipped('pycurl not present')
379
313
 
380
314
    _transport = property(_get_pycurl_maybe)
381
315
 
388
322
    def test_url_parsing(self):
389
323
        f = FakeManager()
390
324
        url = http.extract_auth('http://example.com', f)
391
 
        self.assertEqual('http://example.com', url)
392
 
        self.assertEqual(0, len(f.credentials))
 
325
        self.assertEquals('http://example.com', url)
 
326
        self.assertEquals(0, len(f.credentials))
393
327
        url = http.extract_auth(
394
 
            'http://user:pass@example.com/bzr/bzr.dev', f)
395
 
        self.assertEqual('http://example.com/bzr/bzr.dev', url)
396
 
        self.assertEqual(1, len(f.credentials))
397
 
        self.assertEqual([None, 'example.com', 'user', 'pass'],
398
 
                         f.credentials[0])
 
328
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
 
329
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
 
330
        self.assertEquals(1, len(f.credentials))
 
331
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
 
332
                          f.credentials[0])
399
333
 
400
334
 
401
335
class TestHttpTransportUrls(tests.TestCase):
428
362
    def test_http_impl_urls(self):
429
363
        """There are servers which ask for particular clients to connect"""
430
364
        server = self._server()
431
 
        server.start_server()
432
365
        try:
 
366
            server.setUp()
433
367
            url = server.get_url()
434
368
            self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
435
369
        finally:
436
 
            server.stop_server()
 
370
            server.tearDown()
437
371
 
438
372
 
439
373
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
448
382
        https by supplying a fake version_info that do not
449
383
        support it.
450
384
        """
451
 
        self.requireFeature(features.pycurl)
452
 
        # Import the module locally now that we now it's available.
453
 
        pycurl = features.pycurl.module
 
385
        try:
 
386
            import pycurl
 
387
        except ImportError:
 
388
            raise tests.TestSkipped('pycurl not present')
454
389
 
455
 
        self.overrideAttr(pycurl, 'version_info',
456
 
                          # Fake the pycurl version_info This was taken from
457
 
                          # a windows pycurl without SSL (thanks to bialix)
458
 
                          lambda : (2,
459
 
                                    '7.13.2',
460
 
                                    462082,
461
 
                                    'i386-pc-win32',
462
 
                                    2576,
463
 
                                    None,
464
 
                                    0,
465
 
                                    None,
466
 
                                    ('ftp', 'gopher', 'telnet',
467
 
                                     'dict', 'ldap', 'http', 'file'),
468
 
                                    None,
469
 
                                    0,
470
 
                                    None))
471
 
        self.assertRaises(errors.DependencyNotPresent, self._transport,
472
 
                          'https://launchpad.net')
 
390
        version_info_orig = pycurl.version_info
 
391
        try:
 
392
            # Now that we have pycurl imported, we can fake its version_info
 
393
            # This was taken from a windows pycurl without SSL
 
394
            # (thanks to bialix)
 
395
            pycurl.version_info = lambda : (2,
 
396
                                            '7.13.2',
 
397
                                            462082,
 
398
                                            'i386-pc-win32',
 
399
                                            2576,
 
400
                                            None,
 
401
                                            0,
 
402
                                            None,
 
403
                                            ('ftp', 'gopher', 'telnet',
 
404
                                             'dict', 'ldap', 'http', 'file'),
 
405
                                            None,
 
406
                                            0,
 
407
                                            None)
 
408
            self.assertRaises(errors.DependencyNotPresent, self._transport,
 
409
                              'https://launchpad.net')
 
410
        finally:
 
411
            # Restore the right function
 
412
            pycurl.version_info = version_info_orig
473
413
 
474
414
 
475
415
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
507
447
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
508
448
            % bzrlib.__version__) > -1)
509
449
 
 
450
    def test_get_smart_medium(self):
 
451
        # For HTTP, get_smart_medium should return the transport object.
 
452
        server = self.get_readonly_server()
 
453
        http_transport = self._transport(server.get_url())
 
454
        medium = http_transport.get_smart_medium()
 
455
        self.assertIs(medium, http_transport)
 
456
 
510
457
    def test_has_on_bogus_host(self):
511
458
        # Get a free address and don't 'accept' on it, so that we
512
459
        # can be sure there is no http handler there, but set a
534
481
class TestPost(tests.TestCase):
535
482
 
536
483
    def test_post_body_is_received(self):
537
 
        server = RecordingServer(expect_body_tail='end-of-body',
538
 
            scheme=self._qualified_prefix)
539
 
        self.start_server(server)
540
 
        url = server.get_url()
 
484
        server = RecordingServer(expect_body_tail='end-of-body')
 
485
        server.setUp()
 
486
        self.addCleanup(server.tearDown)
 
487
        scheme = self._qualified_prefix
 
488
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
541
489
        http_transport = self._transport(url)
542
490
        code, response = http_transport._post('abc def end-of-body')
543
491
        self.assertTrue(
592
540
                                      protocol_version=self._protocol_version)
593
541
 
594
542
    def _testing_pycurl(self):
595
 
        # TODO: This is duplicated for lots of the classes in this file
596
 
        return (features.pycurl.available()
597
 
                and self._transport == PyCurlTransport)
 
543
        return pycurl_present and self._transport == PyCurlTransport
598
544
 
599
545
 
600
546
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
617
563
        # for details) make no distinction between a closed
618
564
        # socket and badly formatted status line, so we can't
619
565
        # just test for ConnectionError, we have to test
620
 
        # InvalidHttpResponse too. And pycurl may raise ConnectionReset
621
 
        # instead of ConnectionError too.
622
 
        self.assertRaises(( errors.ConnectionError, errors.ConnectionReset,
623
 
                            errors.InvalidHttpResponse),
 
566
        # InvalidHttpResponse too.
 
567
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
624
568
                          t.has, 'foo/bar')
625
569
 
626
570
    def test_http_get(self):
627
571
        server = self.get_readonly_server()
628
572
        t = self._transport(server.get_url())
629
 
        self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
630
 
                           errors.InvalidHttpResponse),
 
573
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
631
574
                          t.get, 'foo/bar')
632
575
 
633
576
 
709
652
    _req_handler_class = BadProtocolRequestHandler
710
653
 
711
654
    def setUp(self):
712
 
        if self._testing_pycurl():
 
655
        if pycurl_present and self._transport == PyCurlTransport:
713
656
            raise tests.TestNotApplicable(
714
657
                "pycurl doesn't check the protocol version")
715
658
        super(TestBadProtocolServer, self).setUp()
759
702
        self.assertEqual(None, server.host)
760
703
        self.assertEqual(None, server.port)
761
704
 
762
 
    def test_setUp_and_stop(self):
 
705
    def test_setUp_and_tearDown(self):
763
706
        server = RecordingServer(expect_body_tail=None)
764
 
        server.start_server()
 
707
        server.setUp()
765
708
        try:
766
709
            self.assertNotEqual(None, server.host)
767
710
            self.assertNotEqual(None, server.port)
768
711
        finally:
769
 
            server.stop_server()
 
712
            server.tearDown()
770
713
        self.assertEqual(None, server.host)
771
714
        self.assertEqual(None, server.port)
772
715
 
773
716
    def test_send_receive_bytes(self):
774
 
        server = RecordingServer(expect_body_tail='c', scheme='http')
775
 
        self.start_server(server)
 
717
        server = RecordingServer(expect_body_tail='c')
 
718
        server.setUp()
 
719
        self.addCleanup(server.tearDown)
776
720
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
777
721
        sock.connect((server.host, server.port))
778
722
        sock.sendall('abc')
872
816
        # bytes on the socket
873
817
        ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
874
818
        self.assertEqual((0, '0'), ireadv.next())
875
 
        # The server should have issued one request so far
 
819
        # The server should have issued one request so far 
876
820
        self.assertEqual(1, server.GET_request_nb)
877
821
        self.assertEqual('0123456789', t.get_bytes('a'))
878
822
        # get_bytes issued an additional request, the readv pending ones are
1169
1113
        if self._testing_pycurl():
1170
1114
            # Oh my ! pycurl does not check for the port as part of
1171
1115
            # no_proxy :-( So we just test the host part
1172
 
            self.no_proxy_host = self.server.host
 
1116
            self.no_proxy_host = 'localhost'
1173
1117
        else:
1174
1118
            self.no_proxy_host = self.proxy_address
1175
1119
        # The secondary server is the proxy
1178
1122
        self._old_env = {}
1179
1123
 
1180
1124
    def _testing_pycurl(self):
1181
 
        # TODO: This is duplicated for lots of the classes in this file
1182
 
        return (features.pycurl.available()
1183
 
                and self._transport == PyCurlTransport)
 
1125
        return pycurl_present and self._transport == PyCurlTransport
1184
1126
 
1185
1127
    def create_transport_secondary_server(self):
1186
1128
        """Creates an http server that will serve files with
1201
1143
        url = self.server.get_url()
1202
1144
        t = self._transport(url)
1203
1145
        try:
1204
 
            self.assertEqual('proxied contents of foo\n', t.get('foo').read())
 
1146
            self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
1205
1147
        finally:
1206
1148
            self._restore_env()
1207
1149
 
1210
1152
        url = self.server.get_url()
1211
1153
        t = self._transport(url)
1212
1154
        try:
1213
 
            self.assertEqual('contents of foo\n', t.get('foo').read())
 
1155
            self.assertEqual(t.get('foo').read(), 'contents of foo\n')
1214
1156
        finally:
1215
1157
            self._restore_env()
1216
1158
 
1326
1268
                                  ('bundle',
1327
1269
                                  '# Bazaar revision bundle v0.9\n#\n')
1328
1270
                                  ],)
1329
 
        # The requests to the old server will be redirected to the new server
 
1271
 
1330
1272
        self.old_transport = self._transport(self.old_server.get_url())
1331
1273
 
1332
1274
    def test_redirected(self):
1337
1279
    def test_read_redirected_bundle_from_url(self):
1338
1280
        from bzrlib.bundle import read_bundle_from_url
1339
1281
        url = self.old_transport.abspath('bundle')
1340
 
        bundle = self.applyDeprecated(deprecated_in((1, 12, 0)),
1341
 
                read_bundle_from_url, url)
 
1282
        bundle = read_bundle_from_url(url)
1342
1283
        # If read_bundle_from_url was successful we get an empty bundle
1343
1284
        self.assertEqual([], bundle.revisions)
1344
1285
 
1355
1296
        # Since the tests using this class will replace
1356
1297
        # _urllib2_wrappers.Request, we can't just call the base class __init__
1357
1298
        # or we'll loop.
1358
 
        RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
 
1299
        RedirectedRequest.init_orig(self, method, url, args, kwargs)
1359
1300
        self.follow_redirections = True
1360
1301
 
1361
1302
 
1362
 
def install_redirected_request(test):
1363
 
    test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1364
 
 
1365
 
 
1366
1303
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1367
1304
    """Test redirections.
1368
1305
 
1378
1315
    """
1379
1316
 
1380
1317
    def setUp(self):
1381
 
        if (features.pycurl.available()
1382
 
            and self._transport == PyCurlTransport):
 
1318
        if pycurl_present and self._transport == PyCurlTransport:
1383
1319
            raise tests.TestNotApplicable(
1384
1320
                "pycurl doesn't redirect silently annymore")
1385
1321
        super(TestHTTPSilentRedirections, self).setUp()
1386
 
        install_redirected_request(self)
 
1322
        self.setup_redirected_request()
 
1323
        self.addCleanup(self.cleanup_redirected_request)
1387
1324
        self.build_tree_contents([('a','a'),
1388
1325
                                  ('1/',),
1389
1326
                                  ('1/a', 'redirected once'),
1399
1336
 
1400
1337
        self.old_transport = self._transport(self.old_server.get_url())
1401
1338
 
 
1339
    def setup_redirected_request(self):
 
1340
        self.original_class = _urllib2_wrappers.Request
 
1341
        _urllib2_wrappers.Request = RedirectedRequest
 
1342
 
 
1343
    def cleanup_redirected_request(self):
 
1344
        _urllib2_wrappers.Request = self.original_class
 
1345
 
1402
1346
    def create_transport_secondary_server(self):
1403
1347
        """Create the secondary server, redirections are defined in the tests"""
1404
1348
        return http_utils.HTTPServerRedirecting(
1408
1352
        t = self.old_transport
1409
1353
 
1410
1354
        req = RedirectedRequest('GET', t.abspath('a'))
 
1355
        req.follow_redirections = True
1411
1356
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1412
1357
                                       self.new_server.port)
1413
1358
        self.old_server.redirections = \
1414
1359
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1415
 
        self.assertEqual('redirected once',t._perform(req).read())
 
1360
        self.assertEquals('redirected once',t._perform(req).read())
1416
1361
 
1417
1362
    def test_five_redirections(self):
1418
1363
        t = self.old_transport
1419
1364
 
1420
1365
        req = RedirectedRequest('GET', t.abspath('a'))
 
1366
        req.follow_redirections = True
1421
1367
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1422
1368
                                       self.old_server.port)
1423
1369
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1429
1375
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1430
1376
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1431
1377
            ]
1432
 
        self.assertEqual('redirected 5 times',t._perform(req).read())
 
1378
        self.assertEquals('redirected 5 times',t._perform(req).read())
1433
1379
 
1434
1380
 
1435
1381
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1448
1394
        t = self._transport(self.new_server.get_url())
1449
1395
 
1450
1396
        # We use None for redirected so that we fail if redirected
1451
 
        self.assertEqual('0123456789',
1452
 
                         transport.do_catching_redirections(
 
1397
        self.assertEquals('0123456789',
 
1398
                          transport.do_catching_redirections(
1453
1399
                self.get_a, t, None).read())
1454
1400
 
1455
1401
    def test_one_redirection(self):
1460
1406
            dir, file = urlutils.split(exception.target)
1461
1407
            return self._transport(dir)
1462
1408
 
1463
 
        self.assertEqual('0123456789',
1464
 
                         transport.do_catching_redirections(
 
1409
        self.assertEquals('0123456789',
 
1410
                          transport.do_catching_redirections(
1465
1411
                self.get_a, self.old_transport, redirected).read())
1466
 
        self.assertEqual(1, self.redirections)
 
1412
        self.assertEquals(1, self.redirections)
1467
1413
 
1468
1414
    def test_redirection_loop(self):
1469
1415
 
1483
1429
 
1484
1430
    _auth_header = 'Authorization'
1485
1431
    _password_prompt_prefix = ''
1486
 
    _username_prompt_prefix = ''
1487
 
    # Set by load_tests
1488
 
    _auth_server = None
1489
1432
 
1490
1433
    def setUp(self):
1491
1434
        super(TestAuth, self).setUp()
1494
1437
                                  ('b', 'contents of b\n'),])
1495
1438
 
1496
1439
    def create_transport_readonly_server(self):
1497
 
        return self._auth_server(protocol_version=self._protocol_version)
 
1440
        if self._auth_scheme == 'basic':
 
1441
            server = http_utils.HTTPBasicAuthServer(
 
1442
                protocol_version=self._protocol_version)
 
1443
        else:
 
1444
            if self._auth_scheme != 'digest':
 
1445
                raise AssertionError('Unknown auth scheme: %r'
 
1446
                                     % self._auth_scheme)
 
1447
            server = http_utils.HTTPDigestAuthServer(
 
1448
                protocol_version=self._protocol_version)
 
1449
        return server
1498
1450
 
1499
1451
    def _testing_pycurl(self):
1500
 
        # TODO: This is duplicated for lots of the classes in this file
1501
 
        return (features.pycurl.available()
1502
 
                and self._transport == PyCurlTransport)
 
1452
        return pycurl_present and self._transport == PyCurlTransport
1503
1453
 
1504
 
    def get_user_url(self, user, password):
 
1454
    def get_user_url(self, user=None, password=None):
1505
1455
        """Build an url embedding user and password"""
1506
1456
        url = '%s://' % self.server._url_protocol
1507
1457
        if user is not None:
1512
1462
        url += '%s:%s/' % (self.server.host, self.server.port)
1513
1463
        return url
1514
1464
 
1515
 
    def get_user_transport(self, user, password):
 
1465
    def get_user_transport(self, user=None, password=None):
1516
1466
        return self._transport(self.get_user_url(user, password))
1517
1467
 
1518
1468
    def test_no_user(self):
1519
1469
        self.server.add_user('joe', 'foo')
1520
 
        t = self.get_user_transport(None, None)
 
1470
        t = self.get_user_transport()
1521
1471
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1522
1472
        # Only one 'Authentication Required' error should occur
1523
1473
        self.assertEqual(1, self.server.auth_required_errors)
1553
1503
        # initial 'who are you' and 'this is not you, who are you')
1554
1504
        self.assertEqual(2, self.server.auth_required_errors)
1555
1505
 
1556
 
    def test_prompt_for_username(self):
1557
 
        if self._testing_pycurl():
1558
 
            raise tests.TestNotApplicable(
1559
 
                'pycurl cannot prompt, it handles auth by embedding'
1560
 
                ' user:pass in urls only')
1561
 
 
1562
 
        self.server.add_user('joe', 'foo')
1563
 
        t = self.get_user_transport(None, None)
1564
 
        stdout = tests.StringIOWrapper()
1565
 
        stderr = tests.StringIOWrapper()
1566
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1567
 
                                            stdout=stdout, stderr=stderr)
1568
 
        self.assertEqual('contents of a\n',t.get('a').read())
1569
 
        # stdin should be empty
1570
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
1571
 
        stderr.seek(0)
1572
 
        expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1573
 
        self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1574
 
        self.assertEqual('', stdout.getvalue())
1575
 
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1576
 
                                    stderr.readline())
1577
 
 
1578
1506
    def test_prompt_for_password(self):
1579
1507
        if self._testing_pycurl():
1580
1508
            raise tests.TestNotApplicable(
1584
1512
        self.server.add_user('joe', 'foo')
1585
1513
        t = self.get_user_transport('joe', None)
1586
1514
        stdout = tests.StringIOWrapper()
1587
 
        stderr = tests.StringIOWrapper()
1588
 
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1589
 
                                            stdout=stdout, stderr=stderr)
1590
 
        self.assertEqual('contents of a\n', t.get('a').read())
 
1515
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
 
1516
        self.assertEqual('contents of a\n',t.get('a').read())
1591
1517
        # stdin should be empty
1592
1518
        self.assertEqual('', ui.ui_factory.stdin.readline())
1593
1519
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1594
 
                                    stderr.getvalue())
1595
 
        self.assertEqual('', stdout.getvalue())
 
1520
                                    stdout.getvalue())
1596
1521
        # And we shouldn't prompt again for a different request
1597
1522
        # against the same transport.
1598
1523
        self.assertEqual('contents of b\n',t.get('b').read())
1608
1533
                              % (scheme.upper(),
1609
1534
                                 user, self.server.host, self.server.port,
1610
1535
                                 self.server.auth_realm)))
1611
 
        self.assertEqual(expected_prompt, actual_prompt)
1612
 
 
1613
 
    def _expected_username_prompt(self, scheme):
1614
 
        return (self._username_prompt_prefix
1615
 
                + "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1616
 
                                 self.server.host, self.server.port,
1617
 
                                 self.server.auth_realm))
 
1536
        self.assertEquals(expected_prompt, actual_prompt)
1618
1537
 
1619
1538
    def test_no_prompt_for_password_when_using_auth_config(self):
1620
1539
        if self._testing_pycurl():
1628
1547
        self.server.add_user(user, password)
1629
1548
        t = self.get_user_transport(user, None)
1630
1549
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1631
 
                                            stderr=tests.StringIOWrapper())
 
1550
                                            stdout=tests.StringIOWrapper())
1632
1551
        # Create a minimal config file with the right password
1633
1552
        conf = config.AuthenticationConfig()
1634
1553
        conf._get_config().update(
1642
1561
        # Only one 'Authentication Required' error should occur
1643
1562
        self.assertEqual(1, self.server.auth_required_errors)
1644
1563
 
1645
 
    def test_user_from_auth_conf(self):
1646
 
        if self._testing_pycurl():
1647
 
            raise tests.TestNotApplicable(
1648
 
                'pycurl does not support authentication.conf')
1649
 
        user = 'joe'
1650
 
        password = 'foo'
1651
 
        self.server.add_user(user, password)
1652
 
        # Create a minimal config file with the right password
1653
 
        conf = config.AuthenticationConfig()
1654
 
        conf._get_config().update(
1655
 
            {'httptest': {'scheme': 'http', 'port': self.server.port,
1656
 
                          'user': user, 'password': password}})
1657
 
        conf._save()
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
1564
    def test_changing_nonce(self):
1665
 
        if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1666
 
                                     http_utils.ProxyDigestAuthServer):
1667
 
            raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
 
1565
        if self._auth_scheme != 'digest':
 
1566
            raise tests.TestNotApplicable('HTTP auth digest only test')
1668
1567
        if self._testing_pycurl():
1669
1568
            raise tests.KnownFailure(
1670
1569
                'pycurl does not handle a nonce change')
1688
1587
    """Test proxy authentication schemes."""
1689
1588
 
1690
1589
    _auth_header = 'Proxy-authorization'
1691
 
    _password_prompt_prefix = 'Proxy '
1692
 
    _username_prompt_prefix = 'Proxy '
 
1590
    _password_prompt_prefix='Proxy '
1693
1591
 
1694
1592
    def setUp(self):
1695
1593
        super(TestProxyAuth, self).setUp()
1702
1600
                                  ('b-proxied', 'contents of b\n'),
1703
1601
                                  ])
1704
1602
 
1705
 
    def get_user_transport(self, user, password):
 
1603
    def create_transport_readonly_server(self):
 
1604
        if self._auth_scheme == 'basic':
 
1605
            server = http_utils.ProxyBasicAuthServer(
 
1606
                protocol_version=self._protocol_version)
 
1607
        else:
 
1608
            if self._auth_scheme != 'digest':
 
1609
                raise AssertionError('Unknown auth scheme: %r'
 
1610
                                     % self._auth_scheme)
 
1611
            server = http_utils.ProxyDigestAuthServer(
 
1612
                protocol_version=self._protocol_version)
 
1613
        return server
 
1614
 
 
1615
    def get_user_transport(self, user=None, password=None):
1706
1616
        self._install_env({'all_proxy': self.get_user_url(user, password)})
1707
1617
        return self._transport(self.server.get_url())
1708
1618
 
1757
1667
        return http_utils.HTTPServerWithSmarts(
1758
1668
            protocol_version=self._protocol_version)
1759
1669
 
1760
 
    def test_open_bzrdir(self):
1761
 
        branch = self.make_branch('relpath')
1762
 
        http_server = self.get_readonly_server()
1763
 
        url = http_server.get_url() + 'relpath'
1764
 
        bd = bzrdir.BzrDir.open(url)
1765
 
        self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1766
 
 
1767
1670
    def test_bulk_data(self):
1768
1671
        # We should be able to send and receive bulk data in a single message.
1769
1672
        # The 'readv' command in the smart protocol both sends and receives
1835
1738
        # No need to build a valid smart request here, the server will not even
1836
1739
        # try to interpret it.
1837
1740
        self.assertRaises(errors.SmartProtocolError,
1838
 
                          t.get_smart_medium().send_http_smart_request,
1839
 
                          'whatever')
1840
 
 
1841
 
class Test_redirected_to(tests.TestCase):
1842
 
 
1843
 
    def test_redirected_to_subdir(self):
1844
 
        t = self._transport('http://www.example.com/foo')
1845
 
        r = t._redirected_to('http://www.example.com/foo',
1846
 
                             'http://www.example.com/foo/subdir')
1847
 
        self.assertIsInstance(r, type(t))
1848
 
        # Both transports share the some connection
1849
 
        self.assertEqual(t._get_connection(), r._get_connection())
1850
 
 
1851
 
    def test_redirected_to_self_with_slash(self):
1852
 
        t = self._transport('http://www.example.com/foo')
1853
 
        r = t._redirected_to('http://www.example.com/foo',
1854
 
                             'http://www.example.com/foo/')
1855
 
        self.assertIsInstance(r, type(t))
1856
 
        # Both transports share the some connection (one can argue that we
1857
 
        # should return the exact same transport here, but that seems
1858
 
        # overkill).
1859
 
        self.assertEqual(t._get_connection(), r._get_connection())
1860
 
 
1861
 
    def test_redirected_to_host(self):
1862
 
        t = self._transport('http://www.example.com/foo')
1863
 
        r = t._redirected_to('http://www.example.com/foo',
1864
 
                             'http://foo.example.com/foo/subdir')
1865
 
        self.assertIsInstance(r, type(t))
1866
 
 
1867
 
    def test_redirected_to_same_host_sibling_protocol(self):
1868
 
        t = self._transport('http://www.example.com/foo')
1869
 
        r = t._redirected_to('http://www.example.com/foo',
1870
 
                             'https://www.example.com/foo')
1871
 
        self.assertIsInstance(r, type(t))
1872
 
 
1873
 
    def test_redirected_to_same_host_different_protocol(self):
1874
 
        t = self._transport('http://www.example.com/foo')
1875
 
        r = t._redirected_to('http://www.example.com/foo',
1876
 
                             'ftp://www.example.com/foo')
1877
 
        self.assertNotEquals(type(r), type(t))
1878
 
 
1879
 
    def test_redirected_to_different_host_same_user(self):
1880
 
        t = self._transport('http://joe@www.example.com/foo')
1881
 
        r = t._redirected_to('http://www.example.com/foo',
1882
 
                             'https://foo.example.com/foo')
1883
 
        self.assertIsInstance(r, type(t))
1884
 
        self.assertEqual(t._user, r._user)
1885
 
 
1886
 
 
1887
 
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1888
 
    """Request handler for a unique and pre-defined request.
1889
 
 
1890
 
    The only thing we care about here is how many bytes travel on the wire. But
1891
 
    since we want to measure it for a real http client, we have to send it
1892
 
    correct responses.
1893
 
 
1894
 
    We expect to receive a *single* request nothing more (and we won't even
1895
 
    check what request it is, we just measure the bytes read until an empty
1896
 
    line.
1897
 
    """
1898
 
 
1899
 
    def handle_one_request(self):
1900
 
        tcs = self.server.test_case_server
1901
 
        requestline = self.rfile.readline()
1902
 
        headers = self.MessageClass(self.rfile, 0)
1903
 
        # We just read: the request, the headers, an empty line indicating the
1904
 
        # end of the headers.
1905
 
        bytes_read = len(requestline)
1906
 
        for line in headers.headers:
1907
 
            bytes_read += len(line)
1908
 
        bytes_read += len('\r\n')
1909
 
        if requestline.startswith('POST'):
1910
 
            # The body should be a single line (or we don't know where it ends
1911
 
            # and we don't want to issue a blocking read)
1912
 
            body = self.rfile.readline()
1913
 
            bytes_read += len(body)
1914
 
        tcs.bytes_read = bytes_read
1915
 
 
1916
 
        # We set the bytes written *before* issuing the write, the client is
1917
 
        # supposed to consume every produced byte *before* checking that value.
1918
 
 
1919
 
        # Doing the oppposite may lead to test failure: we may be interrupted
1920
 
        # after the write but before updating the value. The client can then
1921
 
        # continue and read the value *before* we can update it. And yes,
1922
 
        # this has been observed -- vila 20090129
1923
 
        tcs.bytes_written = len(tcs.canned_response)
1924
 
        self.wfile.write(tcs.canned_response)
1925
 
 
1926
 
 
1927
 
class ActivityServerMixin(object):
1928
 
 
1929
 
    def __init__(self, protocol_version):
1930
 
        super(ActivityServerMixin, self).__init__(
1931
 
            request_handler=PredefinedRequestHandler,
1932
 
            protocol_version=protocol_version)
1933
 
        # Bytes read and written by the server
1934
 
        self.bytes_read = 0
1935
 
        self.bytes_written = 0
1936
 
        self.canned_response = None
1937
 
 
1938
 
 
1939
 
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1940
 
    pass
1941
 
 
1942
 
 
1943
 
if tests.HTTPSServerFeature.available():
1944
 
    from bzrlib.tests import https_server
1945
 
    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1946
 
        pass
1947
 
 
1948
 
 
1949
 
class TestActivityMixin(object):
1950
 
    """Test socket activity reporting.
1951
 
 
1952
 
    We use a special purpose server to control the bytes sent and received and
1953
 
    be able to predict the activity on the client socket.
1954
 
    """
1955
 
 
1956
 
    def setUp(self):
1957
 
        tests.TestCase.setUp(self)
1958
 
        self.server = self._activity_server(self._protocol_version)
1959
 
        self.server.start_server()
1960
 
        self.activities = {}
1961
 
        def report_activity(t, bytes, direction):
1962
 
            count = self.activities.get(direction, 0)
1963
 
            count += bytes
1964
 
            self.activities[direction] = count
1965
 
 
1966
 
        # We override at class level because constructors may propagate the
1967
 
        # bound method and render instance overriding ineffective (an
1968
 
        # alternative would be to define a specific ui factory instead...)
1969
 
        self.orig_report_activity = self._transport._report_activity
1970
 
        self._transport._report_activity = report_activity
1971
 
 
1972
 
    def tearDown(self):
1973
 
        self._transport._report_activity = self.orig_report_activity
1974
 
        self.server.stop_server()
1975
 
        tests.TestCase.tearDown(self)
1976
 
 
1977
 
    def get_transport(self):
1978
 
        return self._transport(self.server.get_url())
1979
 
 
1980
 
    def assertActivitiesMatch(self):
1981
 
        self.assertEqual(self.server.bytes_read,
1982
 
                         self.activities.get('write', 0), 'written bytes')
1983
 
        self.assertEqual(self.server.bytes_written,
1984
 
                         self.activities.get('read', 0), 'read bytes')
1985
 
 
1986
 
    def test_get(self):
1987
 
        self.server.canned_response = '''HTTP/1.1 200 OK\r
1988
 
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
1989
 
Server: Apache/2.0.54 (Fedora)\r
1990
 
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
1991
 
ETag: "56691-23-38e9ae00"\r
1992
 
Accept-Ranges: bytes\r
1993
 
Content-Length: 35\r
1994
 
Connection: close\r
1995
 
Content-Type: text/plain; charset=UTF-8\r
1996
 
\r
1997
 
Bazaar-NG meta directory, format 1
1998
 
'''
1999
 
        t = self.get_transport()
2000
 
        self.assertEqual('Bazaar-NG meta directory, format 1\n',
2001
 
                         t.get('foo/bar').read())
2002
 
        self.assertActivitiesMatch()
2003
 
 
2004
 
    def test_has(self):
2005
 
        self.server.canned_response = '''HTTP/1.1 200 OK\r
2006
 
Server: SimpleHTTP/0.6 Python/2.5.2\r
2007
 
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2008
 
Content-type: application/octet-stream\r
2009
 
Content-Length: 20\r
2010
 
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2011
 
\r
2012
 
'''
2013
 
        t = self.get_transport()
2014
 
        self.assertTrue(t.has('foo/bar'))
2015
 
        self.assertActivitiesMatch()
2016
 
 
2017
 
    def test_readv(self):
2018
 
        self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2019
 
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2020
 
Server: Apache/2.0.54 (Fedora)\r
2021
 
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2022
 
ETag: "238a3c-16ec2-805c5540"\r
2023
 
Accept-Ranges: bytes\r
2024
 
Content-Length: 1534\r
2025
 
Connection: close\r
2026
 
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2027
 
\r
2028
 
\r
2029
 
--418470f848b63279b\r
2030
 
Content-type: text/plain; charset=UTF-8\r
2031
 
Content-range: bytes 0-254/93890\r
2032
 
\r
2033
 
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2034
 
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2035
 
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2036
 
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2037
 
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2038
 
\r
2039
 
--418470f848b63279b\r
2040
 
Content-type: text/plain; charset=UTF-8\r
2041
 
Content-range: bytes 1000-2049/93890\r
2042
 
\r
2043
 
40-fd4ec249b6b139ab
2044
 
mbp@sourcefrog.net-20050311063625-07858525021f270b
2045
 
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2046
 
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2047
 
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2048
 
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2049
 
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2050
 
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2051
 
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2052
 
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2053
 
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2054
 
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2055
 
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2056
 
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2057
 
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2058
 
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2059
 
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2060
 
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2061
 
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2062
 
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2063
 
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2064
 
mbp@source\r
2065
 
--418470f848b63279b--\r
2066
 
'''
2067
 
        t = self.get_transport()
2068
 
        # Remember that the request is ignored and that the ranges below
2069
 
        # doesn't have to match the canned response.
2070
 
        l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2071
 
        self.assertEqual(2, len(l))
2072
 
        self.assertActivitiesMatch()
2073
 
 
2074
 
    def test_post(self):
2075
 
        self.server.canned_response = '''HTTP/1.1 200 OK\r
2076
 
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2077
 
Server: Apache/2.0.54 (Fedora)\r
2078
 
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2079
 
ETag: "56691-23-38e9ae00"\r
2080
 
Accept-Ranges: bytes\r
2081
 
Content-Length: 35\r
2082
 
Connection: close\r
2083
 
Content-Type: text/plain; charset=UTF-8\r
2084
 
\r
2085
 
lalala whatever as long as itsssss
2086
 
'''
2087
 
        t = self.get_transport()
2088
 
        # We must send a single line of body bytes, see
2089
 
        # PredefinedRequestHandler.handle_one_request
2090
 
        code, f = t._post('abc def end-of-body\n')
2091
 
        self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2092
 
        self.assertActivitiesMatch()
2093
 
 
2094
 
 
2095
 
class TestActivity(tests.TestCase, TestActivityMixin):
2096
 
 
2097
 
    def setUp(self):
2098
 
        tests.TestCase.setUp(self)
2099
 
        self.server = self._activity_server(self._protocol_version)
2100
 
        self.server.start_server()
2101
 
        self.activities = {}
2102
 
        def report_activity(t, bytes, direction):
2103
 
            count = self.activities.get(direction, 0)
2104
 
            count += bytes
2105
 
            self.activities[direction] = count
2106
 
 
2107
 
        # We override at class level because constructors may propagate the
2108
 
        # bound method and render instance overriding ineffective (an
2109
 
        # alternative would be to define a specific ui factory instead...)
2110
 
        self.orig_report_activity = self._transport._report_activity
2111
 
        self._transport._report_activity = report_activity
2112
 
 
2113
 
    def tearDown(self):
2114
 
        self._transport._report_activity = self.orig_report_activity
2115
 
        self.server.stop_server()
2116
 
        tests.TestCase.tearDown(self)
2117
 
 
2118
 
 
2119
 
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2120
 
 
2121
 
    def setUp(self):
2122
 
        tests.TestCase.setUp(self)
2123
 
        # Unlike TestActivity, we are really testing ReportingFileSocket and
2124
 
        # ReportingSocket, so we don't need all the parametrization. Since
2125
 
        # ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2126
 
        # test them through their use by the transport than directly (that's a
2127
 
        # bit less clean but far more simpler and effective).
2128
 
        self.server = ActivityHTTPServer('HTTP/1.1')
2129
 
        self._transport=_urllib.HttpTransport_urllib
2130
 
 
2131
 
        self.server.start_server()
2132
 
 
2133
 
        # We override at class level because constructors may propagate the
2134
 
        # bound method and render instance overriding ineffective (an
2135
 
        # alternative would be to define a specific ui factory instead...)
2136
 
        self.orig_report_activity = self._transport._report_activity
2137
 
        self._transport._report_activity = None
2138
 
 
2139
 
    def tearDown(self):
2140
 
        self._transport._report_activity = self.orig_report_activity
2141
 
        self.server.stop_server()
2142
 
        tests.TestCase.tearDown(self)
2143
 
 
2144
 
    def assertActivitiesMatch(self):
2145
 
        # Nothing to check here
2146
 
        pass
2147
 
 
2148
 
 
2149
 
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2150
 
    """Test authentication on the redirected http server."""
2151
 
 
2152
 
    _auth_header = 'Authorization'
2153
 
    _password_prompt_prefix = ''
2154
 
    _username_prompt_prefix = ''
2155
 
    _auth_server = http_utils.HTTPBasicAuthServer
2156
 
    _transport = _urllib.HttpTransport_urllib
2157
 
 
2158
 
    def create_transport_readonly_server(self):
2159
 
        return self._auth_server()
2160
 
 
2161
 
    def create_transport_secondary_server(self):
2162
 
        """Create the secondary server redirecting to the primary server"""
2163
 
        new = self.get_readonly_server()
2164
 
 
2165
 
        redirecting = http_utils.HTTPServerRedirecting()
2166
 
        redirecting.redirect_to(new.host, new.port)
2167
 
        return redirecting
2168
 
 
2169
 
    def setUp(self):
2170
 
        super(TestAuthOnRedirected, self).setUp()
2171
 
        self.build_tree_contents([('a','a'),
2172
 
                                  ('1/',),
2173
 
                                  ('1/a', 'redirected once'),
2174
 
                                  ],)
2175
 
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2176
 
                                       self.new_server.port)
2177
 
        self.old_server.redirections = [
2178
 
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
2179
 
        self.old_transport = self._transport(self.old_server.get_url())
2180
 
        self.new_server.add_user('joe', 'foo')
2181
 
 
2182
 
    def get_a(self, transport):
2183
 
        return transport.get('a')
2184
 
 
2185
 
    def test_auth_on_redirected_via_do_catching_redirections(self):
2186
 
        self.redirections = 0
2187
 
 
2188
 
        def redirected(transport, exception, redirection_notice):
2189
 
            self.redirections += 1
2190
 
            dir, file = urlutils.split(exception.target)
2191
 
            return self._transport(dir)
2192
 
 
2193
 
        stdout = tests.StringIOWrapper()
2194
 
        stderr = tests.StringIOWrapper()
2195
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2196
 
                                            stdout=stdout, stderr=stderr)
2197
 
        self.assertEqual('redirected once',
2198
 
                         transport.do_catching_redirections(
2199
 
                self.get_a, self.old_transport, redirected).read())
2200
 
        self.assertEqual(1, self.redirections)
2201
 
        # stdin should be empty
2202
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
2203
 
        # stdout should be empty, stderr will contains the prompts
2204
 
        self.assertEqual('', stdout.getvalue())
2205
 
 
2206
 
    def test_auth_on_redirected_via_following_redirections(self):
2207
 
        self.new_server.add_user('joe', 'foo')
2208
 
        stdout = tests.StringIOWrapper()
2209
 
        stderr = tests.StringIOWrapper()
2210
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2211
 
                                            stdout=stdout, stderr=stderr)
2212
 
        t = self.old_transport
2213
 
        req = RedirectedRequest('GET', t.abspath('a'))
2214
 
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2215
 
                                       self.new_server.port)
2216
 
        self.old_server.redirections = [
2217
 
            ('(.*)', r'%s/1\1' % (new_prefix), 301),]
2218
 
        self.assertEqual('redirected once',t._perform(req).read())
2219
 
        # stdin should be empty
2220
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
2221
 
        # stdout should be empty, stderr will contains the prompts
2222
 
        self.assertEqual('', stdout.getvalue())
2223
 
 
 
1741
                          t.send_http_smart_request, 'whatever')
2224
1742