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
17
"""Tests for HTTP implementations.
19
This module defines a load_tests() method that parametrize tests classes for
20
transport implementation, http protocol versions and authentication schemes.
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
23
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
24
# TODO: What about renaming to bzrlib.tests.transport.http ?
27
import SimpleHTTPServer
38
remote as _mod_remote,
43
from bzrlib.tests import (
49
from bzrlib.tests.scenarios import (
50
load_tests_apply_scenarios,
53
from bzrlib.transport import (
57
from bzrlib.transport.http import (
63
if features.pycurl.available():
64
from bzrlib.transport.http._pycurl import PyCurlTransport
67
load_tests = load_tests_apply_scenarios
70
def vary_by_http_client_implementation():
71
"""Test the two libraries we can use, pycurl and urllib."""
72
transport_scenarios = [
73
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
74
_server=http_server.HttpServer_urllib,
75
_url_protocol='http+urllib',)),
77
if features.pycurl.available():
78
transport_scenarios.append(
79
('pycurl', dict(_transport=PyCurlTransport,
80
_server=http_server.HttpServer_PyCurl,
81
_url_protocol='http+pycurl',)))
82
return transport_scenarios
85
def vary_by_http_protocol_version():
86
"""Test on http/1.0 and 1.1"""
88
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
89
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
93
def vary_by_http_proxy_auth_scheme():
95
('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
96
('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
98
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
102
def vary_by_http_auth_scheme():
104
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
105
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
107
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
111
def vary_by_http_activity():
112
activity_scenarios = [
113
('urllib,http', dict(_activity_server=ActivityHTTPServer,
114
_transport=_urllib.HttpTransport_urllib,)),
116
if tests.HTTPSServerFeature.available():
117
activity_scenarios.append(
118
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
119
_transport=_urllib.HttpTransport_urllib,)),)
120
if features.pycurl.available():
121
activity_scenarios.append(
122
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
123
_transport=PyCurlTransport,)),)
124
if tests.HTTPSServerFeature.available():
125
from bzrlib.tests import (
128
# FIXME: Until we have a better way to handle self-signed
129
# certificates (like allowing them in a test specific
130
# authentication.conf for example), we need some specialized pycurl
131
# transport for tests.
132
class HTTPS_pycurl_transport(PyCurlTransport):
134
def __init__(self, base, _from_transport=None):
135
super(HTTPS_pycurl_transport, self).__init__(
136
base, _from_transport)
137
self.cabundle = str(ssl_certs.build_path('ca.crt'))
139
activity_scenarios.append(
140
('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
141
_transport=HTTPS_pycurl_transport,)),)
142
return activity_scenarios
145
class FakeManager(object):
23
from bzrlib.errors import DependencyNotPresent
24
from bzrlib.tests import TestCase, TestSkipped
25
from bzrlib.transport import Transport
26
from bzrlib.transport.http import extract_auth
27
from bzrlib.transport.http._urllib import HttpTransport_urllib
28
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
30
class FakeManager (object):
147
31
def __init__(self):
148
32
self.credentials = []
150
34
def add_password(self, realm, host, username, password):
151
35
self.credentials.append([realm, host, username, password])
154
class RecordingServer(object):
155
"""A fake HTTP server.
157
It records the bytes sent to it, and replies with a 200.
160
def __init__(self, expect_body_tail=None, scheme=''):
163
:type expect_body_tail: str
164
:param expect_body_tail: a reply won't be sent until this string is
167
self._expect_body_tail = expect_body_tail
170
self.received_bytes = ''
174
return '%s://%s:%s/' % (self.scheme, self.host, self.port)
176
def start_server(self):
177
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
178
self._sock.bind(('127.0.0.1', 0))
179
self.host, self.port = self._sock.getsockname()
180
self._ready = threading.Event()
181
self._thread = test_server.ThreadWithException(
182
event=self._ready, target=self._accept_read_and_reply)
184
if 'threads' in tests.selftest_debug_flags:
185
sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
188
def _accept_read_and_reply(self):
191
conn, address = self._sock.accept()
192
if self._expect_body_tail is not None:
193
while not self.received_bytes.endswith(self._expect_body_tail):
194
self.received_bytes += conn.recv(4096)
195
conn.sendall('HTTP/1.1 200 OK\r\n')
199
# The client may have already closed the socket.
202
def stop_server(self):
204
# Issue a fake connection to wake up the server and allow it to
206
fake_conn = osutils.connect_socket((self.host, self.port))
209
# We might have already closed it. We don't care.
214
if 'threads' in tests.selftest_debug_flags:
215
sys.stderr.write('Thread joined: %s\n' % (self._thread.ident,))
218
class TestAuthHeader(tests.TestCase):
220
def parse_header(self, header, auth_handler_class=None):
221
if auth_handler_class is None:
222
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
223
self.auth_handler = auth_handler_class()
224
return self.auth_handler._parse_auth_header(header)
226
def test_empty_header(self):
227
scheme, remainder = self.parse_header('')
228
self.assertEqual('', scheme)
229
self.assertIs(None, remainder)
231
def test_negotiate_header(self):
232
scheme, remainder = self.parse_header('Negotiate')
233
self.assertEqual('negotiate', scheme)
234
self.assertIs(None, remainder)
236
def test_basic_header(self):
237
scheme, remainder = self.parse_header(
238
'Basic realm="Thou should not pass"')
239
self.assertEqual('basic', scheme)
240
self.assertEqual('realm="Thou should not pass"', remainder)
242
def test_basic_extract_realm(self):
243
scheme, remainder = self.parse_header(
244
'Basic realm="Thou should not pass"',
245
_urllib2_wrappers.BasicAuthHandler)
246
match, realm = self.auth_handler.extract_realm(remainder)
247
self.assertTrue(match is not None)
248
self.assertEqual('Thou should not pass', realm)
250
def test_digest_header(self):
251
scheme, remainder = self.parse_header(
252
'Digest realm="Thou should not pass"')
253
self.assertEqual('digest', scheme)
254
self.assertEqual('realm="Thou should not pass"', remainder)
257
class TestHTTPServer(tests.TestCase):
258
"""Test the HTTP servers implementations."""
260
def test_invalid_protocol(self):
261
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
263
protocol_version = 'HTTP/0.1'
265
self.assertRaises(httplib.UnknownProtocol,
266
http_server.HttpServer, BogusRequestHandler)
268
def test_force_invalid_protocol(self):
269
self.assertRaises(httplib.UnknownProtocol,
270
http_server.HttpServer, protocol_version='HTTP/0.1')
272
def test_server_start_and_stop(self):
273
server = http_server.HttpServer()
274
self.addCleanup(server.stop_server)
275
server.start_server()
276
self.assertTrue(server.server is not None)
277
self.assertTrue(server.server.serving is not None)
278
self.assertTrue(server.server.serving)
280
def test_create_http_server_one_zero(self):
281
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
283
protocol_version = 'HTTP/1.0'
285
server = http_server.HttpServer(RequestHandlerOneZero)
286
self.start_server(server)
287
self.assertIsInstance(server.server, http_server.TestingHTTPServer)
289
def test_create_http_server_one_one(self):
290
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
292
protocol_version = 'HTTP/1.1'
294
server = http_server.HttpServer(RequestHandlerOneOne)
295
self.start_server(server)
296
self.assertIsInstance(server.server,
297
http_server.TestingThreadingHTTPServer)
299
def test_create_http_server_force_one_one(self):
300
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
302
protocol_version = 'HTTP/1.0'
304
server = http_server.HttpServer(RequestHandlerOneZero,
305
protocol_version='HTTP/1.1')
306
self.start_server(server)
307
self.assertIsInstance(server.server,
308
http_server.TestingThreadingHTTPServer)
310
def test_create_http_server_force_one_zero(self):
311
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
313
protocol_version = 'HTTP/1.1'
315
server = http_server.HttpServer(RequestHandlerOneOne,
316
protocol_version='HTTP/1.0')
317
self.start_server(server)
318
self.assertIsInstance(server.server,
319
http_server.TestingHTTPServer)
322
class TestWithTransport_pycurl(object):
323
"""Test case to inherit from if pycurl is present"""
325
def _get_pycurl_maybe(self):
326
self.requireFeature(features.pycurl)
327
return PyCurlTransport
329
_transport = property(_get_pycurl_maybe)
332
class TestHttpUrls(tests.TestCase):
334
# TODO: This should be moved to authorization tests once they
38
class TestHttpUrls(TestCase):
337
39
def test_url_parsing(self):
339
url = http.extract_auth('http://example.com', f)
340
self.assertEqual('http://example.com', url)
341
self.assertEqual(0, len(f.credentials))
342
url = http.extract_auth(
343
'http://user:pass@example.com/bzr/bzr.dev', f)
344
self.assertEqual('http://example.com/bzr/bzr.dev', url)
345
self.assertEqual(1, len(f.credentials))
346
self.assertEqual([None, 'example.com', 'user', 'pass'],
350
class TestHttpTransportUrls(tests.TestCase):
351
"""Test the http urls."""
353
scenarios = vary_by_http_client_implementation()
41
url = extract_auth('http://example.com', f)
42
self.assertEquals('http://example.com', url)
43
self.assertEquals(0, len(f.credentials))
44
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
45
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
46
self.assertEquals(1, len(f.credentials))
47
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'], f.credentials[0])
355
49
def test_abs_url(self):
356
50
"""Construction of absolute http URLs"""
357
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
51
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
358
52
eq = self.assertEqualDiff
359
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
360
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
361
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
54
'http://bazaar-vcs.org/bzr/bzr.dev')
55
eq(t.abspath('foo/bar'),
56
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
58
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
362
59
eq(t.abspath('.bzr/1//2/./3'),
363
60
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
365
62
def test_invalid_http_urls(self):
366
63
"""Trap invalid construction of urls"""
367
self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
368
self.assertRaises(errors.InvalidURL,
370
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
64
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
65
self.assertRaises(ValueError,
68
self.assertRaises(ValueError,
372
72
def test_http_root_urls(self):
373
73
"""Construction of URLs from server root"""
374
t = self._transport('http://bzr.ozlabs.org/')
74
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
375
75
eq = self.assertEqualDiff
376
76
eq(t.abspath('.bzr/tree-version'),
377
77
'http://bzr.ozlabs.org/.bzr/tree-version')
379
79
def test_http_impl_urls(self):
380
80
"""There are servers which ask for particular clients to connect"""
381
server = self._server()
382
server.start_server()
384
url = server.get_url()
385
self.assertTrue(url.startswith('%s://' % self._url_protocol))
390
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
392
# TODO: This should really be moved into another pycurl
393
# specific test. When https tests will be implemented, take
394
# this one into account.
395
def test_pycurl_without_https_support(self):
396
"""Test that pycurl without SSL do not fail with a traceback.
398
For the purpose of the test, we force pycurl to ignore
399
https by supplying a fake version_info that do not
402
self.requireFeature(features.pycurl)
403
# Import the module locally now that we now it's available.
404
pycurl = features.pycurl.module
406
self.overrideAttr(pycurl, 'version_info',
407
# Fake the pycurl version_info This was taken from
408
# a windows pycurl without SSL (thanks to bialix)
417
('ftp', 'gopher', 'telnet',
418
'dict', 'ldap', 'http', 'file'),
422
self.assertRaises(errors.DependencyNotPresent, self._transport,
423
'https://launchpad.net')
426
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
427
"""Test the http connections."""
429
scenarios = multiply_scenarios(
430
vary_by_http_client_implementation(),
431
vary_by_http_protocol_version(),
435
http_utils.TestCaseWithWebserver.setUp(self)
436
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
82
from bzrlib.transport.http._pycurl import HttpServer_PyCurl
83
server = HttpServer_PyCurl()
86
url = server.get_url()
87
self.assertTrue(url.startswith('http+pycurl://'))
90
except DependencyNotPresent:
91
raise TestSkipped('pycurl not present')
93
class TestHttpMixins(object):
96
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
437
97
transport=self.get_transport())
439
99
def test_http_has(self):
440
100
server = self.get_readonly_server()
441
t = self.get_readonly_transport()
101
t = self._transport(server.get_url())
442
102
self.assertEqual(t.has('foo/bar'), True)
443
103
self.assertEqual(len(server.logs), 1)
444
self.assertContainsRe(server.logs[0],
104
self.assertContainsRe(server.logs[0],
445
105
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
447
107
def test_http_has_not_found(self):
448
108
server = self.get_readonly_server()
449
t = self.get_readonly_transport()
109
t = self._transport(server.get_url())
450
110
self.assertEqual(t.has('not-found'), False)
451
self.assertContainsRe(server.logs[1],
111
self.assertContainsRe(server.logs[1],
452
112
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
454
114
def test_http_get(self):
455
115
server = self.get_readonly_server()
456
t = self.get_readonly_transport()
116
t = self._transport(server.get_url())
457
117
fp = t.get('foo/bar')
458
118
self.assertEqualDiff(
460
120
'contents of foo/bar\n')
461
121
self.assertEqual(len(server.logs), 1)
462
122
self.assertTrue(server.logs[0].find(
463
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
464
% bzrlib.__version__) > -1)
466
def test_has_on_bogus_host(self):
467
# Get a free address and don't 'accept' on it, so that we
468
# can be sure there is no http handler there, but set a
469
# reasonable timeout to not slow down tests too much.
470
default_timeout = socket.getdefaulttimeout()
123
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s' % bzrlib.__version__) > -1)
126
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
127
_transport = HttpTransport_urllib
130
TestCaseWithWebserver.setUp(self)
135
class TestHttpConnections_pycurl(TestCaseWithWebserver, TestHttpMixins):
137
def _get_pycurl_maybe(self):
472
socket.setdefaulttimeout(2)
474
s.bind(('localhost', 0))
475
t = self._transport('http://%s:%s/' % s.getsockname())
476
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
478
socket.setdefaulttimeout(default_timeout)
481
class TestHttpTransportRegistration(tests.TestCase):
139
from bzrlib.transport.http._pycurl import PyCurlTransport
140
return PyCurlTransport
141
except DependencyNotPresent:
142
raise TestSkipped('pycurl not present')
144
_transport = property(_get_pycurl_maybe)
147
TestCaseWithWebserver.setUp(self)
152
class TestHttpTransportRegistration(TestCase):
482
153
"""Test registrations of various http implementations"""
484
scenarios = vary_by_http_client_implementation()
486
155
def test_http_registered(self):
487
t = transport.get_transport('%s://foo.com/' % self._url_protocol)
488
self.assertIsInstance(t, transport.Transport)
489
self.assertIsInstance(t, self._transport)
492
class TestPost(tests.TestCase):
494
scenarios = multiply_scenarios(
495
vary_by_http_client_implementation(),
496
vary_by_http_protocol_version(),
499
def test_post_body_is_received(self):
500
server = RecordingServer(expect_body_tail='end-of-body',
501
scheme=self._url_protocol)
502
self.start_server(server)
503
url = server.get_url()
504
# FIXME: needs a cleanup -- vila 20100611
505
http_transport = transport.get_transport(url)
506
code, response = http_transport._post('abc def end-of-body')
508
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
509
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
510
self.assertTrue('content-type: application/octet-stream\r'
511
in server.received_bytes.lower())
512
# The transport should not be assuming that the server can accept
513
# chunked encoding the first time it connects, because HTTP/1.1, so we
514
# check for the literal string.
516
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
519
class TestRangeHeader(tests.TestCase):
520
"""Test range_header method"""
522
def check_header(self, value, ranges=[], tail=0):
523
offsets = [ (start, end - start + 1) for start, end in ranges]
524
coalesce = transport.Transport._coalesce_offsets
525
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
526
range_header = http.HttpTransportBase._range_header
527
self.assertEqual(value, range_header(coalesced, tail))
529
def test_range_header_single(self):
530
self.check_header('0-9', ranges=[(0,9)])
531
self.check_header('100-109', ranges=[(100,109)])
533
def test_range_header_tail(self):
534
self.check_header('-10', tail=10)
535
self.check_header('-50', tail=50)
537
def test_range_header_multi(self):
538
self.check_header('0-9,100-200,300-5000',
539
ranges=[(0,9), (100, 200), (300,5000)])
541
def test_range_header_mixed(self):
542
self.check_header('0-9,300-5000,-50',
543
ranges=[(0,9), (300,5000)],
547
class TestSpecificRequestHandler(http_utils.TestCaseWithWebserver):
548
"""Tests a specific request handler.
550
Daughter classes are expected to override _req_handler_class
553
scenarios = multiply_scenarios(
554
vary_by_http_client_implementation(),
555
vary_by_http_protocol_version(),
558
# Provide a useful default
559
_req_handler_class = http_server.TestingHTTPRequestHandler
561
def create_transport_readonly_server(self):
562
server = http_server.HttpServer(self._req_handler_class,
563
protocol_version=self._protocol_version)
564
server._url_protocol = self._url_protocol
567
def _testing_pycurl(self):
568
# TODO: This is duplicated for lots of the classes in this file
569
return (features.pycurl.available()
570
and self._transport == PyCurlTransport)
573
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
574
"""Whatever request comes in, close the connection"""
576
def _handle_one_request(self):
577
"""Handle a single HTTP request, by abruptly closing the connection"""
578
self.close_connection = 1
581
class TestWallServer(TestSpecificRequestHandler):
582
"""Tests exceptions during the connection phase"""
584
_req_handler_class = WallRequestHandler
586
def test_http_has(self):
587
t = self.get_readonly_transport()
588
# Unfortunately httplib (see HTTPResponse._read_status
589
# for details) make no distinction between a closed
590
# socket and badly formatted status line, so we can't
591
# just test for ConnectionError, we have to test
592
# InvalidHttpResponse too. And pycurl may raise ConnectionReset
593
# instead of ConnectionError too.
594
self.assertRaises(( errors.ConnectionError, errors.ConnectionReset,
595
errors.InvalidHttpResponse),
598
def test_http_get(self):
599
t = self.get_readonly_transport()
600
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
601
errors.InvalidHttpResponse),
605
class BadStatusRequestHandler(http_server.TestingHTTPRequestHandler):
606
"""Whatever request comes in, returns a bad status"""
608
def parse_request(self):
609
"""Fakes handling a single HTTP request, returns a bad status"""
610
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
611
self.send_response(0, "Bad status")
612
self.close_connection = 1
616
class TestBadStatusServer(TestSpecificRequestHandler):
617
"""Tests bad status from server."""
619
_req_handler_class = BadStatusRequestHandler
621
def test_http_has(self):
622
t = self.get_readonly_transport()
623
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
625
def test_http_get(self):
626
t = self.get_readonly_transport()
627
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
630
class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):
631
"""Whatever request comes in, returns an invalid status"""
633
def parse_request(self):
634
"""Fakes handling a single HTTP request, returns a bad status"""
635
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
636
self.wfile.write("Invalid status line\r\n")
637
# If we don't close the connection pycurl will hang. Since this is a
638
# stress test we don't *have* to respect the protocol, but we don't
639
# have to sabotage it too much either.
640
self.close_connection = True
644
class TestInvalidStatusServer(TestBadStatusServer):
645
"""Tests invalid status from server.
647
Both implementations raises the same error as for a bad status.
650
_req_handler_class = InvalidStatusRequestHandler
653
class BadProtocolRequestHandler(http_server.TestingHTTPRequestHandler):
654
"""Whatever request comes in, returns a bad protocol version"""
656
def parse_request(self):
657
"""Fakes handling a single HTTP request, returns a bad status"""
658
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
659
# Returns an invalid protocol version, but curl just
660
# ignores it and those cannot be tested.
661
self.wfile.write("%s %d %s\r\n" % ('HTTP/0.0',
663
'Look at my protocol version'))
667
class TestBadProtocolServer(TestSpecificRequestHandler):
668
"""Tests bad protocol from server."""
670
_req_handler_class = BadProtocolRequestHandler
673
if self._testing_pycurl():
674
raise tests.TestNotApplicable(
675
"pycurl doesn't check the protocol version")
676
super(TestBadProtocolServer, self).setUp()
678
def test_http_has(self):
679
t = self.get_readonly_transport()
680
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
682
def test_http_get(self):
683
t = self.get_readonly_transport()
684
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
687
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
688
"""Whatever request comes in, returns a 403 code"""
690
def parse_request(self):
691
"""Handle a single HTTP request, by replying we cannot handle it"""
692
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
697
class TestForbiddenServer(TestSpecificRequestHandler):
698
"""Tests forbidden server"""
700
_req_handler_class = ForbiddenRequestHandler
702
def test_http_has(self):
703
t = self.get_readonly_transport()
704
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
706
def test_http_get(self):
707
t = self.get_readonly_transport()
708
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
711
class TestRecordingServer(tests.TestCase):
713
def test_create(self):
714
server = RecordingServer(expect_body_tail=None)
715
self.assertEqual('', server.received_bytes)
716
self.assertEqual(None, server.host)
717
self.assertEqual(None, server.port)
719
def test_setUp_and_stop(self):
720
server = RecordingServer(expect_body_tail=None)
721
server.start_server()
723
self.assertNotEqual(None, server.host)
724
self.assertNotEqual(None, server.port)
727
self.assertEqual(None, server.host)
728
self.assertEqual(None, server.port)
730
def test_send_receive_bytes(self):
731
server = RecordingServer(expect_body_tail='c', scheme='http')
732
self.start_server(server)
733
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
734
sock.connect((server.host, server.port))
736
self.assertEqual('HTTP/1.1 200 OK\r\n',
737
osutils.recv_all(sock, 4096))
738
self.assertEqual('abc', server.received_bytes)
741
class TestRangeRequestServer(TestSpecificRequestHandler):
742
"""Tests readv requests against server.
744
We test against default "normal" server.
748
super(TestRangeRequestServer, self).setUp()
749
self.build_tree_contents([('a', '0123456789')],)
751
def test_readv(self):
752
t = self.get_readonly_transport()
753
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
754
self.assertEqual(l[0], (0, '0'))
755
self.assertEqual(l[1], (1, '1'))
756
self.assertEqual(l[2], (3, '34'))
757
self.assertEqual(l[3], (9, '9'))
759
def test_readv_out_of_order(self):
760
t = self.get_readonly_transport()
761
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
762
self.assertEqual(l[0], (1, '1'))
763
self.assertEqual(l[1], (9, '9'))
764
self.assertEqual(l[2], (0, '0'))
765
self.assertEqual(l[3], (3, '34'))
767
def test_readv_invalid_ranges(self):
768
t = self.get_readonly_transport()
770
# This is intentionally reading off the end of the file
771
# since we are sure that it cannot get there
772
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
773
t.readv, 'a', [(1,1), (8,10)])
775
# This is trying to seek past the end of the file, it should
776
# also raise a special error
777
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
778
t.readv, 'a', [(12,2)])
780
def test_readv_multiple_get_requests(self):
781
server = self.get_readonly_server()
782
t = self.get_readonly_transport()
783
# force transport to issue multiple requests
784
t._max_readv_combine = 1
785
t._max_get_ranges = 1
786
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
787
self.assertEqual(l[0], (0, '0'))
788
self.assertEqual(l[1], (1, '1'))
789
self.assertEqual(l[2], (3, '34'))
790
self.assertEqual(l[3], (9, '9'))
791
# The server should have issued 4 requests
792
self.assertEqual(4, server.GET_request_nb)
794
def test_readv_get_max_size(self):
795
server = self.get_readonly_server()
796
t = self.get_readonly_transport()
797
# force transport to issue multiple requests by limiting the number of
798
# bytes by request. Note that this apply to coalesced offsets only, a
799
# single range will keep its size even if bigger than the limit.
801
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
802
self.assertEqual(l[0], (0, '0'))
803
self.assertEqual(l[1], (1, '1'))
804
self.assertEqual(l[2], (2, '2345'))
805
self.assertEqual(l[3], (6, '6789'))
806
# The server should have issued 3 requests
807
self.assertEqual(3, server.GET_request_nb)
809
def test_complete_readv_leave_pipe_clean(self):
810
server = self.get_readonly_server()
811
t = self.get_readonly_transport()
812
# force transport to issue multiple requests
814
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
815
# The server should have issued 3 requests
816
self.assertEqual(3, server.GET_request_nb)
817
self.assertEqual('0123456789', t.get_bytes('a'))
818
self.assertEqual(4, server.GET_request_nb)
820
def test_incomplete_readv_leave_pipe_clean(self):
821
server = self.get_readonly_server()
822
t = self.get_readonly_transport()
823
# force transport to issue multiple requests
825
# Don't collapse readv results into a list so that we leave unread
826
# bytes on the socket
827
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
828
self.assertEqual((0, '0'), ireadv.next())
829
# The server should have issued one request so far
830
self.assertEqual(1, server.GET_request_nb)
831
self.assertEqual('0123456789', t.get_bytes('a'))
832
# get_bytes issued an additional request, the readv pending ones are
834
self.assertEqual(2, server.GET_request_nb)
837
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
838
"""Always reply to range request as if they were single.
840
Don't be explicit about it, just to annoy the clients.
843
def get_multiple_ranges(self, file, file_size, ranges):
844
"""Answer as if it was a single range request and ignores the rest"""
845
(start, end) = ranges[0]
846
return self.get_single_range(file, file_size, start, end)
849
class TestSingleRangeRequestServer(TestRangeRequestServer):
850
"""Test readv against a server which accept only single range requests"""
852
_req_handler_class = SingleRangeRequestHandler
855
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
856
"""Only reply to simple range requests, errors out on multiple"""
858
def get_multiple_ranges(self, file, file_size, ranges):
859
"""Refuses the multiple ranges request"""
862
self.send_error(416, "Requested range not satisfiable")
864
(start, end) = ranges[0]
865
return self.get_single_range(file, file_size, start, end)
868
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
869
"""Test readv against a server which only accept single range requests"""
871
_req_handler_class = SingleOnlyRangeRequestHandler
874
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
875
"""Ignore range requests without notice"""
878
# Update the statistics
879
self.server.test_case_server.GET_request_nb += 1
880
# Just bypass the range handling done by TestingHTTPRequestHandler
881
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
884
class TestNoRangeRequestServer(TestRangeRequestServer):
885
"""Test readv against a server which do not accept range requests"""
887
_req_handler_class = NoRangeRequestHandler
890
class MultipleRangeWithoutContentLengthRequestHandler(
891
http_server.TestingHTTPRequestHandler):
892
"""Reply to multiple range requests without content length header."""
894
def get_multiple_ranges(self, file, file_size, ranges):
895
self.send_response(206)
896
self.send_header('Accept-Ranges', 'bytes')
897
# XXX: this is strange; the 'random' name below seems undefined and
898
# yet the tests pass -- mbp 2010-10-11 bug 658773
899
boundary = "%d" % random.randint(0,0x7FFFFFFF)
900
self.send_header("Content-Type",
901
"multipart/byteranges; boundary=%s" % boundary)
903
for (start, end) in ranges:
904
self.wfile.write("--%s\r\n" % boundary)
905
self.send_header("Content-type", 'application/octet-stream')
906
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
910
self.send_range_content(file, start, end - start + 1)
912
self.wfile.write("--%s\r\n" % boundary)
915
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
917
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
920
class TruncatedMultipleRangeRequestHandler(
921
http_server.TestingHTTPRequestHandler):
922
"""Reply to multiple range requests truncating the last ones.
924
This server generates responses whose Content-Length describes all the
925
ranges, but fail to include the last ones leading to client short reads.
926
This has been observed randomly with lighttpd (bug #179368).
929
_truncated_ranges = 2
931
def get_multiple_ranges(self, file, file_size, ranges):
932
self.send_response(206)
933
self.send_header('Accept-Ranges', 'bytes')
935
self.send_header('Content-Type',
936
'multipart/byteranges; boundary=%s' % boundary)
937
boundary_line = '--%s\r\n' % boundary
938
# Calculate the Content-Length
940
for (start, end) in ranges:
941
content_length += len(boundary_line)
942
content_length += self._header_line_length(
943
'Content-type', 'application/octet-stream')
944
content_length += self._header_line_length(
945
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
946
content_length += len('\r\n') # end headers
947
content_length += end - start # + 1
948
content_length += len(boundary_line)
949
self.send_header('Content-length', content_length)
952
# Send the multipart body
954
for (start, end) in ranges:
955
self.wfile.write(boundary_line)
956
self.send_header('Content-type', 'application/octet-stream')
957
self.send_header('Content-Range', 'bytes %d-%d/%d'
958
% (start, end, file_size))
960
if cur + self._truncated_ranges >= len(ranges):
961
# Abruptly ends the response and close the connection
962
self.close_connection = 1
964
self.send_range_content(file, start, end - start + 1)
967
self.wfile.write(boundary_line)
970
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
972
_req_handler_class = TruncatedMultipleRangeRequestHandler
975
super(TestTruncatedMultipleRangeServer, self).setUp()
976
self.build_tree_contents([('a', '0123456789')],)
978
def test_readv_with_short_reads(self):
979
server = self.get_readonly_server()
980
t = self.get_readonly_transport()
981
# Force separate ranges for each offset
982
t._bytes_to_read_before_seek = 0
983
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
984
self.assertEqual((0, '0'), ireadv.next())
985
self.assertEqual((2, '2'), ireadv.next())
986
if not self._testing_pycurl():
987
# Only one request have been issued so far (except for pycurl that
988
# try to read the whole response at once)
989
self.assertEqual(1, server.GET_request_nb)
990
self.assertEqual((4, '45'), ireadv.next())
991
self.assertEqual((9, '9'), ireadv.next())
992
# Both implementations issue 3 requests but:
993
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
995
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
996
self.assertEqual(3, server.GET_request_nb)
997
# Finally the client have tried a single range request and stays in
999
self.assertEqual('single', t._range_hint)
1002
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1003
"""Errors out when range specifiers exceed the limit"""
1005
def get_multiple_ranges(self, file, file_size, ranges):
1006
"""Refuses the multiple ranges request"""
1007
tcs = self.server.test_case_server
1008
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1010
# Emulate apache behavior
1011
self.send_error(400, "Bad Request")
1013
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1014
self, file, file_size, ranges)
1017
class LimitedRangeHTTPServer(http_server.HttpServer):
1018
"""An HttpServer erroring out on requests with too much range specifiers"""
1020
def __init__(self, request_handler=LimitedRangeRequestHandler,
1021
protocol_version=None,
1023
http_server.HttpServer.__init__(self, request_handler,
1024
protocol_version=protocol_version)
1025
self.range_limit = range_limit
1028
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1029
"""Tests readv requests against a server erroring out on too much ranges."""
1031
scenarios = multiply_scenarios(
1032
vary_by_http_client_implementation(),
1033
vary_by_http_protocol_version(),
1036
# Requests with more range specifiers will error out
1039
def create_transport_readonly_server(self):
1040
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1041
protocol_version=self._protocol_version)
1044
http_utils.TestCaseWithWebserver.setUp(self)
1045
# We need to manipulate ranges that correspond to real chunks in the
1046
# response, so we build a content appropriately.
1047
filler = ''.join(['abcdefghij' for x in range(102)])
1048
content = ''.join(['%04d' % v + filler for v in range(16)])
1049
self.build_tree_contents([('a', content)],)
1051
def test_few_ranges(self):
1052
t = self.get_readonly_transport()
1053
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1054
self.assertEqual(l[0], (0, '0000'))
1055
self.assertEqual(l[1], (1024, '0001'))
1056
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1058
def test_more_ranges(self):
1059
t = self.get_readonly_transport()
1060
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1061
self.assertEqual(l[0], (0, '0000'))
1062
self.assertEqual(l[1], (1024, '0001'))
1063
self.assertEqual(l[2], (4096, '0004'))
1064
self.assertEqual(l[3], (8192, '0008'))
1065
# The server will refuse to serve the first request (too much ranges),
1066
# a second request will succeed.
1067
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1070
class TestHttpProxyWhiteBox(tests.TestCase):
1071
"""Whitebox test proxy http authorization.
1073
Only the urllib implementation is tested here.
1077
tests.TestCase.setUp(self)
1079
self.addCleanup(self._restore_env)
1081
def _install_env(self, env):
1082
for name, value in env.iteritems():
1083
self._old_env[name] = osutils.set_or_unset_env(name, value)
1085
def _restore_env(self):
1086
for name, value in self._old_env.iteritems():
1087
osutils.set_or_unset_env(name, value)
1089
def _proxied_request(self):
1090
handler = _urllib2_wrappers.ProxyHandler()
1091
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
1092
handler.set_proxy(request, 'http')
1095
def test_empty_user(self):
1096
self._install_env({'http_proxy': 'http://bar.com'})
1097
request = self._proxied_request()
1098
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1100
def test_invalid_proxy(self):
1101
"""A proxy env variable without scheme"""
1102
self._install_env({'http_proxy': 'host:1234'})
1103
self.assertRaises(errors.InvalidURL, self._proxied_request)
1106
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1107
"""Tests proxy server.
1109
Be aware that we do not setup a real proxy here. Instead, we
1110
check that the *connection* goes through the proxy by serving
1111
different content (the faked proxy server append '-proxied'
1115
scenarios = multiply_scenarios(
1116
vary_by_http_client_implementation(),
1117
vary_by_http_protocol_version(),
1120
# FIXME: We don't have an https server available, so we don't
1121
# test https connections. --vila toolongago
1124
super(TestProxyHttpServer, self).setUp()
1125
self.transport_secondary_server = http_utils.ProxyServer
1126
self.build_tree_contents([('foo', 'contents of foo\n'),
1127
('foo-proxied', 'proxied contents of foo\n')])
1128
# Let's setup some attributes for tests
1129
server = self.get_readonly_server()
1130
self.server_host_port = '%s:%d' % (server.host, server.port)
1131
if self._testing_pycurl():
1132
# Oh my ! pycurl does not check for the port as part of
1133
# no_proxy :-( So we just test the host part
1134
self.no_proxy_host = server.host
1136
self.no_proxy_host = self.server_host_port
1137
# The secondary server is the proxy
1138
self.proxy_url = self.get_secondary_url()
1141
def _testing_pycurl(self):
1142
# TODO: This is duplicated for lots of the classes in this file
1143
return (features.pycurl.available()
1144
and self._transport == PyCurlTransport)
1146
def _install_env(self, env):
1147
for name, value in env.iteritems():
1148
self._old_env[name] = osutils.set_or_unset_env(name, value)
1150
def _restore_env(self):
1151
for name, value in self._old_env.iteritems():
1152
osutils.set_or_unset_env(name, value)
1154
def proxied_in_env(self, env):
1155
self._install_env(env)
1156
t = self.get_readonly_transport()
1158
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1162
def not_proxied_in_env(self, env):
1163
self._install_env(env)
1164
t = self.get_readonly_transport()
1166
self.assertEqual('contents of foo\n', t.get('foo').read())
1170
def test_http_proxy(self):
1171
self.proxied_in_env({'http_proxy': self.proxy_url})
1173
def test_HTTP_PROXY(self):
1174
if self._testing_pycurl():
1175
# pycurl does not check HTTP_PROXY for security reasons
1176
# (for use in a CGI context that we do not care
1177
# about. Should we ?)
1178
raise tests.TestNotApplicable(
1179
'pycurl does not check HTTP_PROXY for security reasons')
1180
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
1182
def test_all_proxy(self):
1183
self.proxied_in_env({'all_proxy': self.proxy_url})
1185
def test_ALL_PROXY(self):
1186
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1188
def test_http_proxy_with_no_proxy(self):
1189
self.not_proxied_in_env({'http_proxy': self.proxy_url,
1190
'no_proxy': self.no_proxy_host})
1192
def test_HTTP_PROXY_with_NO_PROXY(self):
1193
if self._testing_pycurl():
1194
raise tests.TestNotApplicable(
1195
'pycurl does not check HTTP_PROXY for security reasons')
1196
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
1197
'NO_PROXY': self.no_proxy_host})
1199
def test_all_proxy_with_no_proxy(self):
1200
self.not_proxied_in_env({'all_proxy': self.proxy_url,
1201
'no_proxy': self.no_proxy_host})
1203
def test_ALL_PROXY_with_NO_PROXY(self):
1204
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
1205
'NO_PROXY': self.no_proxy_host})
1207
def test_http_proxy_without_scheme(self):
1208
if self._testing_pycurl():
1209
# pycurl *ignores* invalid proxy env variables. If that ever change
1210
# in the future, this test will fail indicating that pycurl do not
1211
# ignore anymore such variables.
1212
self.not_proxied_in_env({'http_proxy': self.server_host_port})
1214
self.assertRaises(errors.InvalidURL,
1215
self.proxied_in_env,
1216
{'http_proxy': self.server_host_port})
1219
class TestRanges(http_utils.TestCaseWithWebserver):
1220
"""Test the Range header in GET methods."""
1222
scenarios = multiply_scenarios(
1223
vary_by_http_client_implementation(),
1224
vary_by_http_protocol_version(),
1228
http_utils.TestCaseWithWebserver.setUp(self)
1229
self.build_tree_contents([('a', '0123456789')],)
1231
def create_transport_readonly_server(self):
1232
return http_server.HttpServer(protocol_version=self._protocol_version)
1234
def _file_contents(self, relpath, ranges):
1235
t = self.get_readonly_transport()
1236
offsets = [ (start, end - start + 1) for start, end in ranges]
1237
coalesce = t._coalesce_offsets
1238
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1239
code, data = t._get(relpath, coalesced)
1240
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1241
for start, end in ranges:
1243
yield data.read(end - start + 1)
1245
def _file_tail(self, relpath, tail_amount):
1246
t = self.get_readonly_transport()
1247
code, data = t._get(relpath, [], tail_amount)
1248
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1249
data.seek(-tail_amount, 2)
1250
return data.read(tail_amount)
1252
def test_range_header(self):
1254
map(self.assertEqual,['0', '234'],
1255
list(self._file_contents('a', [(0,0), (2,4)])),)
1257
def test_range_header_tail(self):
1258
self.assertEqual('789', self._file_tail('a', 3))
1260
def test_syntactically_invalid_range_header(self):
1261
self.assertListRaises(errors.InvalidHttpRange,
1262
self._file_contents, 'a', [(4, 3)])
1264
def test_semantically_invalid_range_header(self):
1265
self.assertListRaises(errors.InvalidHttpRange,
1266
self._file_contents, 'a', [(42, 128)])
1269
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1270
"""Test redirection between http servers."""
1272
scenarios = multiply_scenarios(
1273
vary_by_http_client_implementation(),
1274
vary_by_http_protocol_version(),
1278
super(TestHTTPRedirections, self).setUp()
1279
self.build_tree_contents([('a', '0123456789'),
1281
'# Bazaar revision bundle v0.9\n#\n')
1284
def test_redirected(self):
1285
self.assertRaises(errors.RedirectRequested,
1286
self.get_old_transport().get, 'a')
1287
self.assertEqual('0123456789', self.get_new_transport().get('a').read())
1290
class RedirectedRequest(_urllib2_wrappers.Request):
1291
"""Request following redirections. """
1293
init_orig = _urllib2_wrappers.Request.__init__
1295
def __init__(self, method, url, *args, **kwargs):
1299
# Since the tests using this class will replace
1300
# _urllib2_wrappers.Request, we can't just call the base class __init__
1302
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1303
self.follow_redirections = True
1306
def install_redirected_request(test):
1307
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1310
def cleanup_http_redirection_connections(test):
1311
# Some sockets are opened but never seen by _urllib, so we trap them at
1312
# the _urllib2_wrappers level to be able to clean them up.
1313
def socket_disconnect(sock):
1315
sock.shutdown(socket.SHUT_RDWR)
1317
except socket.error:
1319
def connect(connection):
1320
test.http_connect_orig(connection)
1321
test.addCleanup(socket_disconnect, connection.sock)
1322
test.http_connect_orig = test.overrideAttr(
1323
_urllib2_wrappers.HTTPConnection, 'connect', connect)
1324
def connect(connection):
1325
test.https_connect_orig(connection)
1326
test.addCleanup(socket_disconnect, connection.sock)
1327
test.https_connect_orig = test.overrideAttr(
1328
_urllib2_wrappers.HTTPSConnection, 'connect', connect)
1331
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1332
"""Test redirections.
1334
http implementations do not redirect silently anymore (they
1335
do not redirect at all in fact). The mechanism is still in
1336
place at the _urllib2_wrappers.Request level and these tests
1339
For the pycurl implementation
1340
the redirection have been deleted as we may deprecate pycurl
1341
and I have no place to keep a working implementation.
1345
scenarios = multiply_scenarios(
1346
vary_by_http_client_implementation(),
1347
vary_by_http_protocol_version(),
1351
if (features.pycurl.available()
1352
and self._transport == PyCurlTransport):
1353
raise tests.TestNotApplicable(
1354
"pycurl doesn't redirect silently anymore")
1355
super(TestHTTPSilentRedirections, self).setUp()
1356
install_redirected_request(self)
1357
cleanup_http_redirection_connections(self)
1358
self.build_tree_contents([('a','a'),
1360
('1/a', 'redirected once'),
1362
('2/a', 'redirected twice'),
1364
('3/a', 'redirected thrice'),
1366
('4/a', 'redirected 4 times'),
1368
('5/a', 'redirected 5 times'),
1371
def test_one_redirection(self):
1372
t = self.get_old_transport()
1373
req = RedirectedRequest('GET', t._remote_path('a'))
1374
new_prefix = 'http://%s:%s' % (self.new_server.host,
1375
self.new_server.port)
1376
self.old_server.redirections = \
1377
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1378
self.assertEqual('redirected once', t._perform(req).read())
1380
def test_five_redirections(self):
1381
t = self.get_old_transport()
1382
req = RedirectedRequest('GET', t._remote_path('a'))
1383
old_prefix = 'http://%s:%s' % (self.old_server.host,
1384
self.old_server.port)
1385
new_prefix = 'http://%s:%s' % (self.new_server.host,
1386
self.new_server.port)
1387
self.old_server.redirections = [
1388
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1389
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1390
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1391
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1392
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1394
self.assertEqual('redirected 5 times', t._perform(req).read())
1397
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1398
"""Test transport.do_catching_redirections."""
1400
scenarios = multiply_scenarios(
1401
vary_by_http_client_implementation(),
1402
vary_by_http_protocol_version(),
1406
super(TestDoCatchRedirections, self).setUp()
1407
self.build_tree_contents([('a', '0123456789'),],)
1408
cleanup_http_redirection_connections(self)
1410
self.old_transport = self.get_old_transport()
1415
def test_no_redirection(self):
1416
t = self.get_new_transport()
1418
# We use None for redirected so that we fail if redirected
1419
self.assertEqual('0123456789',
1420
transport.do_catching_redirections(
1421
self.get_a, t, None).read())
1423
def test_one_redirection(self):
1424
self.redirections = 0
1426
def redirected(t, exception, redirection_notice):
1427
self.redirections += 1
1428
redirected_t = t._redirected_to(exception.source, exception.target)
1431
self.assertEqual('0123456789',
1432
transport.do_catching_redirections(
1433
self.get_a, self.old_transport, redirected).read())
1434
self.assertEqual(1, self.redirections)
1436
def test_redirection_loop(self):
1438
def redirected(transport, exception, redirection_notice):
1439
# By using the redirected url as a base dir for the
1440
# *old* transport, we create a loop: a => a/a =>
1442
return self.old_transport.clone(exception.target)
1444
self.assertRaises(errors.TooManyRedirections,
1445
transport.do_catching_redirections,
1446
self.get_a, self.old_transport, redirected)
1449
class TestAuth(http_utils.TestCaseWithWebserver):
1450
"""Test authentication scheme"""
1452
scenarios = multiply_scenarios(
1453
vary_by_http_client_implementation(),
1454
vary_by_http_protocol_version(),
1455
vary_by_http_auth_scheme(),
1458
_auth_header = 'Authorization'
1459
_password_prompt_prefix = ''
1460
_username_prompt_prefix = ''
1465
super(TestAuth, self).setUp()
1466
self.server = self.get_readonly_server()
1467
self.build_tree_contents([('a', 'contents of a\n'),
1468
('b', 'contents of b\n'),])
1470
def create_transport_readonly_server(self):
1471
server = self._auth_server(protocol_version=self._protocol_version)
1472
server._url_protocol = self._url_protocol
1475
def _testing_pycurl(self):
1476
# TODO: This is duplicated for lots of the classes in this file
1477
return (features.pycurl.available()
1478
and self._transport == PyCurlTransport)
1480
def get_user_url(self, user, password):
1481
"""Build an url embedding user and password"""
1482
url = '%s://' % self.server._url_protocol
1483
if user is not None:
1485
if password is not None:
1486
url += ':' + password
1488
url += '%s:%s/' % (self.server.host, self.server.port)
1491
def get_user_transport(self, user, password):
1492
t = transport.get_transport(self.get_user_url(user, password))
1495
def test_no_user(self):
1496
self.server.add_user('joe', 'foo')
1497
t = self.get_user_transport(None, None)
1498
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1499
# Only one 'Authentication Required' error should occur
1500
self.assertEqual(1, self.server.auth_required_errors)
1502
def test_empty_pass(self):
1503
self.server.add_user('joe', '')
1504
t = self.get_user_transport('joe', '')
1505
self.assertEqual('contents of a\n', t.get('a').read())
1506
# Only one 'Authentication Required' error should occur
1507
self.assertEqual(1, self.server.auth_required_errors)
1509
def test_user_pass(self):
1510
self.server.add_user('joe', 'foo')
1511
t = self.get_user_transport('joe', 'foo')
1512
self.assertEqual('contents of a\n', t.get('a').read())
1513
# Only one 'Authentication Required' error should occur
1514
self.assertEqual(1, self.server.auth_required_errors)
1516
def test_unknown_user(self):
1517
self.server.add_user('joe', 'foo')
1518
t = self.get_user_transport('bill', 'foo')
1519
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1520
# Two 'Authentication Required' errors should occur (the
1521
# initial 'who are you' and 'I don't know you, who are
1523
self.assertEqual(2, self.server.auth_required_errors)
1525
def test_wrong_pass(self):
1526
self.server.add_user('joe', 'foo')
1527
t = self.get_user_transport('joe', 'bar')
1528
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1529
# Two 'Authentication Required' errors should occur (the
1530
# initial 'who are you' and 'this is not you, who are you')
1531
self.assertEqual(2, self.server.auth_required_errors)
1533
def test_prompt_for_username(self):
1534
if self._testing_pycurl():
1535
raise tests.TestNotApplicable(
1536
'pycurl cannot prompt, it handles auth by embedding'
1537
' user:pass in urls only')
1539
self.server.add_user('joe', 'foo')
1540
t = self.get_user_transport(None, None)
1541
stdout = tests.StringIOWrapper()
1542
stderr = tests.StringIOWrapper()
1543
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1544
stdout=stdout, stderr=stderr)
1545
self.assertEqual('contents of a\n',t.get('a').read())
1546
# stdin should be empty
1547
self.assertEqual('', ui.ui_factory.stdin.readline())
1549
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1550
self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1551
self.assertEqual('', stdout.getvalue())
1552
self._check_password_prompt(t._unqualified_scheme, 'joe',
1555
def test_prompt_for_password(self):
1556
if self._testing_pycurl():
1557
raise tests.TestNotApplicable(
1558
'pycurl cannot prompt, it handles auth by embedding'
1559
' user:pass in urls only')
1561
self.server.add_user('joe', 'foo')
1562
t = self.get_user_transport('joe', None)
1563
stdout = tests.StringIOWrapper()
1564
stderr = tests.StringIOWrapper()
1565
ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1566
stdout=stdout, stderr=stderr)
1567
self.assertEqual('contents of a\n', t.get('a').read())
1568
# stdin should be empty
1569
self.assertEqual('', ui.ui_factory.stdin.readline())
1570
self._check_password_prompt(t._unqualified_scheme, 'joe',
1572
self.assertEqual('', stdout.getvalue())
1573
# And we shouldn't prompt again for a different request
1574
# against the same transport.
1575
self.assertEqual('contents of b\n',t.get('b').read())
1577
# And neither against a clone
1578
self.assertEqual('contents of b\n',t2.get('b').read())
1579
# Only one 'Authentication Required' error should occur
1580
self.assertEqual(1, self.server.auth_required_errors)
1582
def _check_password_prompt(self, scheme, user, actual_prompt):
1583
expected_prompt = (self._password_prompt_prefix
1584
+ ("%s %s@%s:%d, Realm: '%s' password: "
1586
user, self.server.host, self.server.port,
1587
self.server.auth_realm)))
1588
self.assertEqual(expected_prompt, actual_prompt)
1590
def _expected_username_prompt(self, scheme):
1591
return (self._username_prompt_prefix
1592
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1593
self.server.host, self.server.port,
1594
self.server.auth_realm))
1596
def test_no_prompt_for_password_when_using_auth_config(self):
1597
if self._testing_pycurl():
1598
raise tests.TestNotApplicable(
1599
'pycurl does not support authentication.conf'
1600
' since it cannot prompt')
1604
stdin_content = 'bar\n' # Not the right password
1605
self.server.add_user(user, password)
1606
t = self.get_user_transport(user, None)
1607
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1608
stderr=tests.StringIOWrapper())
1609
# Create a minimal config file with the right password
1610
_setup_authentication_config(
1612
port=self.server.port,
1615
# Issue a request to the server to connect
1616
self.assertEqual('contents of a\n',t.get('a').read())
1617
# stdin should have been left untouched
1618
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1619
# Only one 'Authentication Required' error should occur
1620
self.assertEqual(1, self.server.auth_required_errors)
1622
def test_changing_nonce(self):
1623
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1624
http_utils.ProxyDigestAuthServer):
1625
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1626
if self._testing_pycurl():
1627
raise tests.KnownFailure(
1628
'pycurl does not handle a nonce change')
1629
self.server.add_user('joe', 'foo')
1630
t = self.get_user_transport('joe', 'foo')
1631
self.assertEqual('contents of a\n', t.get('a').read())
1632
self.assertEqual('contents of b\n', t.get('b').read())
1633
# Only one 'Authentication Required' error should have
1635
self.assertEqual(1, self.server.auth_required_errors)
1636
# The server invalidates the current nonce
1637
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1638
self.assertEqual('contents of a\n', t.get('a').read())
1639
# Two 'Authentication Required' errors should occur (the
1640
# initial 'who are you' and a second 'who are you' with the new nonce)
1641
self.assertEqual(2, self.server.auth_required_errors)
1643
def test_user_from_auth_conf(self):
1644
if self._testing_pycurl():
1645
raise tests.TestNotApplicable(
1646
'pycurl does not support authentication.conf')
1649
self.server.add_user(user, password)
1650
_setup_authentication_config(
1652
port=self.server.port,
1655
t = self.get_user_transport(None, None)
1656
# Issue a request to the server to connect
1657
self.assertEqual('contents of a\n', t.get('a').read())
1658
# Only one 'Authentication Required' error should occur
1659
self.assertEqual(1, self.server.auth_required_errors)
1662
def _setup_authentication_config(**kwargs):
1663
conf = config.AuthenticationConfig()
1664
conf._get_config().update({'httptest': kwargs})
1669
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1670
"""Unit tests for glue by which urllib2 asks us for authentication"""
1672
def test_get_user_password_without_port(self):
1673
"""We cope if urllib2 doesn't tell us the port.
1675
See https://bugs.launchpad.net/bzr/+bug/654684
1679
_setup_authentication_config(
1684
handler = _urllib2_wrappers.HTTPAuthHandler()
1685
got_pass = handler.get_user_password(dict(
1692
self.assertEquals((user, password), got_pass)
1695
class TestProxyAuth(TestAuth):
1696
"""Test proxy authentication schemes."""
1698
scenarios = multiply_scenarios(
1699
vary_by_http_client_implementation(),
1700
vary_by_http_protocol_version(),
1701
vary_by_http_proxy_auth_scheme(),
1704
_auth_header = 'Proxy-authorization'
1705
_password_prompt_prefix = 'Proxy '
1706
_username_prompt_prefix = 'Proxy '
1709
super(TestProxyAuth, self).setUp()
1711
self.addCleanup(self._restore_env)
1712
# Override the contents to avoid false positives
1713
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1714
('b', 'not proxied contents of b\n'),
1715
('a-proxied', 'contents of a\n'),
1716
('b-proxied', 'contents of b\n'),
1719
def get_user_transport(self, user, password):
1720
self._install_env({'all_proxy': self.get_user_url(user, password)})
1721
return TestAuth.get_user_transport(self, user, password)
1723
def _install_env(self, env):
1724
for name, value in env.iteritems():
1725
self._old_env[name] = osutils.set_or_unset_env(name, value)
1727
def _restore_env(self):
1728
for name, value in self._old_env.iteritems():
1729
osutils.set_or_unset_env(name, value)
1731
def test_empty_pass(self):
1732
if self._testing_pycurl():
1734
if pycurl.version_info()[1] < '7.16.0':
1735
raise tests.KnownFailure(
1736
'pycurl < 7.16.0 does not handle empty proxy passwords')
1737
super(TestProxyAuth, self).test_empty_pass()
1740
class SampleSocket(object):
1741
"""A socket-like object for use in testing the HTTP request handler."""
1743
def __init__(self, socket_read_content):
1744
"""Constructs a sample socket.
1746
:param socket_read_content: a byte sequence
1748
# Use plain python StringIO so we can monkey-patch the close method to
1749
# not discard the contents.
1750
from StringIO import StringIO
1751
self.readfile = StringIO(socket_read_content)
1752
self.writefile = StringIO()
1753
self.writefile.close = lambda: None
1754
self.close = lambda: None
1756
def makefile(self, mode='r', bufsize=None):
1758
return self.readfile
1760
return self.writefile
1763
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1765
scenarios = multiply_scenarios(
1766
vary_by_http_client_implementation(),
1767
vary_by_http_protocol_version(),
1771
super(SmartHTTPTunnellingTest, self).setUp()
1772
# We use the VFS layer as part of HTTP tunnelling tests.
1773
self._captureVar('BZR_NO_SMART_VFS', None)
1774
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1775
self.http_server = self.get_readonly_server()
1777
def create_transport_readonly_server(self):
1778
server = http_utils.HTTPServerWithSmarts(
1779
protocol_version=self._protocol_version)
1780
server._url_protocol = self._url_protocol
1783
def test_open_bzrdir(self):
1784
branch = self.make_branch('relpath')
1785
url = self.http_server.get_url() + 'relpath'
1786
bd = bzrdir.BzrDir.open(url)
1787
self.addCleanup(bd.transport.disconnect)
1788
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1790
def test_bulk_data(self):
1791
# We should be able to send and receive bulk data in a single message.
1792
# The 'readv' command in the smart protocol both sends and receives
1793
# bulk data, so we use that.
1794
self.build_tree(['data-file'])
1795
http_transport = transport.get_transport(self.http_server.get_url())
1796
medium = http_transport.get_smart_medium()
1797
# Since we provide the medium, the url below will be mostly ignored
1798
# during the test, as long as the path is '/'.
1799
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1802
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1804
def test_http_send_smart_request(self):
1806
post_body = 'hello\n'
1807
expected_reply_body = 'ok\x012\n'
1809
http_transport = transport.get_transport(self.http_server.get_url())
1810
medium = http_transport.get_smart_medium()
1811
response = medium.send_http_smart_request(post_body)
1812
reply_body = response.read()
1813
self.assertEqual(expected_reply_body, reply_body)
1815
def test_smart_http_server_post_request_handler(self):
1816
httpd = self.http_server.server
1818
socket = SampleSocket(
1819
'POST /.bzr/smart %s \r\n' % self._protocol_version
1820
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1822
+ 'Content-Length: 6\r\n'
1825
# Beware: the ('localhost', 80) below is the
1826
# client_address parameter, but we don't have one because
1827
# we have defined a socket which is not bound to an
1828
# address. The test framework never uses this client
1829
# address, so far...
1830
request_handler = http_utils.SmartRequestHandler(socket,
1833
response = socket.writefile.getvalue()
1834
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1835
# This includes the end of the HTTP headers, and all the body.
1836
expected_end_of_response = '\r\n\r\nok\x012\n'
1837
self.assertEndsWith(response, expected_end_of_response)
1840
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1841
"""No smart server here request handler."""
1844
self.send_error(403, "Forbidden")
1847
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1848
"""Test smart client behaviour against an http server without smarts."""
1850
_req_handler_class = ForbiddenRequestHandler
1852
def test_probe_smart_server(self):
1853
"""Test error handling against server refusing smart requests."""
1854
t = self.get_readonly_transport()
1855
# No need to build a valid smart request here, the server will not even
1856
# try to interpret it.
1857
self.assertRaises(errors.SmartProtocolError,
1858
t.get_smart_medium().send_http_smart_request,
1862
class Test_redirected_to(tests.TestCase):
1864
scenarios = vary_by_http_client_implementation()
1866
def test_redirected_to_subdir(self):
1867
t = self._transport('http://www.example.com/foo')
1868
r = t._redirected_to('http://www.example.com/foo',
1869
'http://www.example.com/foo/subdir')
1870
self.assertIsInstance(r, type(t))
1871
# Both transports share the some connection
1872
self.assertEqual(t._get_connection(), r._get_connection())
1874
def test_redirected_to_self_with_slash(self):
1875
t = self._transport('http://www.example.com/foo')
1876
r = t._redirected_to('http://www.example.com/foo',
1877
'http://www.example.com/foo/')
1878
self.assertIsInstance(r, type(t))
1879
# Both transports share the some connection (one can argue that we
1880
# should return the exact same transport here, but that seems
1882
self.assertEqual(t._get_connection(), r._get_connection())
1884
def test_redirected_to_host(self):
1885
t = self._transport('http://www.example.com/foo')
1886
r = t._redirected_to('http://www.example.com/foo',
1887
'http://foo.example.com/foo/subdir')
1888
self.assertIsInstance(r, type(t))
1890
def test_redirected_to_same_host_sibling_protocol(self):
1891
t = self._transport('http://www.example.com/foo')
1892
r = t._redirected_to('http://www.example.com/foo',
1893
'https://www.example.com/foo')
1894
self.assertIsInstance(r, type(t))
1896
def test_redirected_to_same_host_different_protocol(self):
1897
t = self._transport('http://www.example.com/foo')
1898
r = t._redirected_to('http://www.example.com/foo',
1899
'ftp://www.example.com/foo')
1900
self.assertNotEquals(type(r), type(t))
1902
def test_redirected_to_different_host_same_user(self):
1903
t = self._transport('http://joe@www.example.com/foo')
1904
r = t._redirected_to('http://www.example.com/foo',
1905
'https://foo.example.com/foo')
1906
self.assertIsInstance(r, type(t))
1907
self.assertEqual(t._user, r._user)
1910
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1911
"""Request handler for a unique and pre-defined request.
1913
The only thing we care about here is how many bytes travel on the wire. But
1914
since we want to measure it for a real http client, we have to send it
1917
We expect to receive a *single* request nothing more (and we won't even
1918
check what request it is, we just measure the bytes read until an empty
1922
def _handle_one_request(self):
1923
tcs = self.server.test_case_server
1924
requestline = self.rfile.readline()
1925
headers = self.MessageClass(self.rfile, 0)
1926
# We just read: the request, the headers, an empty line indicating the
1927
# end of the headers.
1928
bytes_read = len(requestline)
1929
for line in headers.headers:
1930
bytes_read += len(line)
1931
bytes_read += len('\r\n')
1932
if requestline.startswith('POST'):
1933
# The body should be a single line (or we don't know where it ends
1934
# and we don't want to issue a blocking read)
1935
body = self.rfile.readline()
1936
bytes_read += len(body)
1937
tcs.bytes_read = bytes_read
1939
# We set the bytes written *before* issuing the write, the client is
1940
# supposed to consume every produced byte *before* checking that value.
1942
# Doing the oppposite may lead to test failure: we may be interrupted
1943
# after the write but before updating the value. The client can then
1944
# continue and read the value *before* we can update it. And yes,
1945
# this has been observed -- vila 20090129
1946
tcs.bytes_written = len(tcs.canned_response)
1947
self.wfile.write(tcs.canned_response)
1950
class ActivityServerMixin(object):
1952
def __init__(self, protocol_version):
1953
super(ActivityServerMixin, self).__init__(
1954
request_handler=PredefinedRequestHandler,
1955
protocol_version=protocol_version)
1956
# Bytes read and written by the server
1958
self.bytes_written = 0
1959
self.canned_response = None
1962
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1966
if tests.HTTPSServerFeature.available():
1967
from bzrlib.tests import https_server
1968
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1972
class TestActivityMixin(object):
1973
"""Test socket activity reporting.
1975
We use a special purpose server to control the bytes sent and received and
1976
be able to predict the activity on the client socket.
1980
tests.TestCase.setUp(self)
1981
self.server = self._activity_server(self._protocol_version)
1982
self.server.start_server()
1983
self.activities = {}
1984
def report_activity(t, bytes, direction):
1985
count = self.activities.get(direction, 0)
1987
self.activities[direction] = count
1989
# We override at class level because constructors may propagate the
1990
# bound method and render instance overriding ineffective (an
1991
# alternative would be to define a specific ui factory instead...)
1992
self.overrideAttr(self._transport, '_report_activity', report_activity)
1993
self.addCleanup(self.server.stop_server)
1995
def get_transport(self):
1996
t = self._transport(self.server.get_url())
1997
# FIXME: Needs cleanup -- vila 20100611
2000
def assertActivitiesMatch(self):
2001
self.assertEqual(self.server.bytes_read,
2002
self.activities.get('write', 0), 'written bytes')
2003
self.assertEqual(self.server.bytes_written,
2004
self.activities.get('read', 0), 'read bytes')
2007
self.server.canned_response = '''HTTP/1.1 200 OK\r
2008
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2009
Server: Apache/2.0.54 (Fedora)\r
2010
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2011
ETag: "56691-23-38e9ae00"\r
2012
Accept-Ranges: bytes\r
2013
Content-Length: 35\r
2015
Content-Type: text/plain; charset=UTF-8\r
2017
Bazaar-NG meta directory, format 1
2019
t = self.get_transport()
2020
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2021
t.get('foo/bar').read())
2022
self.assertActivitiesMatch()
2025
self.server.canned_response = '''HTTP/1.1 200 OK\r
2026
Server: SimpleHTTP/0.6 Python/2.5.2\r
2027
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2028
Content-type: application/octet-stream\r
2029
Content-Length: 20\r
2030
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2033
t = self.get_transport()
2034
self.assertTrue(t.has('foo/bar'))
2035
self.assertActivitiesMatch()
2037
def test_readv(self):
2038
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2039
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2040
Server: Apache/2.0.54 (Fedora)\r
2041
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2042
ETag: "238a3c-16ec2-805c5540"\r
2043
Accept-Ranges: bytes\r
2044
Content-Length: 1534\r
2046
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2049
--418470f848b63279b\r
2050
Content-type: text/plain; charset=UTF-8\r
2051
Content-range: bytes 0-254/93890\r
2053
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2054
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2055
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2056
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2057
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2059
--418470f848b63279b\r
2060
Content-type: text/plain; charset=UTF-8\r
2061
Content-range: bytes 1000-2049/93890\r
2064
mbp@sourcefrog.net-20050311063625-07858525021f270b
2065
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2066
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2067
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2068
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2069
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2070
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2071
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2072
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2073
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2074
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2075
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2076
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2077
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2078
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2079
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2080
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2081
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2082
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2083
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2085
--418470f848b63279b--\r
2087
t = self.get_transport()
2088
# Remember that the request is ignored and that the ranges below
2089
# doesn't have to match the canned response.
2090
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2091
self.assertEqual(2, len(l))
2092
self.assertActivitiesMatch()
2094
def test_post(self):
2095
self.server.canned_response = '''HTTP/1.1 200 OK\r
2096
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2097
Server: Apache/2.0.54 (Fedora)\r
2098
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2099
ETag: "56691-23-38e9ae00"\r
2100
Accept-Ranges: bytes\r
2101
Content-Length: 35\r
2103
Content-Type: text/plain; charset=UTF-8\r
2105
lalala whatever as long as itsssss
2107
t = self.get_transport()
2108
# We must send a single line of body bytes, see
2109
# PredefinedRequestHandler._handle_one_request
2110
code, f = t._post('abc def end-of-body\n')
2111
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2112
self.assertActivitiesMatch()
2115
class TestActivity(tests.TestCase, TestActivityMixin):
2117
scenarios = multiply_scenarios(
2118
vary_by_http_activity(),
2119
vary_by_http_protocol_version(),
2123
TestActivityMixin.setUp(self)
2126
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2128
# Unlike TestActivity, we are really testing ReportingFileSocket and
2129
# ReportingSocket, so we don't need all the parametrization. Since
2130
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2131
# test them through their use by the transport than directly (that's a
2132
# bit less clean but far more simpler and effective).
2133
_activity_server = ActivityHTTPServer
2134
_protocol_version = 'HTTP/1.1'
2137
self._transport =_urllib.HttpTransport_urllib
2138
TestActivityMixin.setUp(self)
2140
def assertActivitiesMatch(self):
2141
# Nothing to check here
2145
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2146
"""Test authentication on the redirected http server."""
2148
scenarios = vary_by_http_protocol_version()
2150
_auth_header = 'Authorization'
2151
_password_prompt_prefix = ''
2152
_username_prompt_prefix = ''
2153
_auth_server = http_utils.HTTPBasicAuthServer
2154
_transport = _urllib.HttpTransport_urllib
2157
super(TestAuthOnRedirected, self).setUp()
2158
self.build_tree_contents([('a','a'),
2160
('1/a', 'redirected once'),
2162
new_prefix = 'http://%s:%s' % (self.new_server.host,
2163
self.new_server.port)
2164
self.old_server.redirections = [
2165
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2166
self.old_transport = self.get_old_transport()
2167
self.new_server.add_user('joe', 'foo')
2168
cleanup_http_redirection_connections(self)
2170
def create_transport_readonly_server(self):
2171
server = self._auth_server(protocol_version=self._protocol_version)
2172
server._url_protocol = self._url_protocol
2178
def test_auth_on_redirected_via_do_catching_redirections(self):
2179
self.redirections = 0
2181
def redirected(t, exception, redirection_notice):
2182
self.redirections += 1
2183
redirected_t = t._redirected_to(exception.source, exception.target)
2184
self.addCleanup(redirected_t.disconnect)
2187
stdout = tests.StringIOWrapper()
2188
stderr = tests.StringIOWrapper()
2189
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2190
stdout=stdout, stderr=stderr)
2191
self.assertEqual('redirected once',
2192
transport.do_catching_redirections(
2193
self.get_a, self.old_transport, redirected).read())
2194
self.assertEqual(1, self.redirections)
2195
# stdin should be empty
2196
self.assertEqual('', ui.ui_factory.stdin.readline())
2197
# stdout should be empty, stderr will contains the prompts
2198
self.assertEqual('', stdout.getvalue())
2200
def test_auth_on_redirected_via_following_redirections(self):
2201
self.new_server.add_user('joe', 'foo')
2202
stdout = tests.StringIOWrapper()
2203
stderr = tests.StringIOWrapper()
2204
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2205
stdout=stdout, stderr=stderr)
2206
t = self.old_transport
2207
req = RedirectedRequest('GET', t.abspath('a'))
2208
new_prefix = 'http://%s:%s' % (self.new_server.host,
2209
self.new_server.port)
2210
self.old_server.redirections = [
2211
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2212
self.assertEqual('redirected once', t._perform(req).read())
2213
# stdin should be empty
2214
self.assertEqual('', ui.ui_factory.stdin.readline())
2215
# stdout should be empty, stderr will contains the prompts
2216
self.assertEqual('', stdout.getvalue())
156
import bzrlib.transport.http._urllib
157
from bzrlib.transport import get_transport
158
# urlllib should always be present
159
t = get_transport('http+urllib://bzr.google.com/')
160
self.assertIsInstance(t, Transport)
161
self.assertIsInstance(t, bzrlib.transport.http._urllib.HttpTransport_urllib)
added
removed
bzrlib/tests/test_http.py
