← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Martin Pool
  • Date: 2009-01-13 03:11:04 UTC
  • mto: This revision was merged to the branch mainline in revision 3937.
  • Revision ID: mbp@sourcefrog.net-20090113031104-03my054s02i9l2pe
Bump version to 1.12 and add news template

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, 2007, 2008 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
 
44
44
    ui,
45
45
    urlutils,
46
46
    )
47
 
from bzrlib.symbol_versioning import (
48
 
    deprecated_in,
49
 
    )
50
47
from bzrlib.tests import (
51
 
    features,
52
48
    http_server,
53
49
    http_utils,
54
50
    )
62
58
    )
63
59
 
64
60
 
65
 
if features.pycurl.available():
 
61
try:
66
62
    from bzrlib.transport.http._pycurl import PyCurlTransport
67
 
 
 
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)
68
116
 
69
117
def load_tests(standard_tests, module, loader):
70
118
    """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,
 
119
    # one for each transport
 
120
    t_adapter = TransportAdapter()
 
121
    t_classes= (TestHttpTransportRegistration,
77
122
                TestHttpTransportUrls,
78
123
                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
124
                )
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
 
 
 
125
    is_testing_for_transports = tests.condition_isinstance(t_classes)
 
126
 
 
127
    # multiplied by one for each protocol version
 
128
    tp_adapter = TransportProtocolAdapter()
 
129
    tp_classes= (SmartHTTPTunnellingTest,
 
130
                 TestDoCatchRedirections,
 
131
                 TestHTTPConnections,
 
132
                 TestHTTPRedirections,
 
133
                 TestHTTPSilentRedirections,
 
134
                 TestLimitedRangeRequestServer,
 
135
                 TestPost,
 
136
                 TestProxyHttpServer,
 
137
                 TestRanges,
 
138
                 TestSpecificRequestHandler,
 
139
                 )
 
140
    is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)
 
141
 
 
142
    # multiplied by one for each authentication scheme
 
143
    tpa_adapter = TransportProtocolAuthenticationAdapter()
 
144
    tpa_classes = (TestAuth,
 
145
                   )
 
146
    is_also_testing_for_authentication = tests.condition_isinstance(
 
147
        tpa_classes)
 
148
 
 
149
    result = loader.suiteClass()
 
150
    for test_class in tests.iter_suite_tests(standard_tests):
 
151
        # Each test class is either standalone or testing for some combination
 
152
        # of transport, protocol version, authentication scheme. Use the right
 
153
        # adpater (or none) depending on the class.
 
154
        if is_testing_for_transports(test_class):
 
155
            result.addTests(t_adapter.adapt(test_class))
 
156
        elif is_also_testing_for_protocols(test_class):
 
157
            result.addTests(tp_adapter.adapt(test_class))
 
158
        elif is_also_testing_for_authentication(test_class):
 
159
            result.addTests(tpa_adapter.adapt(test_class))
 
160
        else:
 
161
            result.addTest(test_class)
187
162
    return result
188
163
 
189
164
 
198
173
 
199
174
class RecordingServer(object):
200
175
    """A fake HTTP server.
201
 
 
 
176
    
202
177
    It records the bytes sent to it, and replies with a 200.
203
178
    """
204
179
 
205
 
    def __init__(self, expect_body_tail=None, scheme=''):
 
180
    def __init__(self, expect_body_tail=None):
206
181
        """Constructor.
207
182
 
208
183
        :type expect_body_tail: str
213
188
        self.host = None
214
189
        self.port = None
215
190
        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):
 
191
 
 
192
    def setUp(self):
222
193
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
223
194
        self._sock.bind(('127.0.0.1', 0))
224
195
        self.host, self.port = self._sock.getsockname()
247
218
            # The client may have already closed the socket.
248
219
            pass
249
220
 
250
 
    def stop_server(self):
 
221
    def tearDown(self):
251
222
        try:
252
223
            self._sock.close()
253
224
        except socket.error:
257
228
        self.port = None
258
229
 
259
230
 
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
231
class TestHTTPServer(tests.TestCase):
300
232
    """Test the HTTP servers implementations."""
