/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: Martin von Gagern
  • Date: 2010-04-20 08:47:38 UTC
  • mfrom: (5167 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5195.
  • Revision ID: martin.vgagern@gmx.net-20100420084738-ygymnqmdllzrhpfn
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
1
 
# Copyright (C) 2005, 2006 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Tests for HTTP implementations.
18
18
 
44
44
    ui,
45
45
    urlutils,
46
46
    )
 
47
from bzrlib.symbol_versioning import (
 
48
    deprecated_in,
 
49
    )
47
50
from bzrlib.tests import (
 
51
    features,
48
52
    http_server,
49
53
    http_utils,
50
54
    )
58
62
    )
59
63
 
60
64
 
61
 
try:
 
65
if features.pycurl.available():
62
66
    from bzrlib.transport.http._pycurl import PyCurlTransport
63
 
    pycurl_present = True
64
 
except errors.DependencyNotPresent:
65
 
    pycurl_present = False
66
 
 
67
 
 
68
 
class TransportAdapter(tests.TestScenarioApplier):
69
 
    """Generate the same test for each transport implementation."""
70
 
 
71
 
    def __init__(self):
72
 
        transport_scenarios = [
73
 
            ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
74
 
                            _server=http_server.HttpServer_urllib,
75
 
                            _qualified_prefix='http+urllib',)),
76
 
            ]
77
 
        if pycurl_present:
78
 
            transport_scenarios.append(
79
 
                ('pycurl', dict(_transport=PyCurlTransport,
80
 
                                _server=http_server.HttpServer_PyCurl,
81
 
                                _qualified_prefix='http+pycurl',)))
82
 
        self.scenarios = transport_scenarios
83
 
 
84
 
 
85
 
class TransportProtocolAdapter(TransportAdapter):
86
 
    """Generate the same test for each protocol implementation.
87
 
 
88
 
    In addition to the transport adaptatation that we inherit from.
89
 
    """
90
 
 
91
 
    def __init__(self):
92
 
        super(TransportProtocolAdapter, self).__init__()
93
 
        protocol_scenarios = [
94
 
            ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
95
 
            ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
96
 
            ]
97
 
        self.scenarios = tests.multiply_scenarios(self.scenarios,
98
 
                                                  protocol_scenarios)
99
 
 
100
 
 
101
 
class TransportProtocolAuthenticationAdapter(TransportProtocolAdapter):
102
 
    """Generate the same test for each authentication scheme implementation.
103
 
 
104
 
    In addition to the protocol adaptatation that we inherit from.
105
 
    """
106
 
 
107
 
    def __init__(self):
108
 
        super(TransportProtocolAuthenticationAdapter, self).__init__()
109
 
        auth_scheme_scenarios = [
110
 
            ('basic', dict(_auth_scheme='basic')),
111
 
            ('digest', dict(_auth_scheme='digest')),
112
 
            ]
113
 
 
114
 
        self.scenarios = tests.multiply_scenarios(self.scenarios,
115
 
                                                  auth_scheme_scenarios)
 
67
 
116
68
 
117
69
def load_tests(standard_tests, module, loader):
118
70
    """Multiply tests for http clients and protocol versions."""
119
 
    # one for each transport
120
 
    t_adapter = TransportAdapter()
121
 
    t_classes= (TestHttpTransportRegistration,
 
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,
122
77
                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,
123
164
                )
124
 
    is_testing_for_transports = tests.condition_isinstance(t_classes)
125
 
 
126
 
    # multiplied by one for each protocol version
127
 
    tp_adapter = TransportProtocolAdapter()
128
 
    tp_classes= (SmartHTTPTunnellingTest,
129
 
                 TestDoCatchRedirections,
130
 
                 TestHTTPConnections,
131
 
                 TestHTTPRedirections,
132
 
                 TestHTTPSilentRedirections,
133
 
                 TestLimitedRangeRequestServer,
134
 
                 TestPost,
135
 
                 TestProxyHttpServer,
136
 
                 TestRanges,
137
 
                 TestSpecificRequestHandler,
138
 
                 )
139
 
    is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)
140
 
 
141
 
    # multiplied by one for each authentication scheme
142
 
    tpa_adapter = TransportProtocolAuthenticationAdapter()
143
 
    tpa_classes = (TestAuth,
144
 
                   )
145
 
    is_also_testing_for_authentication = tests.condition_isinstance(
146
 
        tpa_classes)
147
 
 
148
 
    result = loader.suiteClass()
149
 
    for test_class in tests.iter_suite_tests(standard_tests):
150
 
        # Each test class is either standalone or testing for some combination
151
 
        # of transport, protocol version, authentication scheme. Use the right
152
 
        # adpater (or none) depending on the class.
153
 
        if is_testing_for_transports(test_class):
154
 
            result.addTests(t_adapter.adapt(test_class))
155
 
        elif is_also_testing_for_protocols(test_class):
156
 
            result.addTests(tp_adapter.adapt(test_class))
157
 
        elif is_also_testing_for_authentication(test_class):
158
 
            result.addTests(tpa_adapter.adapt(test_class))
159
 
        else:
160
 
            result.addTest(test_class)
 
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
 
161
187
    return result
162
188
 
163
189
 
172
198
 
173
199
class RecordingServer(object):
174
200
    """A fake HTTP server.
175
 
    
 
201
 
176
202
    It records the bytes sent to it, and replies with a 200.
177
203
    """
178
204
 
179
 
    def __init__(self, expect_body_tail=None):
 
205
    def __init__(self, expect_body_tail=None, scheme=''):