301
233
 
306
238
 
307
239
        server = http_server.HttpServer(BogusRequestHandler)
308
240
        try:
309
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
241
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
310
242
        except:
311
 
            server.stop_server()
 
243
            server.tearDown()
312
244
            self.fail('HTTP Server creation did not raise UnknownProtocol')
313
245
 
314
246
    def test_force_invalid_protocol(self):
315
247
        server = http_server.HttpServer(protocol_version='HTTP/0.1')
316
248
        try:
317
 
            self.assertRaises(httplib.UnknownProtocol, server.start_server)
 
249
            self.assertRaises(httplib.UnknownProtocol,server.setUp)
318
250
        except:
319
 
            server.stop_server()
 
251
            server.tearDown()
320
252
            self.fail('HTTP Server creation did not raise UnknownProtocol')
321
253
 
322
254
    def test_server_start_and_stop(self):
323
255
        server = http_server.HttpServer()
324
 
        server.start_server()
325
 
        try:
326
 
            self.assertTrue(server._http_running)
327
 
        finally:
328
 
            server.stop_server()
 
256
        server.setUp()
 
257
        self.assertTrue(server._http_running)
 
258
        server.tearDown()
329
259
        self.assertFalse(server._http_running)
330
260
 
331
261
    def test_create_http_server_one_zero(self):
334
264
            protocol_version = 'HTTP/1.0'
335
265
 
336
266
        server = http_server.HttpServer(RequestHandlerOneZero)
337
 
        self.start_server(server)
 
267
        server.setUp()
 
268
        self.addCleanup(server.tearDown)
338
269
        self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
339
270
 
340
271
    def test_create_http_server_one_one(self):
343
274
            protocol_version = 'HTTP/1.1'
344
275
 
345
276
        server = http_server.HttpServer(RequestHandlerOneOne)
346
 
        self.start_server(server)
 
277
        server.setUp()
 
278
        self.addCleanup(server.tearDown)
347
279
        self.assertIsInstance(server._httpd,
348
280
                              http_server.TestingThreadingHTTPServer)
349
281
 
354
286
 
355
287
        server = http_server.HttpServer(RequestHandlerOneZero,
356
288
                                        protocol_version='HTTP/1.1')
357
 
        self.start_server(server)
 
289
        server.setUp()
 
290
        self.addCleanup(server.tearDown)
358
291
        self.assertIsInstance(server._httpd,
359
292
                              http_server.TestingThreadingHTTPServer)
360
293
 
365
298
 
366
299
        server = http_server.HttpServer(RequestHandlerOneOne,
367
300
                                        protocol_version='HTTP/1.0')
368
 
        self.start_server(server)
 
301
        server.setUp()
 
302
        self.addCleanup(server.tearDown)
369
303
        self.assertIsInstance(server._httpd,
370
304
                              http_server.TestingHTTPServer)
371
305
 
374
308
    """Test case to inherit from if pycurl is present"""
375
309
 
376
310
    def _get_pycurl_maybe(self):
377
 
        self.requireFeature(features.pycurl)
378
 
        return PyCurlTransport
 
311
        try:
 
312
            from bzrlib.transport.http._pycurl import PyCurlTransport
 
313
            return PyCurlTransport
 
314
        except errors.DependencyNotPresent:
 
315
            raise tests.TestSkipped('pycurl not present')
379
316
 
380
317
    _transport = property(_get_pycurl_maybe)
381
318
 
388
325
    def test_url_parsing(self):
389
326
        f = FakeManager()
390
327
        url = http.extract_auth('http://example.com', f)
391
 
        self.assertEqual('http://example.com', url)
392
 
        self.assertEqual(0, len(f.credentials))
 
328
        self.assertEquals('http://example.com', url)
 
329
        self.assertEquals(0, len(f.credentials))
393
330
        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])
 
331
            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
 
332
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
 
333
        self.assertEquals(1, len(f.credentials))
 
334
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
 
335
                          f.credentials[0])
399
336
 
400
337
 
401
338
class TestHttpTransportUrls(tests.TestCase):
428
365
    def test_http_impl_urls(self):