180
206
        """Constructor.
181
207
 
182
208
        :type expect_body_tail: str
187
213
        self.host = None
188
214
        self.port = None
189
215
        self.received_bytes = ''
190
 
 
191
 
    def setUp(self):
 
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):
192
222
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
193
223
        self._sock.bind(('127.0.0.1', 0))
194
224
        self.host, self.port = self._sock.getsockname()
217
247
            # The client may have already closed the socket.
218
248
            pass
219
249
 
220
 
    def tearDown(self):
 
250
    def stop_server(self):
221
251
        try:
222
252
            self._sock.close()
223
253
        except socket.error:
227
257
        self.port = None
228
258
 
229
259
 
 
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
 
230
299
class TestHTTPServer(tests.TestCase):
231
300
    """Test the HTTP servers implementations."""
232
301
 
237
306
 
238
307
        server = http_server.HttpServer(BogusRequestHandler)
239
308
        try:
240
 
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
 
309
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
241
310
        except:
242
 
            server.tearDown()
 
311
            server.stop_server()
243
312
            self.fail('HTTP Server creation did not raise UnknownProtocol')
244
313
 
245
314
    def test_force_invalid_protocol(self):
246
315
        server = http_server.HttpServer(protocol_version='HTTP/0.1')
247
316
        try:
248
 
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
 
317
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
249
318
        except:
250
 
            server.tearDown()
 
319
            server.stop_server()
251
320
            self.fail('HTTP Server creation did not raise UnknownProtocol')
252
321
 
253
322
    def test_server_start_and_stop(self):
254
323
        server = http_server.HttpServer()
255
 
        server.setUp()
256
 
        self.assertTrue(server._http_running)
257
 
        server.tearDown()
 
324
        server.start_server()
 
325
        try:
 
326
            self.assertTrue(server._http_running)
 
327
        finally:
 
328
            server.stop_server()
258
329
        self.assertFalse(server._http_running)
259
330
 
260
331
    def test_create_http_server_one_zero(self):
263
334
            protocol_version = 'HTTP/1.0'
264
335
 
265
336
        server = http_server.HttpServer(RequestHandlerOneZero)
266
 
        server.setUp()
267
 
        self.addCleanup(server.tearDown)
 
337
        self.start_server(server)
268
338
        self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
269
339
 
270
340
    def test_create_http_server_one_one(self):
273
343
            protocol_version = 'HTTP/1.1'
274
344
 
275
345
        server = http_server.HttpServer(RequestHandlerOneOne)
276
 
        server.setUp()
277
 
        self.addCleanup(server.tearDown)
 
346
        self.start_server(server)
278
347
        self.assertIsInstance(server._httpd,
279
348
                              http_server.TestingThreadingHTTPServer)
280
349
 
285
354
 
286
355
        server = http_server.HttpServer(RequestHandlerOneZero,
287
356
                                        protocol_version='HTTP/1.1')
288
 
        server.setUp()
289
 
        self.addCleanup(server.tearDown)
 
357
        self.start_server(server)
290
358
        self.assertIsInstance(server._httpd,
291
359
                              http_server.TestingThreadingHTTPServer)
292
360
 
297
365
 
298
366
        server = http_server.HttpServer(RequestHandlerOneOne,
299
367
                                        protocol_version='HTTP/1.0')
300
 
        server.setUp()
301
 
        self.addCleanup(server.tearDown)
 
368
        self.start_server(server)
302
369
        self.assertIsInstance(server._httpd,
303
370
                              http_server.TestingHTTPServer)
304
371
 
307
374
    """Test case to inherit from if pycurl is present"""
308
375
 
309
376
    def _get_pycurl_maybe(self):
310
 
        try:
311
 
            from bzrlib.transport.http._pycurl import PyCurlTransport
312
 
            return PyCurlTransport
313
 
        except errors.DependencyNotPresent:
314
 
            raise tests.TestSkipped('pycurl not present')
 
377
        self.requireFeature(features.pycurl)
 
378
        return PyCurlTransport
315
379
 
316
380
    _transport = property(_get_pycurl_maybe)
317
381
 
324
388
    def test_url_parsing(self):
325
389
        f = FakeManager()
326
390
        url = http.extract_auth('http://example.com', f)
327
 
        self.assertEquals('http://example.com', url)
328
 
        self.assertEquals(0, len(f.credentials))
 
391
        self.assertEqual('http://example.com', url)
 
392
        self.assertEqual(0, len(f.credentials))
329
393
        url = http.extract_auth(
330
 
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
331
 
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
332
 
        self.assertEquals(1, len(f.credentials))
333
 
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
334
 
                          f.credentials[0])
 
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])
335
399
 
336
400
 
337
401
class TestHttpTransportUrls(tests.TestCase):
364
428
    def test_http_impl_urls(self):
365
429
        """There are servers which ask for particular clients to connect"""
366
430
        server = self._server()
 
431
        server.start_server()
367
432
        try:
368
 
            server.setUp()
369
433
            url = server.get_url()
370
434
            self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
371
435
        finally:
372
 
            server.tearDown()
 
436
            server.stop_server()
373
437
 
374
438
 
375
439
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
384
448
        https by supplying a fake version_info that do not
385
449
        support it.
386
450
        """
387
 
        try:
388
 
            import pycurl
389
 
        except ImportError:
390
 
            raise tests.TestSkipped('pycurl not present')
 
451
        self.requireFeature(features.pycurl)
 
452
        # Import the module locally now that we now it's available.
 
453
        pycurl = features.pycurl.module
391
454
 
392
 
        version_info_orig = pycurl.version_info
393
 
        try:
394
 
            # Now that we have pycurl imported, we can fake its version_info
395
 
            # This was taken from a windows pycurl without SSL
396
 
            # (thanks to bialix)
397
 
            pycurl.version_info = lambda : (2,
398
 
                                            '7.13.2',
399
 
                                            462082,
400
 
                                            'i386-pc-win32',
401
 
                                            2576,
402
 
                                            None,
403
 
                                            0,
404
 
                                            None,
405
 
                                            ('ftp', 'gopher', 'telnet',
406
 
                                             'dict', 'ldap', 'http', 'file'),
407
 
                                            None,
408
 
                                            0,
409
 
                                            None)
410
 
            self.assertRaises(errors.DependencyNotPresent, self._transport,
411
 
                              'https://launchpad.net')
412
 
        finally:
413
 
            # Restore the right function
414
 
            pycurl.version_info = version_info_orig
 
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')
415
473
 