429
366
        """There are servers which ask for particular clients to connect"""
430
367
        server = self._server()
431
 
        server.start_server()
432
368
        try:
 
369
            server.setUp()
433
370
            url = server.get_url()
434
371
            self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
435
372
        finally:
436
 
            server.stop_server()
 
373
            server.tearDown()
437
374
 
438
375
 
439
376
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
448
385
        https by supplying a fake version_info that do not
449
386
        support it.
450
387
        """
451
 
        self.requireFeature(features.pycurl)
452
 
        # Import the module locally now that we now it's available.
453
 
        pycurl = features.pycurl.module
 
388
        try:
 
389
            import pycurl
 
390
        except ImportError:
 
391
            raise tests.TestSkipped('pycurl not present')
454
392
 
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')
 
393
        version_info_orig = pycurl.version_info
 
394
        try:
 
395
            # Now that we have pycurl imported, we can fake its version_info
 
396
            # This was taken from a windows pycurl without SSL
 
397
            # (thanks to bialix)
 
398
            pycurl.version_info = lambda : (2,
 
399
                                            '7.13.2',
 
400
                                            462082,
 
401
                                            'i386-pc-win32',
 
402
                                            2576,
 
403
                                            None,
 
404
                                            0,
 
405
                                            None,
 
406
                                            ('ftp', 'gopher', 'telnet',
 
407
                                             'dict', 'ldap', 'http', 'file'),
 
408
                                            None,
 
409
                                            0,
 
410
                                            None)
 
411
            self.assertRaises(errors.DependencyNotPresent, self._transport,
 
412
                              'https://launchpad.net')
 
413
        finally:
 
414
            # Restore the right function
 
415
            pycurl.version_info = version_info_orig
473
416
 
474
417
 
475
418
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
534
477
class TestPost(tests.TestCase):
535
478
 
536
479
    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()
 
480
        server = RecordingServer(expect_body_tail='end-of-body')
 
481
        server.setUp()
 
482
        self.addCleanup(server.tearDown)
 
483
        scheme = self._qualified_prefix
 
484
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
541
485
        http_transport = self._transport(url)
542
486
        code, response = http_transport._post('abc def end-of-body')
543
487
        self.assertTrue(
592
536
                                      protocol_version=self._protocol_version)
593
537
 
594
538
    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)
 
539
        return pycurl_present and self._transport == PyCurlTransport
598
540
 
599
541
 
600
542
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
617
559
        # for details) make no distinction between a closed
618
560
        # socket and badly formatted status line, so we can't
619
561
        # 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),
 
562
        # InvalidHttpResponse too.
 
563
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
624
564
                          t.has, 'foo/bar')
625
565
 
626
566
    def test_http_get(self):
627
567
        server = self.get_readonly_server()
628
568
        t = self._transport(server.get_url())
629
 
        self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
630
 
                           errors.InvalidHttpResponse),
 
569
        self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
631
570
                          t.get, 'foo/bar')
632
571
 
633
572
 
709
648
    _req_handler_class = BadProtocolRequestHandler
710
649
 
711
650
    def setUp(self):
712
 
        if self._testing_pycurl():
 
651
        if pycurl_present and self._transport == PyCurlTransport:
713
652
            raise tests.TestNotApplicable(
714
653
                "pycurl doesn't check the protocol version")
715
654
        super(TestBadProtocolServer, self).setUp()
759
698
        self.assertEqual(None, server.host)
760
699
        self.assertEqual(None, server.port)
761
700
 
762
 
    def test_setUp_and_stop(self):
 
701
    def test_setUp_and_tearDown(self):
763
702
        server = RecordingServer(expect_body_tail=None)
764
 
        server.start_server()
 
703
        server.setUp()
765
704
        try:
766
705
            self.assertNotEqual(None, server.host)
767
706
            self.assertNotEqual(None, server.port)
768
707
        finally:
769
 
            server.stop_server()
 
708
            server.tearDown()
770
709
        self.assertEqual(None, server.host)
771
710
        self.assertEqual(None, server.port)
772
711
 
773
712
    def test_send_receive_bytes(self):
774
 
        server = RecordingServer(expect_body_tail='c', scheme='http')
775
 
        self.start_server(server)
 
713
        server = RecordingServer(expect_body_tail='c')
 
714
        server.setUp()
 
715
        self.addCleanup(server.tearDown)
776
716
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
777
717
        sock.connect((server.host, server.port))
778
718
        sock.sendall('abc')
872
812
        # bytes on the socket
873
813
        ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
874
814
        self.assertEqual((0, '0'), ireadv.next())
875
 
        # The server should have issued one request so far
 
815
        # The server should have issued one request so far 
876
816
        self.assertEqual(1, server.GET_request_nb)
877
817
        self.assertEqual('0123456789', t.get_bytes('a'))
878
818
        # get_bytes issued an additional request, the readv pending ones are
1169
1109
        if self._testing_pycurl():
1170
1110
            # Oh my ! pycurl does not check for the port as part of
1171
1111
            # no_proxy :-( So we just test the host part
1172
 
            self.no_proxy_host = self.server.host
 
1112
            self.no_proxy_host = 'localhost'
1173
1113
        else:
1174
1114
            self.no_proxy_host = self.proxy_address
1175
1115
        # The secondary server is the proxy
1178
1118
        self._old_env = {}
1179
1119
 
1180
1120
    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)
 
1121
        return pycurl_present and self._transport == PyCurlTransport
1184
1122
 
1185
1123
    def create_transport_secondary_server(self):
1186
1124
        """Creates an http server that will serve files with