416
474
 
417
475
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
476
534
class TestPost(tests.TestCase):
477
535
 
478
536
    def test_post_body_is_received(self):
479
 
        server = RecordingServer(expect_body_tail='end-of-body')
480
 
        server.setUp()
481
 
        self.addCleanup(server.tearDown)
482
 
        scheme = self._qualified_prefix
483
 
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
 
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
541
        http_transport = self._transport(url)
485
542
        code, response = http_transport._post('abc def end-of-body')
486
543
        self.assertTrue(
535
592
                                      protocol_version=self._protocol_version)
536
593
 
537
594
    def _testing_pycurl(self):
538
 
        return pycurl_present and self._transport == PyCurlTransport
 
595
        # TODO: This is duplicated for lots of the classes in this file
 
596
        return (features.pycurl.available()
 
597
                and self._transport == PyCurlTransport)
539
598
 
540
599
 
541
600
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
558
617
        # for details) make no distinction between a closed
559
618
        # socket and badly formatted status line, so we can't
560
619
        # just test for ConnectionError, we have to test
561
 
        # InvalidHttpResponse too.
562
 
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
 
620
        # InvalidHttpResponse too. And pycurl may raise ConnectionReset
 
621
        # instead of ConnectionError too.
 
622
        self.assertRaises(( errors.ConnectionError, errors.ConnectionReset,
 
623
                            errors.InvalidHttpResponse),
563
624
                          t.has, 'foo/bar')
564
625
 
565
626
    def test_http_get(self):
566
627
        server = self.get_readonly_server()
567
628
        t = self._transport(server.get_url())
568
 
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
 
629
        self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
 
630
                           errors.InvalidHttpResponse),
569
631
                          t.get, 'foo/bar')
570
632
 
571
633
 
647
709
    _req_handler_class = BadProtocolRequestHandler
648
710
 
649
711
    def setUp(self):
650
 
        if pycurl_present and self._transport == PyCurlTransport:
 
712
        if self._testing_pycurl():
651
713
            raise tests.TestNotApplicable(
652
714
                "pycurl doesn't check the protocol version")
653
715
        super(TestBadProtocolServer, self).setUp()
697
759
        self.assertEqual(None, server.host)
698
760
        self.assertEqual(None, server.port)
699
761
 
700
 
    def test_setUp_and_tearDown(self):
 
762
    def test_setUp_and_stop(self):
701
763
        server = RecordingServer(expect_body_tail=None)
702
 
        server.setUp()
 
764
        server.start_server()
703
765
        try:
704
766
            self.assertNotEqual(None, server.host)
705
767
            self.assertNotEqual(None, server.port)
706
768
        finally:
707
 
            server.tearDown()
 
769
            server.stop_server()
708
770
        self.assertEqual(None, server.host)
709
771
        self.assertEqual(None, server.port)
710
772
 
711
773
    def test_send_receive_bytes(self):
712
 
        server = RecordingServer(expect_body_tail='c')
713
 
        server.setUp()
714
 
        self.addCleanup(server.tearDown)
 
774
        server = RecordingServer(expect_body_tail='c', scheme='http')
 
775
        self.start_server(server)
715
776
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
716
777
        sock.connect((server.host, server.port))
717
778
        sock.sendall('abc')
811
872
        # bytes on the socket
812
873
        ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
813
874
        self.assertEqual((0, '0'), ireadv.next())
814
 
        # The server should have issued one request so far 
 
875
        # The server should have issued one request so far
815
876
        self.assertEqual(1, server.GET_request_nb)
816
877
        self.assertEqual('0123456789', t.get_bytes('a'))
817
878
        # get_bytes issued an additional request, the readv pending ones are
1108
1169
        if self._testing_pycurl():
1109
1170
            # Oh my ! pycurl does not check for the port as part of
1110
1171
            # no_proxy :-( So we just test the host part
1111
 
            self.no_proxy_host = 'localhost'
 
1172
            self.no_proxy_host = self.server.host
1112
1173
        else:
1113
1174
            self.no_proxy_host = self.proxy_address