1337
1275
    def test_read_redirected_bundle_from_url(self):
1338
1276
        from bzrlib.bundle import read_bundle_from_url
1339
1277
        url = self.old_transport.abspath('bundle')
1340
 
        bundle = self.applyDeprecated(deprecated_in((1, 12, 0)),
1341
 
                read_bundle_from_url, url)
 
1278
        bundle = read_bundle_from_url(url)
1342
1279
        # If read_bundle_from_url was successful we get an empty bundle
1343
1280
        self.assertEqual([], bundle.revisions)
1344
1281
 
1355
1292
        # Since the tests using this class will replace
1356
1293
        # _urllib2_wrappers.Request, we can't just call the base class __init__
1357
1294
        # or we'll loop.
1358
 
        RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
 
1295
        RedirectedRequest.init_orig(self, method, url, args, kwargs)
1359
1296
        self.follow_redirections = True
1360
1297
 
1361
1298
 
1362
 
def install_redirected_request(test):
1363
 
    test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1364
 
 
1365
 
 
1366
1299
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1367
1300
    """Test redirections.
1368
1301
 
1378
1311
    """
1379
1312
 
1380
1313
    def setUp(self):
1381
 
        if (features.pycurl.available()
1382
 
            and self._transport == PyCurlTransport):
 
1314
        if pycurl_present and self._transport == PyCurlTransport:
1383
1315
            raise tests.TestNotApplicable(
1384
1316
                "pycurl doesn't redirect silently annymore")
1385
1317
        super(TestHTTPSilentRedirections, self).setUp()
1386
 
        install_redirected_request(self)
 
1318
        self.setup_redirected_request()
 
1319
        self.addCleanup(self.cleanup_redirected_request)