1114
1175
        # The secondary server is the proxy
1117
1178
        self._old_env = {}
1118
1179
 
1119
1180
    def _testing_pycurl(self):
1120
 
        return pycurl_present and self._transport == PyCurlTransport
 
1181
        # TODO: This is duplicated for lots of the classes in this file
 
1182
        return (features.pycurl.available()
 
1183
                and self._transport == PyCurlTransport)
1121
1184
 
1122
1185
    def create_transport_secondary_server(self):
1123
1186
        """Creates an http server that will serve files with
1263
1326
                                  ('bundle',
1264
1327
                                  '# Bazaar revision bundle v0.9\n#\n')
1265
1328
                                  ],)
1266
 
 
 
1329
        # The requests to the old server will be redirected to the new server
1267
1330
        self.old_transport = self._transport(self.old_server.get_url())
1268
1331
 
1269
1332
    def test_redirected(self):
1274
1337
    def test_read_redirected_bundle_from_url(self):
1275
1338
        from bzrlib.bundle import read_bundle_from_url
1276
1339
        url = self.old_transport.abspath('bundle')
1277
 
        bundle = read_bundle_from_url(url)
 
1340
        bundle = self.applyDeprecated(deprecated_in((1, 12, 0)),
 
1341
                read_bundle_from_url, url)
1278
1342
        # If read_bundle_from_url was successful we get an empty bundle
1279
1343
        self.assertEqual([], bundle.revisions)
1280
1344
 
1291
1355
        # Since the tests using this class will replace
1292
1356
        # _urllib2_wrappers.Request, we can't just call the base class __init__
1293
1357
        # or we'll loop.
1294
 
        RedirectedRequest.init_orig(self, method, url, args, kwargs)
 
1358
        RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1295
1359
        self.follow_redirections = True
1296
1360
 
1297
1361
 
 
1362
def install_redirected_request(test):
 
1363
    test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
 
1364
 
 
1365
 
1298
1366
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1299
1367
    """Test redirections.
1300
1368
 