1387
1320
        self.build_tree_contents([('a','a'),
1388
1321
                                  ('1/',),
1389
1322
                                  ('1/a', 'redirected once'),
1399
1332
 
1400
1333
        self.old_transport = self._transport(self.old_server.get_url())
1401
1334
 
 
1335
    def setup_redirected_request(self):
 
1336
        self.original_class = _urllib2_wrappers.Request
 
1337
        _urllib2_wrappers.Request = RedirectedRequest
 
1338
 
 
1339
    def cleanup_redirected_request(self):
 
1340
        _urllib2_wrappers.Request = self.original_class
 
1341
 
1402
1342
    def create_transport_secondary_server(self):
1403
1343
        """Create the secondary server, redirections are defined in the tests"""
1404
1344
        return http_utils.HTTPServerRedirecting(
1408
1348
        t = self.old_transport
1409
1349
 
1410
1350
        req = RedirectedRequest('GET', t.abspath('a'))
 
1351
        req.follow_redirections = True
1411
1352
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1412
1353
                                       self.new_server.port)
1413
1354
        self.old_server.redirections = \
1414
1355
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
1415
 
        self.assertEqual('redirected once',t._perform(req).read())
 
1356
        self.assertEquals('redirected once',t._perform(req).read())
1416
1357
 
1417
1358
    def test_five_redirections(self):
1418
1359
        t = self.old_transport
1419
1360
 
1420
1361
        req = RedirectedRequest('GET', t.abspath('a'))
 
1362
        req.follow_redirections = True
1421
1363
        old_prefix = 'http://%s:%s' % (self.old_server.host,
1422
1364
                                       self.old_server.port)
1423
1365
        new_prefix = 'http://%s:%s' % (self.new_server.host,
1429
1371
            ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1430
1372
            ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1431
1373
            ]
1432
 
        self.assertEqual('redirected 5 times',t._perform(req).read())
 
1374
        self.assertEquals('redirected 5 times',t._perform(req).read())
1433
1375
 
1434
1376
 
1435
1377
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1448
1390
        t = self._transport(self.new_server.get_url())
1449
1391
 
1450
1392
        # We use None for redirected so that we fail if redirected
1451
 
        self.assertEqual('0123456789',
1452
 
                         transport.do_catching_redirections(
 
1393
        self.assertEquals('0123456789',
 
1394
                          transport.do_catching_redirections(
1453
1395
                self.get_a, t, None).read())
1454
1396
 
1455
1397
    def test_one_redirection(self):
1460
1402
            dir, file = urlutils.split(exception.target)
1461
1403
            return self._transport(dir)
1462
1404
 
1463
 
        self.assertEqual('0123456789',
1464
 
                         transport.do_catching_redirections(
 
1405
        self.assertEquals('0123456789',
 
1406
                          transport.do_catching_redirections(
1465
1407
                self.get_a, self.old_transport, redirected).read())
1466
 
        self.assertEqual(1, self.redirections)
 
1408
        self.assertEquals(1, self.redirections)
1467
1409
 
1468
1410
    def test_redirection_loop(self):
1469
1411
 
1483
1425
 
1484
1426
    _auth_header = 'Authorization'
1485
1427
    _password_prompt_prefix = ''
1486
 
    _username_prompt_prefix = ''
1487
 
    # Set by load_tests
1488
 
    _auth_server = None
1489
1428
 
1490
1429
    def setUp(self):
1491
1430
        super(TestAuth, self).setUp()
1494
1433
                                  ('b', 'contents of b\n'),])
1495
1434
 
1496
1435
    def create_transport_readonly_server(self):
1497
 
        return self._auth_server(protocol_version=self._protocol_version)
 
1436
        if self._auth_scheme == 'basic':
 
1437
            server = http_utils.HTTPBasicAuthServer(
 
1438
                protocol_version=self._protocol_version)
 
1439
        else:
 
1440
            if self._auth_scheme != 'digest':
 
1441
                raise AssertionError('Unknown auth scheme: %r'
 
1442
                                     % self._auth_scheme)
 
1443
            server = http_utils.HTTPDigestAuthServer(
 
1444
                protocol_version=self._protocol_version)
 
1445
        return server
1498
1446
 
1499
1447
    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)
 
1448
        return pycurl_present and self._transport == PyCurlTransport
1503
1449
 
1504
1450
    def get_user_url(self, user, password):
1505
1451
        """Build an url embedding user and password"""
1553
1499
        # initial 'who are you' and 'this is not you, who are you')
1554
1500
        self.assertEqual(2, self.server.auth_required_errors)
1555
1501
 
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
1502
    def test_prompt_for_password(self):
1579
1503
        if self._testing_pycurl():
1580
1504
            raise tests.TestNotApplicable(
1584
1508
        self.server.add_user('joe', 'foo')
1585
1509
        t = self.get_user_transport('joe', None)
1586
1510
        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())
 
1511
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
 
1512
        self.assertEqual('contents of a\n',t.get('a').read())
1591
1513
        # stdin should be empty
1592
1514
        self.assertEqual('', ui.ui_factory.stdin.readline())
1593
1515
        self._check_password_prompt(t._unqualified_scheme, 'joe',
1594
 
                                    stderr.getvalue())
1595
 
        self.assertEqual('', stdout.getvalue())
 
1516
                                    stdout.getvalue())
1596
1517
        # And we shouldn't prompt again for a different request
1597
1518
        # against the same transport.
1598
1519
        self.assertEqual('contents of b\n',t.get('b').read())
1608
1529
                              % (scheme.upper(),
1609
1530
                                 user, self.server.host, self.server.port,
1610
1531
                                 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))
 
1532
        self.assertEquals(expected_prompt, actual_prompt)
1618
1533
 
1619
1534
    def test_no_prompt_for_password_when_using_auth_config(self):
1620
1535
        if self._testing_pycurl():
1628
1543
        self.server.add_user(user, password)
1629
1544
        t = self.get_user_transport(user, None)
1630
1545
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1631
 
                                            stderr=tests.StringIOWrapper())
 