1310
1378
    """
1311
1379
 
1312
1380
    def setUp(self):
1313
 
        if pycurl_present and self._transport == PyCurlTransport:
 
1381
        if (features.pycurl.available()
 
1382
            and self._transport == PyCurlTransport):
1314
1383
            raise tests.TestNotApplicable(
1315
1384
                "pycurl doesn't redirect silently annymore")
1316
1385
        super(TestHTTPSilentRedirections, self).setUp()
1317
 
        self.setup_redirected_request()
1318
 
        self.addCleanup(self.cleanup_redirected_request)
 
1386
        install_redirected_request(self)
1319
1387
        self.build_tree_contents([('a','a'),
1320
1388
                                  ('1/',),
1321
1389
                                  ('1/a', 'redirected once'),
1331
1399
 
1332
1400
        self.old_transport = self._transport(self.old_server.get_url())
1333
1401
 
1334
 
    def setup_redirected_request(self):
1335
 
        self.original_class = _urllib2_wrappers.Request
1336
 
        _urllib2_wrappers.Request = RedirectedRequest
1337
 
 
1338
 
    def cleanup_redirected_request(self):
1339
 
        _urllib2_wrappers.Request = self.original_class
1340
 
 
1341
1402
    def create_transport_secondary_server(self):
1342
1403
        """Create the secondary server, redirections are defined in the tests"""
1343
1404
        return http_utils.HTTPServerRedirecting(
1347
1408
        t = self.old_transport
1348
1409
 
1349
1410
        req = RedirectedRequest('GET', t.abspath('a'))
1350
 
        req.follow_redirections = True
1351
1411
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1352
1412
                                       self.new_server.port)
1353
1413
        self.old_server.redirections = \
1354
1414
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1355
 
        self.assertEquals('redirected once',t._perform(req).read())
 
1415
        self.assertEqual('redirected once',t._perform(req).read())
1356
1416
 
1357
1417
    def test_five_redirections(self):
1358
1418
        t = self.old_transport
1359
1419
 
1360
1420
        req = RedirectedRequest('GET', t.abspath('a'))
1361
 
        req.follow_redirections = True
1362
1421
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1363
1422
                                       self.old_server.port)
1364
1423
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1370
1429
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1371
1430
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1372
1431
            ]
1373
 
        self.assertEquals('redirected 5 times',t._perform(req).read())
 
1432
        self.assertEqual('redirected 5 times',t._perform(req).read())
1374
1433
 
1375
1434
 
1376
1435
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1389
1448
        t = self._transport(self.new_server.get_url())
1390
1449
 
1391
1450
        # We use None for redirected so that we fail if redirected
1392
 
        self.assertEquals('0123456789',
1393
 
                          transport.do_catching_redirections(
 
1451
        self.assertEqual('0123456789',
 
1452
                         transport.do_catching_redirections(
1394
1453
                self.get_a, t, None).read())
1395
1454
 
1396
1455
    def test_one_redirection(self):
1401
1460
            dir, file = urlutils.split(exception.target)
1402
1461
            return self._transport(dir)
1403
1462
 
1404
 
        self.assertEquals('0123456789',
1405
 
                          transport.do_catching_redirections(
 
1463
        self.assertEqual('0123456789',
 
1464
                         transport.do_catching_redirections(
1406
1465
                self.get_a, self.old_transport, redirected).read())
1407
 
        self.assertEquals(1, self.redirections)
 
1466
        self.assertEqual(1, self.redirections)
1408
1467
 
1409
1468
    def test_redirection_loop(self):
1410
1469
 
1424
1483
 
1425
1484
    _auth_header = 'Authorization'
1426
1485
    _password_prompt_prefix = ''
 
1486
    _username_prompt_prefix = ''
 
1487
    # Set by load_tests
 
1488
    _auth_server = None
1427
1489
 
1428
1490
    def setUp(self):
1429
1491
        super(TestAuth, self).setUp()
1432
1494
                                  ('b', 'contents of b\n'),])
1433
1495
 
1434
1496
    def create_transport_readonly_server(self):
1435
 
        if self._auth_scheme == 'basic':
1436
 
            server = http_utils.HTTPBasicAuthServer(
1437
 
                protocol_version=self._protocol_version)
1438
 
        else:
1439
 
            if self._auth_scheme != 'digest':
1440
 
                raise AssertionError('Unknown auth scheme: %r'
1441
 
                                     % self._auth_scheme)
1442
 
            server = http_utils.HTTPDigestAuthServer(
1443
 
                protocol_version=self._protocol_version)
1444
 
        return server
 
1497
        return self._auth_server(protocol_version=self._protocol_version)
1445
1498
 
1446
1499
    def _testing_pycurl(self):
1447
 
        return pycurl_present and self._transport == PyCurlTransport
 
1500
        # TODO: This is duplicated for lots of the classes in this file
 
1501
        return (features.pycurl.available()
 
1502
                and self._transport == PyCurlTransport)
1448
1503
 
1449
 
    def get_user_url(self, user=None, password=None):
 
1504
    def get_user_url(self, user, password):
1450
1505
        """Build an url embedding user and password"""
1451
1506
        url = '%s://' % self.server._url_protocol
1452
1507
        if user is not None:
1457
1512
        url += '%s:%s/' % (self.server.host, self.server.port)
1458
1513
        return url
1459
1514
 
1460
 
    def get_user_transport(self, user=None, password=None):
 
1515
    def get_user_transport(self, user, password):
1461
1516
        return self._transport(self.get_user_url(user, password))
1462
1517
 
1463
1518
    def test_no_user(self):
1464
1519
        self.server.add_user('joe', 'foo')
1465
 
        t = self.get_user_transport()
 
1520
        t = self.get_user_transport(None, None)
1466
1521
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1467
1522
        # Only one 'Authentication Required' error should occur
1468
1523
        self.assertEqual(1, self.server.auth_required_errors)
1498
1553
        # initial 'who are you' and 'this is not you, who are you')
1499
1554
        self.assertEqual(2, self.server.auth_required_errors)
1500
1555
 
 
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
 
1501
1578
    def test_prompt_for_password(self):
1502
1579
        if self._testing_pycurl():
1503
1580
            raise tests.TestNotApplicable(
1507
1584
        self.server.add_user('joe', 'foo')
1508
1585
        t = self.get_user_transport('joe', None)
1509
1586
        stdout = tests.StringIOWrapper()
1510
 
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1511
 
        self.assertEqual('contents of a\n',t.get('a').read())
 
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())
1512
1591
        # stdin should be empty
1513
1592
        self.assertEqual('', ui.ui_factory.stdin.readline())
1514
1593
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1515
 
                                    stdout.getvalue())
 
1594
                                    stderr.getvalue())
 
1595
        self.assertEqual('', stdout.getvalue())
1516
1596
        # And we shouldn't prompt again for a different request
1517
1597
        # against the same transport.
1518
1598
        self.assertEqual('contents of b\n',t.get('b').read())
1528
1608
                              % (scheme.upper(),
1529
1609
                                 user, self.server.host, self.server.port,
1530
1610
                                 self.server.auth_realm)))
1531
 
        self.assertEquals(expected_prompt, actual_prompt)
 
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))
1532
1618
 
1533
1619
    def test_no_prompt_for_password_when_using_auth_config(self):
1534
1620
        if self._testing_pycurl():
1542
1628
        self.server.add_user(user, password)
1543
1629
        t = self.get_user_transport(user, None)
1544
1630
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1545
 
                                            stdout=tests.StringIOWrapper())
 
1631
                                            stderr=tests.StringIOWrapper())
1546
1632
        # Create a minimal config file with the right password
1547
1633
        conf = config.AuthenticationConfig()
1548
1634
        conf._get_config().update(
1556
1642
        # Only one 'Authentication Required' error should occur
1557
1643
        self.assertEqual(1, self.server.auth_required_errors)
1558
1644
 
 
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
 
1559
1664
    def test_changing_nonce(self):
1560
 
        if self._auth_scheme != 'digest':
1561
 
            raise tests.TestNotApplicable('HTTP auth digest only test')
 
1665
        if self._auth_server not in (http_utils.HTTPDigestAuthServer,
 
1666
                                     http_utils.ProxyDigestAuthServer):
 
1667
            raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1562
1668
        if self._testing_pycurl():
1563
1669
            raise tests.KnownFailure(
1564
1670
                'pycurl does not handle a nonce change')
1582
1688
    """Test proxy authentication schemes."""
1583
1689
 
1584
1690
    _auth_header = 'Proxy-authorization'
1585
 
    _password_prompt_prefix='Proxy '
 
1691
    _password_prompt_prefix = 'Proxy '
 
1692
    _username_prompt_prefix = 'Proxy '
1586
1693
 
1587
1694
    def setUp(self):
1588
1695
        super(TestProxyAuth, self).setUp()
1595
1702
                                  ('b-proxied', 'contents of b\n'),
1596
1703
                                  ])
1597
1704
 
1598
 
    def create_transport_readonly_server(self):
1599
 
        if self._auth_scheme == 'basic':
1600
 
            server = http_utils.ProxyBasicAuthServer(
1601
 
                protocol_version=self._protocol_version)
1602
 
        else:
1603
 
            if self._auth_scheme != 'digest':
1604
 
                raise AssertionError('Unknown auth scheme: %r'
1605
 
                                     % self._auth_scheme)
1606
 
            server = http_utils.ProxyDigestAuthServer(
1607
 
                protocol_version=self._protocol_version)
1608
 
        return server
1609
 
 
1610
 
    def get_user_transport(self, user=None, password=None):
 
1705
    def get_user_transport(self, user, password):
1611
1706
        self._install_env({'all_proxy': self.get_user_url(user, password)})
1612
1707
        return self._transport(self.server.get_url())
1613
1708
 
1743
1838
                          t.get_smart_medium().send_http_smart_request,
1744
1839
                          'whatever')
1745
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
 
 
2224