1546
                                            stdout=tests.StringIOWrapper())
1632
1547
        # Create a minimal config file with the right password
1633
1548
        conf = config.AuthenticationConfig()
1634
1549
        conf._get_config().update(
1662
1577
        self.assertEqual(1, self.server.auth_required_errors)
1663
1578
 
1664
1579
    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')
 
1580
        if self._auth_scheme != 'digest':
 
1581
            raise tests.TestNotApplicable('HTTP auth digest only test')
1668
1582
        if self._testing_pycurl():
1669
1583
            raise tests.KnownFailure(
1670
1584
                'pycurl does not handle a nonce change')
1688
1602
    """Test proxy authentication schemes."""
1689
1603
 
1690
1604
    _auth_header = 'Proxy-authorization'
1691
 
    _password_prompt_prefix = 'Proxy '
1692
 
    _username_prompt_prefix = 'Proxy '
 
1605
    _password_prompt_prefix='Proxy '
1693
1606
 
1694
1607
    def setUp(self):
1695
1608
        super(TestProxyAuth, self).setUp()
1702
1615
                                  ('b-proxied', 'contents of b\n'),
1703
1616
                                  ])
1704
1617
 
 
1618
    def create_transport_readonly_server(self):
 
1619
        if self._auth_scheme == 'basic':
 
1620
            server = http_utils.ProxyBasicAuthServer(
 
1621
                protocol_version=self._protocol_version)
 
1622
        else:
 
1623
            if self._auth_scheme != 'digest':
 
1624
                raise AssertionError('Unknown auth scheme: %r'
 
1625
                                     % self._auth_scheme)
 
1626
            server = http_utils.ProxyDigestAuthServer(
 
1627
                protocol_version=self._protocol_version)
 
1628
        return server
 
1629
 
1705
1630
    def get_user_transport(self, user, password):
1706
1631
        self._install_env({'all_proxy': self.get_user_url(user, password)})
1707
1632
        return self._transport(self.server.get_url())
1846
1771
                             'http://www.example.com/foo/subdir')
1847
1772
        self.assertIsInstance(r, type(t))
1848
1773
        # Both transports share the some connection
1849
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1774
        self.assertEquals(t._get_connection(), r._get_connection())
1850
1775
 
1851
1776
    def test_redirected_to_self_with_slash(self):
1852
1777
        t = self._transport('http://www.example.com/foo')
1856
1781
        # Both transports share the some connection (one can argue that we
1857
1782
        # should return the exact same transport here, but that seems
1858
1783
        # overkill).
1859
 
        self.assertEqual(t._get_connection(), r._get_connection())
 
1784
        self.assertEquals(t._get_connection(), r._get_connection())
1860
1785
 
1861
1786
    def test_redirected_to_host(self):
1862
1787
        t = self._transport('http://www.example.com/foo')
1881
1806
        r = t._redirected_to('http://www.example.com/foo',
1882
1807
                             'https://foo.example.com/foo')
1883
1808
        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
 
 
 
1809
        self.assertEquals(t._user, r._user)