1
# Copyright (C) 2005-2012, 2015, 2016, 2017 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
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.
23
# TODO: Should be renamed to breezy.transport.http.tests?
24
# TODO: What about renaming to breezy.tests.transport.http ?
28
import SimpleHTTPServer
40
remote as _mod_remote,
52
from .scenarios import (
53
load_tests_apply_scenarios,
56
from ..transport import (
60
from ..transport.http import (
66
load_tests = load_tests_apply_scenarios
69
def vary_by_http_client_implementation():
70
"""Test the libraries we can use, currently just urllib."""
71
transport_scenarios = [
72
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
73
_server=http_server.HttpServer_urllib,
74
_url_protocol='http+urllib',)),
76
return transport_scenarios
79
def vary_by_http_protocol_version():
80
"""Test on http/1.0 and 1.1"""
82
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
83
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
87
def vary_by_http_auth_scheme():
89
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
90
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
92
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
94
# Add some attributes common to all scenarios
95
for scenario_id, scenario_dict in scenarios:
96
scenario_dict.update(_auth_header='Authorization',
97
_username_prompt_prefix='',
98
_password_prompt_prefix='')
102
def vary_by_http_proxy_auth_scheme():
104
('proxy-basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
105
('proxy-digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
106
('proxy-basicdigest',
107
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
109
# Add some attributes common to all scenarios
110
for scenario_id, scenario_dict in scenarios:
111
scenario_dict.update(_auth_header='Proxy-Authorization',
112
_username_prompt_prefix='Proxy ',
113
_password_prompt_prefix='Proxy ')
117
def vary_by_http_activity():
118
activity_scenarios = [
119
('urllib,http', dict(_activity_server=ActivityHTTPServer,
120
_transport=_urllib.HttpTransport_urllib,)),
122
if features.HTTPSServerFeature.available():
123
# FIXME: Until we have a better way to handle self-signed certificates
124
# (like allowing them in a test specific authentication.conf for
125
# example), we need some specialized urllib transport for tests.
130
class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
132
def __init__(self, base, _from_transport=None):
133
super(HTTPS_urllib_transport, self).__init__(
134
base, _from_transport=_from_transport,
135
ca_certs=ssl_certs.build_path('ca.crt'))
137
activity_scenarios.append(
138
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
139
_transport=HTTPS_urllib_transport,)),)
140
return activity_scenarios
143
class FakeManager(object):
146
self.credentials = []
148
def add_password(self, realm, host, username, password):
149
self.credentials.append([realm, host, username, password])
152
class RecordingServer(object):
153
"""A fake HTTP server.
155
It records the bytes sent to it, and replies with a 200.
158
def __init__(self, expect_body_tail=None, scheme=''):
161
:type expect_body_tail: str
162
:param expect_body_tail: a reply won't be sent until this string is
165
self._expect_body_tail = expect_body_tail
168
self.received_bytes = ''
172
return '%s://%s:%s/' % (self.scheme, self.host, self.port)
174
def start_server(self):
175
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
176
self._sock.bind(('127.0.0.1', 0))
177
self.host, self.port = self._sock.getsockname()
178
self._ready = threading.Event()
179
self._thread = test_server.TestThread(
180
sync_event=self._ready, target=self._accept_read_and_reply)
182
if 'threads' in tests.selftest_debug_flags:
183
sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
186
def _accept_read_and_reply(self):
189
conn, address = self._sock.accept()
190
if self._expect_body_tail is not None:
191
while not self.received_bytes.endswith(self._expect_body_tail):
192
self.received_bytes += conn.recv(4096)
193
conn.sendall('HTTP/1.1 200 OK\r\n')
197
# The client may have already closed the socket.
200
def stop_server(self):
202
# Issue a fake connection to wake up the server and allow it to
204
fake_conn = osutils.connect_socket((self.host, self.port))
207
# We might have already closed it. We don't care.
212
if 'threads' in tests.selftest_debug_flags:
213
sys.stderr.write('Thread joined: %s\n' % (self._thread.ident,))
216
class TestAuthHeader(tests.TestCase):
218
def parse_header(self, header, auth_handler_class=None):
219
if auth_handler_class is None:
220
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
221
self.auth_handler = auth_handler_class()
222
return self.auth_handler._parse_auth_header(header)
224
def test_empty_header(self):
225
scheme, remainder = self.parse_header('')
226
self.assertEqual('', scheme)
227
self.assertIs(None, remainder)
229
def test_negotiate_header(self):
230
scheme, remainder = self.parse_header('Negotiate')
231
self.assertEqual('negotiate', scheme)
232
self.assertIs(None, remainder)
234
def test_basic_header(self):
235
scheme, remainder = self.parse_header(
236
'Basic realm="Thou should not pass"')
237
self.assertEqual('basic', scheme)
238
self.assertEqual('realm="Thou should not pass"', remainder)
240
def test_build_basic_header_with_long_creds(self):
241
handler = _urllib2_wrappers.BasicAuthHandler()
242
user = 'user' * 10 # length 40
243
password = 'password' * 5 # length 40
244
header = handler.build_auth_header(
245
dict(user=user, password=password), None)
246
# https://bugs.launchpad.net/bzr/+bug/1606203 was caused by incorrectly
247
# creating a header value with an embedded '\n'
248
self.assertFalse('\n' in header)
250
def test_basic_extract_realm(self):
251
scheme, remainder = self.parse_header(
252
'Basic realm="Thou should not pass"',
253
_urllib2_wrappers.BasicAuthHandler)
254
match, realm = self.auth_handler.extract_realm(remainder)
255
self.assertTrue(match is not None)
256
self.assertEqual('Thou should not pass', realm)
258
def test_digest_header(self):
259
scheme, remainder = self.parse_header(
260
'Digest realm="Thou should not pass"')
261
self.assertEqual('digest', scheme)
262
self.assertEqual('realm="Thou should not pass"', remainder)
265
class TestHTTPRangeParsing(tests.TestCase):
268
super(TestHTTPRangeParsing, self).setUp()
269
# We focus on range parsing here and ignore everything else
270
class RequestHandler(http_server.TestingHTTPRequestHandler):
271
def setup(self): pass
272
def handle(self): pass
273
def finish(self): pass
275
self.req_handler = RequestHandler(None, None, None)
277
def assertRanges(self, ranges, header, file_size):
278
self.assertEqual(ranges,
279
self.req_handler._parse_ranges(header, file_size))
281
def test_simple_range(self):
282
self.assertRanges([(0,2)], 'bytes=0-2', 12)
285
self.assertRanges([(8, 11)], 'bytes=-4', 12)
287
def test_tail_bigger_than_file(self):
288
self.assertRanges([(0, 11)], 'bytes=-99', 12)
290
def test_range_without_end(self):
291
self.assertRanges([(4, 11)], 'bytes=4-', 12)
293
def test_invalid_ranges(self):
294
self.assertRanges(None, 'bytes=12-22', 12)
295
self.assertRanges(None, 'bytes=1-3,12-22', 12)
296
self.assertRanges(None, 'bytes=-', 12)
299
class TestHTTPServer(tests.TestCase):
300
"""Test the HTTP servers implementations."""
302
def test_invalid_protocol(self):
303
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
305
protocol_version = 'HTTP/0.1'
307
self.assertRaises(httplib.UnknownProtocol,
308
http_server.HttpServer, BogusRequestHandler)
310
def test_force_invalid_protocol(self):
311
self.assertRaises(httplib.UnknownProtocol,
312
http_server.HttpServer, protocol_version='HTTP/0.1')
314
def test_server_start_and_stop(self):
315
server = http_server.HttpServer()
316
self.addCleanup(server.stop_server)
317
server.start_server()
318
self.assertTrue(server.server is not None)
319
self.assertTrue(server.server.serving is not None)
320
self.assertTrue(server.server.serving)
322
def test_create_http_server_one_zero(self):
323
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
325
protocol_version = 'HTTP/1.0'
327
server = http_server.HttpServer(RequestHandlerOneZero)
328
self.start_server(server)
329
self.assertIsInstance(server.server, http_server.TestingHTTPServer)
331
def test_create_http_server_one_one(self):
332
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
334
protocol_version = 'HTTP/1.1'
336
server = http_server.HttpServer(RequestHandlerOneOne)
337
self.start_server(server)
338
self.assertIsInstance(server.server,
339
http_server.TestingThreadingHTTPServer)
341
def test_create_http_server_force_one_one(self):
342
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
344
protocol_version = 'HTTP/1.0'
346
server = http_server.HttpServer(RequestHandlerOneZero,
347
protocol_version='HTTP/1.1')
348
self.start_server(server)
349
self.assertIsInstance(server.server,
350
http_server.TestingThreadingHTTPServer)
352
def test_create_http_server_force_one_zero(self):
353
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
355
protocol_version = 'HTTP/1.1'
357
server = http_server.HttpServer(RequestHandlerOneOne,
358
protocol_version='HTTP/1.0')
359
self.start_server(server)
360
self.assertIsInstance(server.server,
361
http_server.TestingHTTPServer)
364
class TestHttpTransportUrls(tests.TestCase):
365
"""Test the http urls."""
367
scenarios = vary_by_http_client_implementation()
369
def test_abs_url(self):
370
"""Construction of absolute http URLs"""
371
t = self._transport('http://example.com/bzr/bzr.dev/')
372
eq = self.assertEqualDiff
373
eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
374
eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
375
eq(t.abspath('.bzr'), 'http://example.com/bzr/bzr.dev/.bzr')
376
eq(t.abspath('.bzr/1//2/./3'),
377
'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
379
def test_invalid_http_urls(self):
380
"""Trap invalid construction of urls"""
381
self._transport('http://example.com/bzr/bzr.dev/')
382
self.assertRaises(errors.InvalidURL,
384
'http://http://example.com/bzr/bzr.dev/')
386
def test_http_root_urls(self):
387
"""Construction of URLs from server root"""
388
t = self._transport('http://example.com/')
389
eq = self.assertEqualDiff
390
eq(t.abspath('.bzr/tree-version'),
391
'http://example.com/.bzr/tree-version')
393
def test_http_impl_urls(self):
394
"""There are servers which ask for particular clients to connect"""
395
server = self._server()
396
server.start_server()
398
url = server.get_url()
399
self.assertTrue(url.startswith('%s://' % self._url_protocol))
404
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
405
"""Test the http connections."""
407
scenarios = multiply_scenarios(
408
vary_by_http_client_implementation(),
409
vary_by_http_protocol_version(),
413
super(TestHTTPConnections, self).setUp()
414
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
415
transport=self.get_transport())
417
def test_http_has(self):
418
server = self.get_readonly_server()
419
t = self.get_readonly_transport()
420
self.assertEqual(t.has('foo/bar'), True)
421
self.assertEqual(len(server.logs), 1)
422
self.assertContainsRe(server.logs[0],
423
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
425
def test_http_has_not_found(self):
426
server = self.get_readonly_server()
427
t = self.get_readonly_transport()
428
self.assertEqual(t.has('not-found'), False)
429
self.assertContainsRe(server.logs[1],
430
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
432
def test_http_get(self):
433
server = self.get_readonly_server()
434
t = self.get_readonly_transport()
435
fp = t.get('foo/bar')
436
self.assertEqualDiff(
438
'contents of foo/bar\n')
439
self.assertEqual(len(server.logs), 1)
440
self.assertTrue(server.logs[0].find(
441
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
442
% breezy.__version__) > -1)
444
def test_has_on_bogus_host(self):
445
# Get a free address and don't 'accept' on it, so that we
446
# can be sure there is no http handler there, but set a
447
# reasonable timeout to not slow down tests too much.
448
default_timeout = socket.getdefaulttimeout()
450
socket.setdefaulttimeout(2)
452
s.bind(('localhost', 0))
453
t = self._transport('http://%s:%s/' % s.getsockname())
454
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
456
socket.setdefaulttimeout(default_timeout)
459
class TestHttpTransportRegistration(tests.TestCase):
460
"""Test registrations of various http implementations"""
462
scenarios = vary_by_http_client_implementation()
464
def test_http_registered(self):
465
t = transport.get_transport_from_url(
466
'%s://foo.com/' % self._url_protocol)
467
self.assertIsInstance(t, transport.Transport)
468
self.assertIsInstance(t, self._transport)
471
class TestPost(tests.TestCase):
473
scenarios = multiply_scenarios(
474
vary_by_http_client_implementation(),
475
vary_by_http_protocol_version(),
478
def test_post_body_is_received(self):
479
server = RecordingServer(expect_body_tail='end-of-body',
480
scheme=self._url_protocol)
481
self.start_server(server)
482
url = server.get_url()
483
# FIXME: needs a cleanup -- vila 20100611
484
http_transport = transport.get_transport_from_url(url)
485
code, response = http_transport._post('abc def end-of-body')
487
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
488
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
489
self.assertTrue('content-type: application/octet-stream\r'
490
in server.received_bytes.lower())
491
# The transport should not be assuming that the server can accept
492
# chunked encoding the first time it connects, because HTTP/1.1, so we
493
# check for the literal string.
495
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
498
class TestRangeHeader(tests.TestCase):
499
"""Test range_header method"""
501
def check_header(self, value, ranges=[], tail=0):
502
offsets = [ (start, end - start + 1) for start, end in ranges]
503
coalesce = transport.Transport._coalesce_offsets
504
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
505
range_header = http.HttpTransportBase._range_header
506
self.assertEqual(value, range_header(coalesced, tail))
508
def test_range_header_single(self):
509
self.check_header('0-9', ranges=[(0,9)])
510
self.check_header('100-109', ranges=[(100,109)])
512
def test_range_header_tail(self):
513
self.check_header('-10', tail=10)
514
self.check_header('-50', tail=50)
516
def test_range_header_multi(self):
517
self.check_header('0-9,100-200,300-5000',
518
ranges=[(0,9), (100, 200), (300,5000)])
520
def test_range_header_mixed(self):
521
self.check_header('0-9,300-5000,-50',
522
ranges=[(0,9), (300,5000)],
526
class TestSpecificRequestHandler(http_utils.TestCaseWithWebserver):
527
"""Tests a specific request handler.
529
Daughter classes are expected to override _req_handler_class
532
scenarios = multiply_scenarios(
533
vary_by_http_client_implementation(),
534
vary_by_http_protocol_version(),
537
# Provide a useful default
538
_req_handler_class = http_server.TestingHTTPRequestHandler
540
def create_transport_readonly_server(self):
541
server = http_server.HttpServer(self._req_handler_class,
542
protocol_version=self._protocol_version)
543
server._url_protocol = self._url_protocol
547
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
548
"""Whatever request comes in, close the connection"""
550
def _handle_one_request(self):
551
"""Handle a single HTTP request, by abruptly closing the connection"""
552
self.close_connection = 1
555
class TestWallServer(TestSpecificRequestHandler):
556
"""Tests exceptions during the connection phase"""
558
_req_handler_class = WallRequestHandler
560
def test_http_has(self):
561
t = self.get_readonly_transport()
562
# Unfortunately httplib (see HTTPResponse._read_status
563
# for details) make no distinction between a closed
564
# socket and badly formatted status line, so we can't
565
# just test for ConnectionError, we have to test
566
# InvalidHttpResponse too.
567
self.assertRaises((errors.ConnectionError,
568
errors.InvalidHttpResponse),
571
def test_http_get(self):
572
t = self.get_readonly_transport()
573
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
574
errors.InvalidHttpResponse),
578
class BadStatusRequestHandler(http_server.TestingHTTPRequestHandler):
579
"""Whatever request comes in, returns a bad status"""
581
def parse_request(self):
582
"""Fakes handling a single HTTP request, returns a bad status"""
583
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
584
self.send_response(0, "Bad status")
585
self.close_connection = 1
589
class TestBadStatusServer(TestSpecificRequestHandler):
590
"""Tests bad status from server."""
592
_req_handler_class = BadStatusRequestHandler
595
super(TestBadStatusServer, self).setUp()
596
# See https://bugs.launchpad.net/bzr/+bug/1451448 for details.
597
# TD;LR: Running both a TCP client and server in the same process and
598
# thread uncovers a race in python. The fix is to run the server in a
599
# different process. Trying to fix yet another race here is not worth
600
# the effort. -- vila 2015-09-06
601
if 'HTTP/1.0' in self.id():
602
raise tests.TestSkipped(
603
'Client/Server in the same process and thread can hang')
605
def test_http_has(self):
606
t = self.get_readonly_transport()
607
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
608
errors.InvalidHttpResponse),
611
def test_http_get(self):
612
t = self.get_readonly_transport()
613
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
614
errors.InvalidHttpResponse),
618
class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):
619
"""Whatever request comes in, returns an invalid status"""
621
def parse_request(self):
622
"""Fakes handling a single HTTP request, returns a bad status"""
623
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
624
self.wfile.write("Invalid status line\r\n")
625
# If we don't close the connection pycurl will hang. Since this is a
626
# stress test we don't *have* to respect the protocol, but we don't
627
# have to sabotage it too much either.
628
self.close_connection = True
632
class TestInvalidStatusServer(TestBadStatusServer):
633
"""Tests invalid status from server.
635
Both implementations raises the same error as for a bad status.
638
_req_handler_class = InvalidStatusRequestHandler
641
class BadProtocolRequestHandler(http_server.TestingHTTPRequestHandler):
642
"""Whatever request comes in, returns a bad protocol version"""
644
def parse_request(self):
645
"""Fakes handling a single HTTP request, returns a bad status"""
646
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
647
# Returns an invalid protocol version, but curl just
648
# ignores it and those cannot be tested.
649
self.wfile.write("%s %d %s\r\n" % ('HTTP/0.0',
651
'Look at my protocol version'))
655
class TestBadProtocolServer(TestSpecificRequestHandler):
656
"""Tests bad protocol from server."""
658
_req_handler_class = BadProtocolRequestHandler
660
def test_http_has(self):
661
t = self.get_readonly_transport()
662
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
664
def test_http_get(self):
665
t = self.get_readonly_transport()
666
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
669
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
670
"""Whatever request comes in, returns a 403 code"""
672
def parse_request(self):
673
"""Handle a single HTTP request, by replying we cannot handle it"""
674
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
679
class TestForbiddenServer(TestSpecificRequestHandler):
680
"""Tests forbidden server"""
682
_req_handler_class = ForbiddenRequestHandler
684
def test_http_has(self):
685
t = self.get_readonly_transport()
686
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
688
def test_http_get(self):
689
t = self.get_readonly_transport()
690
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
693
class TestRecordingServer(tests.TestCase):
695
def test_create(self):
696
server = RecordingServer(expect_body_tail=None)
697
self.assertEqual('', server.received_bytes)
698
self.assertEqual(None, server.host)
699
self.assertEqual(None, server.port)
701
def test_setUp_and_stop(self):
702
server = RecordingServer(expect_body_tail=None)
703
server.start_server()
705
self.assertNotEqual(None, server.host)
706
self.assertNotEqual(None, server.port)
709
self.assertEqual(None, server.host)
710
self.assertEqual(None, server.port)
712
def test_send_receive_bytes(self):
713
server = RecordingServer(expect_body_tail='c', scheme='http')
714
self.start_server(server)
715
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
716
sock.connect((server.host, server.port))
718
self.assertEqual('HTTP/1.1 200 OK\r\n',
719
osutils.recv_all(sock, 4096))
720
self.assertEqual('abc', server.received_bytes)
723
class TestRangeRequestServer(TestSpecificRequestHandler):
724
"""Tests readv requests against server.
726
We test against default "normal" server.
730
super(TestRangeRequestServer, self).setUp()
731
self.build_tree_contents([('a', '0123456789')],)
733
def test_readv(self):
734
t = self.get_readonly_transport()
735
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
736
self.assertEqual(l[0], (0, '0'))
737
self.assertEqual(l[1], (1, '1'))
738
self.assertEqual(l[2], (3, '34'))
739
self.assertEqual(l[3], (9, '9'))
741
def test_readv_out_of_order(self):
742
t = self.get_readonly_transport()
743
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
744
self.assertEqual(l[0], (1, '1'))
745
self.assertEqual(l[1], (9, '9'))
746
self.assertEqual(l[2], (0, '0'))
747
self.assertEqual(l[3], (3, '34'))
749
def test_readv_invalid_ranges(self):
750
t = self.get_readonly_transport()
752
# This is intentionally reading off the end of the file
753
# since we are sure that it cannot get there
754
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
755
t.readv, 'a', [(1,1), (8,10)])
757
# This is trying to seek past the end of the file, it should
758
# also raise a special error
759
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
760
t.readv, 'a', [(12,2)])
762
def test_readv_multiple_get_requests(self):
763
server = self.get_readonly_server()
764
t = self.get_readonly_transport()
765
# force transport to issue multiple requests
766
t._max_readv_combine = 1
767
t._max_get_ranges = 1
768
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
769
self.assertEqual(l[0], (0, '0'))
770
self.assertEqual(l[1], (1, '1'))
771
self.assertEqual(l[2], (3, '34'))
772
self.assertEqual(l[3], (9, '9'))
773
# The server should have issued 4 requests
774
self.assertEqual(4, server.GET_request_nb)
776
def test_readv_get_max_size(self):
777
server = self.get_readonly_server()
778
t = self.get_readonly_transport()
779
# force transport to issue multiple requests by limiting the number of
780
# bytes by request. Note that this apply to coalesced offsets only, a
781
# single range will keep its size even if bigger than the limit.
783
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
784
self.assertEqual(l[0], (0, '0'))
785
self.assertEqual(l[1], (1, '1'))
786
self.assertEqual(l[2], (2, '2345'))
787
self.assertEqual(l[3], (6, '6789'))
788
# The server should have issued 3 requests
789
self.assertEqual(3, server.GET_request_nb)
791
def test_complete_readv_leave_pipe_clean(self):
792
server = self.get_readonly_server()
793
t = self.get_readonly_transport()
794
# force transport to issue multiple requests
796
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
797
# The server should have issued 3 requests
798
self.assertEqual(3, server.GET_request_nb)
799
self.assertEqual('0123456789', t.get_bytes('a'))
800
self.assertEqual(4, server.GET_request_nb)
802
def test_incomplete_readv_leave_pipe_clean(self):
803
server = self.get_readonly_server()
804
t = self.get_readonly_transport()
805
# force transport to issue multiple requests
807
# Don't collapse readv results into a list so that we leave unread
808
# bytes on the socket
809
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
810
self.assertEqual((0, '0'), next(ireadv))
811
# The server should have issued one request so far
812
self.assertEqual(1, server.GET_request_nb)
813
self.assertEqual('0123456789', t.get_bytes('a'))
814
# get_bytes issued an additional request, the readv pending ones are
816
self.assertEqual(2, server.GET_request_nb)
819
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
820
"""Always reply to range request as if they were single.
822
Don't be explicit about it, just to annoy the clients.
825
def get_multiple_ranges(self, file, file_size, ranges):
826
"""Answer as if it was a single range request and ignores the rest"""
827
(start, end) = ranges[0]
828
return self.get_single_range(file, file_size, start, end)
831
class TestSingleRangeRequestServer(TestRangeRequestServer):
832
"""Test readv against a server which accept only single range requests"""
834
_req_handler_class = SingleRangeRequestHandler
837
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
838
"""Only reply to simple range requests, errors out on multiple"""
840
def get_multiple_ranges(self, file, file_size, ranges):
841
"""Refuses the multiple ranges request"""
844
self.send_error(416, "Requested range not satisfiable")
846
(start, end) = ranges[0]
847
return self.get_single_range(file, file_size, start, end)
850
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
851
"""Test readv against a server which only accept single range requests"""
853
_req_handler_class = SingleOnlyRangeRequestHandler
856
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
857
"""Ignore range requests without notice"""
860
# Update the statistics
861
self.server.test_case_server.GET_request_nb += 1
862
# Just bypass the range handling done by TestingHTTPRequestHandler
863
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
866
class TestNoRangeRequestServer(TestRangeRequestServer):
867
"""Test readv against a server which do not accept range requests"""
869
_req_handler_class = NoRangeRequestHandler
872
class MultipleRangeWithoutContentLengthRequestHandler(
873
http_server.TestingHTTPRequestHandler):
874
"""Reply to multiple range requests without content length header."""
876
def get_multiple_ranges(self, file, file_size, ranges):
877
self.send_response(206)
878
self.send_header('Accept-Ranges', 'bytes')
879
# XXX: this is strange; the 'random' name below seems undefined and
880
# yet the tests pass -- mbp 2010-10-11 bug 658773
881
boundary = "%d" % random.randint(0,0x7FFFFFFF)
882
self.send_header("Content-Type",
883
"multipart/byteranges; boundary=%s" % boundary)
885
for (start, end) in ranges:
886
self.wfile.write("--%s\r\n" % boundary)
887
self.send_header("Content-type", 'application/octet-stream')
888
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
892
self.send_range_content(file, start, end - start + 1)
894
self.wfile.write("--%s\r\n" % boundary)
897
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
899
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
902
class TruncatedMultipleRangeRequestHandler(
903
http_server.TestingHTTPRequestHandler):
904
"""Reply to multiple range requests truncating the last ones.
906
This server generates responses whose Content-Length describes all the
907
ranges, but fail to include the last ones leading to client short reads.
908
This has been observed randomly with lighttpd (bug #179368).
911
_truncated_ranges = 2
913
def get_multiple_ranges(self, file, file_size, ranges):
914
self.send_response(206)
915
self.send_header('Accept-Ranges', 'bytes')
917
self.send_header('Content-Type',
918
'multipart/byteranges; boundary=%s' % boundary)
919
boundary_line = '--%s\r\n' % boundary
920
# Calculate the Content-Length
922
for (start, end) in ranges:
923
content_length += len(boundary_line)
924
content_length += self._header_line_length(
925
'Content-type', 'application/octet-stream')
926
content_length += self._header_line_length(
927
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
928
content_length += len('\r\n') # end headers
929
content_length += end - start # + 1
930
content_length += len(boundary_line)
931
self.send_header('Content-length', content_length)
934
# Send the multipart body
936
for (start, end) in ranges:
937
self.wfile.write(boundary_line)
938
self.send_header('Content-type', 'application/octet-stream')
939
self.send_header('Content-Range', 'bytes %d-%d/%d'
940
% (start, end, file_size))
942
if cur + self._truncated_ranges >= len(ranges):
943
# Abruptly ends the response and close the connection
944
self.close_connection = 1
946
self.send_range_content(file, start, end - start + 1)
949
self.wfile.write(boundary_line)
952
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
954
_req_handler_class = TruncatedMultipleRangeRequestHandler
957
super(TestTruncatedMultipleRangeServer, self).setUp()
958
self.build_tree_contents([('a', '0123456789')],)
960
def test_readv_with_short_reads(self):
961
server = self.get_readonly_server()
962
t = self.get_readonly_transport()
963
# Force separate ranges for each offset
964
t._bytes_to_read_before_seek = 0
965
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
966
self.assertEqual((0, '0'), next(ireadv))
967
self.assertEqual((2, '2'), next(ireadv))
968
# Only one request have been issued so far
969
self.assertEqual(1, server.GET_request_nb)
970
self.assertEqual((4, '45'), next(ireadv))
971
self.assertEqual((9, '9'), next(ireadv))
972
# We issue 3 requests: two multiple (4 ranges, then 2 ranges) then a
974
self.assertEqual(3, server.GET_request_nb)
975
# Finally the client have tried a single range request and stays in
977
self.assertEqual('single', t._range_hint)
980
class TruncatedBeforeBoundaryRequestHandler(
981
http_server.TestingHTTPRequestHandler):
982
"""Truncation before a boundary, like in bug 198646"""
984
_truncated_ranges = 1
986
def get_multiple_ranges(self, file, file_size, ranges):
987
self.send_response(206)
988
self.send_header('Accept-Ranges', 'bytes')
990
self.send_header('Content-Type',
991
'multipart/byteranges; boundary=%s' % boundary)
992
boundary_line = '--%s\r\n' % boundary
993
# Calculate the Content-Length
995
for (start, end) in ranges:
996
content_length += len(boundary_line)
997
content_length += self._header_line_length(
998
'Content-type', 'application/octet-stream')
999
content_length += self._header_line_length(
1000
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
1001
content_length += len('\r\n') # end headers
1002
content_length += end - start # + 1
1003
content_length += len(boundary_line)
1004
self.send_header('Content-length', content_length)
1007
# Send the multipart body
1009
for (start, end) in ranges:
1010
if cur + self._truncated_ranges >= len(ranges):
1011
# Abruptly ends the response and close the connection
1012
self.close_connection = 1
1014
self.wfile.write(boundary_line)
1015
self.send_header('Content-type', 'application/octet-stream')
1016
self.send_header('Content-Range', 'bytes %d-%d/%d'
1017
% (start, end, file_size))
1019
self.send_range_content(file, start, end - start + 1)
1022
self.wfile.write(boundary_line)
1025
class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):
1026
"""Tests the case of bug 198646, disconnecting before a boundary."""
1028
_req_handler_class = TruncatedBeforeBoundaryRequestHandler
1031
super(TestTruncatedBeforeBoundary, self).setUp()
1032
self.build_tree_contents([('a', '0123456789')],)
1034
def test_readv_with_short_reads(self):
1035
server = self.get_readonly_server()
1036
t = self.get_readonly_transport()
1037
# Force separate ranges for each offset
1038
t._bytes_to_read_before_seek = 0
1039
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1040
self.assertEqual((0, '0'), next(ireadv))
1041
self.assertEqual((2, '2'), next(ireadv))
1042
self.assertEqual((4, '45'), next(ireadv))
1043
self.assertEqual((9, '9'), next(ireadv))
1046
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1047
"""Errors out when range specifiers exceed the limit"""
1049
def get_multiple_ranges(self, file, file_size, ranges):
1050
"""Refuses the multiple ranges request"""
1051
tcs = self.server.test_case_server
1052
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1054
# Emulate apache behavior
1055
self.send_error(400, "Bad Request")
1057
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1058
self, file, file_size, ranges)
1061
class LimitedRangeHTTPServer(http_server.HttpServer):
1062
"""An HttpServer erroring out on requests with too much range specifiers"""
1064
def __init__(self, request_handler=LimitedRangeRequestHandler,
1065
protocol_version=None,
1067
http_server.HttpServer.__init__(self, request_handler,
1068
protocol_version=protocol_version)
1069
self.range_limit = range_limit
1072
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1073
"""Tests readv requests against a server erroring out on too much ranges."""
1075
scenarios = multiply_scenarios(
1076
vary_by_http_client_implementation(),
1077
vary_by_http_protocol_version(),
1080
# Requests with more range specifiers will error out
1083
def create_transport_readonly_server(self):
1084
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1085
protocol_version=self._protocol_version)
1088
super(TestLimitedRangeRequestServer, self).setUp()
1089
# We need to manipulate ranges that correspond to real chunks in the
1090
# response, so we build a content appropriately.
1091
filler = ''.join(['abcdefghij' for x in range(102)])
1092
content = ''.join(['%04d' % v + filler for v in range(16)])
1093
self.build_tree_contents([('a', content)],)
1095
def test_few_ranges(self):
1096
t = self.get_readonly_transport()
1097
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1098
self.assertEqual(l[0], (0, '0000'))
1099
self.assertEqual(l[1], (1024, '0001'))
1100
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1102
def test_more_ranges(self):
1103
t = self.get_readonly_transport()
1104
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1105
self.assertEqual(l[0], (0, '0000'))
1106
self.assertEqual(l[1], (1024, '0001'))
1107
self.assertEqual(l[2], (4096, '0004'))
1108
self.assertEqual(l[3], (8192, '0008'))
1109
# The server will refuse to serve the first request (too much ranges),
1110
# a second request will succeed.
1111
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1114
class TestHttpProxyWhiteBox(tests.TestCase):
1115
"""Whitebox test proxy http authorization.
1117
Only the urllib implementation is tested here.
1120
def _proxied_request(self):
1121
handler = _urllib2_wrappers.ProxyHandler()
1122
request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
1123
handler.set_proxy(request, 'http')
1126
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1127
handler = _urllib2_wrappers.ProxyHandler()
1128
self.assertEqual(expected,
1129
handler.evaluate_proxy_bypass(host, no_proxy))
1131
def test_empty_user(self):
1132
self.overrideEnv('http_proxy', 'http://bar.com')
1133
request = self._proxied_request()
1134
self.assertFalse('Proxy-authorization' in request.headers)
1136
def test_user_with_at(self):
1137
self.overrideEnv('http_proxy',
1138
'http://username@domain:password@proxy_host:1234')
1139
request = self._proxied_request()
1140
self.assertFalse('Proxy-authorization' in request.headers)
1142
def test_invalid_proxy(self):
1143
"""A proxy env variable without scheme"""
1144
self.overrideEnv('http_proxy', 'host:1234')
1145
self.assertRaises(errors.InvalidURL, self._proxied_request)
1147
def test_evaluate_proxy_bypass_true(self):
1148
"""The host is not proxied"""
1149
self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
1150
self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
1152
def test_evaluate_proxy_bypass_false(self):
1153
"""The host is proxied"""
1154
self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
1156
def test_evaluate_proxy_bypass_unknown(self):
1157
"""The host is not explicitly proxied"""
1158
self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
1159
self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
1161
def test_evaluate_proxy_bypass_empty_entries(self):
1162
"""Ignore empty entries"""
1163
self.assertEvaluateProxyBypass(None, 'example.com', '')
1164
self.assertEvaluateProxyBypass(None, 'example.com', ',')
1165
self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
1168
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1169
"""Tests proxy server.
1171
Be aware that we do not setup a real proxy here. Instead, we
1172
check that the *connection* goes through the proxy by serving
1173
different content (the faked proxy server append '-proxied'
1177
scenarios = multiply_scenarios(
1178
vary_by_http_client_implementation(),
1179
vary_by_http_protocol_version(),
1182
# FIXME: We don't have an https server available, so we don't
1183
# test https connections. --vila toolongago
1186
super(TestProxyHttpServer, self).setUp()
1187
self.transport_secondary_server = http_utils.ProxyServer
1188
self.build_tree_contents([('foo', 'contents of foo\n'),
1189
('foo-proxied', 'proxied contents of foo\n')])
1190
# Let's setup some attributes for tests
1191
server = self.get_readonly_server()
1192
self.server_host_port = '%s:%d' % (server.host, server.port)
1193
self.no_proxy_host = self.server_host_port
1194
# The secondary server is the proxy
1195
self.proxy_url = self.get_secondary_url()
1197
def assertProxied(self):
1198
t = self.get_readonly_transport()
1199
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1201
def assertNotProxied(self):
1202
t = self.get_readonly_transport()
1203
self.assertEqual('contents of foo\n', t.get('foo').read())
1205
def test_http_proxy(self):
1206
self.overrideEnv('http_proxy', self.proxy_url)
1207
self.assertProxied()
1209
def test_HTTP_PROXY(self):
1210
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1211
self.assertProxied()
1213
def test_all_proxy(self):
1214
self.overrideEnv('all_proxy', self.proxy_url)
1215
self.assertProxied()
1217
def test_ALL_PROXY(self):
1218
self.overrideEnv('ALL_PROXY', self.proxy_url)
1219
self.assertProxied()
1221
def test_http_proxy_with_no_proxy(self):
1222
self.overrideEnv('no_proxy', self.no_proxy_host)
1223
self.overrideEnv('http_proxy', self.proxy_url)
1224
self.assertNotProxied()
1226
def test_HTTP_PROXY_with_NO_PROXY(self):
1227
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1228
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1229
self.assertNotProxied()
1231
def test_all_proxy_with_no_proxy(self):
1232
self.overrideEnv('no_proxy', self.no_proxy_host)
1233
self.overrideEnv('all_proxy', self.proxy_url)
1234
self.assertNotProxied()
1236
def test_ALL_PROXY_with_NO_PROXY(self):
1237
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1238
self.overrideEnv('ALL_PROXY', self.proxy_url)
1239
self.assertNotProxied()
1241
def test_http_proxy_without_scheme(self):
1242
self.overrideEnv('http_proxy', self.server_host_port)
1243
self.assertRaises(errors.InvalidURL, self.assertProxied)
1246
class TestRanges(http_utils.TestCaseWithWebserver):
1247
"""Test the Range header in GET methods."""
1249
scenarios = multiply_scenarios(
1250
vary_by_http_client_implementation(),
1251
vary_by_http_protocol_version(),
1255
super(TestRanges, self).setUp()
1256
self.build_tree_contents([('a', '0123456789')],)
1258
def create_transport_readonly_server(self):
1259
return http_server.HttpServer(protocol_version=self._protocol_version)
1261
def _file_contents(self, relpath, ranges):
1262
t = self.get_readonly_transport()
1263
offsets = [ (start, end - start + 1) for start, end in ranges]
1264
coalesce = t._coalesce_offsets
1265
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1266
code, data = t._get(relpath, coalesced)
1267
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1268
for start, end in ranges:
1270
yield data.read(end - start + 1)
1272
def _file_tail(self, relpath, tail_amount):
1273
t = self.get_readonly_transport()
1274
code, data = t._get(relpath, [], tail_amount)
1275
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1276
data.seek(-tail_amount, 2)
1277
return data.read(tail_amount)
1279
def test_range_header(self):
1282
['0', '234'], list(self._file_contents('a', [(0,0), (2,4)])))
1284
def test_range_header_tail(self):
1285
self.assertEqual('789', self._file_tail('a', 3))
1287
def test_syntactically_invalid_range_header(self):
1288
self.assertListRaises(errors.InvalidHttpRange,
1289
self._file_contents, 'a', [(4, 3)])
1291
def test_semantically_invalid_range_header(self):
1292
self.assertListRaises(errors.InvalidHttpRange,
1293
self._file_contents, 'a', [(42, 128)])
1296
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1297
"""Test redirection between http servers."""
1299
scenarios = multiply_scenarios(
1300
vary_by_http_client_implementation(),
1301
vary_by_http_protocol_version(),
1305
super(TestHTTPRedirections, self).setUp()
1306
self.build_tree_contents([('a', '0123456789'),
1308
'# Bazaar revision bundle v0.9\n#\n')
1311
def test_redirected(self):
1312
self.assertRaises(errors.RedirectRequested,
1313
self.get_old_transport().get, 'a')
1314
self.assertEqual('0123456789', self.get_new_transport().get('a').read())
1317
class RedirectedRequest(_urllib2_wrappers.Request):
1318
"""Request following redirections. """
1320
init_orig = _urllib2_wrappers.Request.__init__
1322
def __init__(self, method, url, *args, **kwargs):
1326
# Since the tests using this class will replace
1327
# _urllib2_wrappers.Request, we can't just call the base class __init__
1329
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1330
self.follow_redirections = True
1333
def install_redirected_request(test):
1334
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1337
def cleanup_http_redirection_connections(test):
1338
# Some sockets are opened but never seen by _urllib, so we trap them at
1339
# the _urllib2_wrappers level to be able to clean them up.
1340
def socket_disconnect(sock):
1342
sock.shutdown(socket.SHUT_RDWR)
1344
except socket.error:
1346
def connect(connection):
1347
test.http_connect_orig(connection)
1348
test.addCleanup(socket_disconnect, connection.sock)
1349
test.http_connect_orig = test.overrideAttr(
1350
_urllib2_wrappers.HTTPConnection, 'connect', connect)
1351
def connect(connection):
1352
test.https_connect_orig(connection)
1353
test.addCleanup(socket_disconnect, connection.sock)
1354
test.https_connect_orig = test.overrideAttr(
1355
_urllib2_wrappers.HTTPSConnection, 'connect', connect)
1358
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1359
"""Test redirections.
1361
http implementations do not redirect silently anymore (they
1362
do not redirect at all in fact). The mechanism is still in
1363
place at the _urllib2_wrappers.Request level and these tests
1367
scenarios = multiply_scenarios(
1368
vary_by_http_client_implementation(),
1369
vary_by_http_protocol_version(),
1373
super(TestHTTPSilentRedirections, self).setUp()
1374
install_redirected_request(self)
1375
cleanup_http_redirection_connections(self)
1376
self.build_tree_contents([('a','a'),
1378
('1/a', 'redirected once'),
1380
('2/a', 'redirected twice'),
1382
('3/a', 'redirected thrice'),
1384
('4/a', 'redirected 4 times'),
1386
('5/a', 'redirected 5 times'),
1389
def test_one_redirection(self):
1390
t = self.get_old_transport()
1391
req = RedirectedRequest('GET', t._remote_path('a'))
1392
new_prefix = 'http://%s:%s' % (self.new_server.host,
1393
self.new_server.port)
1394
self.old_server.redirections = \
1395
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1396
self.assertEqual('redirected once', t._perform(req).read())
1398
def test_five_redirections(self):
1399
t = self.get_old_transport()
1400
req = RedirectedRequest('GET', t._remote_path('a'))
1401
old_prefix = 'http://%s:%s' % (self.old_server.host,
1402
self.old_server.port)
1403
new_prefix = 'http://%s:%s' % (self.new_server.host,
1404
self.new_server.port)
1405
self.old_server.redirections = [
1406
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1407
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1408
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1409
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1410
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1412
self.assertEqual('redirected 5 times', t._perform(req).read())
1415
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1416
"""Test transport.do_catching_redirections."""
1418
scenarios = multiply_scenarios(
1419
vary_by_http_client_implementation(),
1420
vary_by_http_protocol_version(),
1424
super(TestDoCatchRedirections, self).setUp()
1425
self.build_tree_contents([('a', '0123456789'),],)
1426
cleanup_http_redirection_connections(self)
1428
self.old_transport = self.get_old_transport()
1433
def test_no_redirection(self):
1434
t = self.get_new_transport()
1436
# We use None for redirected so that we fail if redirected
1437
self.assertEqual('0123456789',
1438
transport.do_catching_redirections(
1439
self.get_a, t, None).read())
1441
def test_one_redirection(self):
1442
self.redirections = 0
1444
def redirected(t, exception, redirection_notice):
1445
self.redirections += 1
1446
redirected_t = t._redirected_to(exception.source, exception.target)
1449
self.assertEqual('0123456789',
1450
transport.do_catching_redirections(
1451
self.get_a, self.old_transport, redirected).read())
1452
self.assertEqual(1, self.redirections)
1454
def test_redirection_loop(self):
1456
def redirected(transport, exception, redirection_notice):
1457
# By using the redirected url as a base dir for the
1458
# *old* transport, we create a loop: a => a/a =>
1460
return self.old_transport.clone(exception.target)
1462
self.assertRaises(errors.TooManyRedirections,
1463
transport.do_catching_redirections,
1464
self.get_a, self.old_transport, redirected)
1467
def _setup_authentication_config(**kwargs):
1468
conf = config.AuthenticationConfig()
1469
conf._get_config().update({'httptest': kwargs})
1473
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1474
"""Unit tests for glue by which urllib2 asks us for authentication"""
1476
def test_get_user_password_without_port(self):
1477
"""We cope if urllib2 doesn't tell us the port.
1479
See https://bugs.launchpad.net/bzr/+bug/654684
1483
_setup_authentication_config(scheme='http', host='localhost',
1484
user=user, password=password)
1485
handler = _urllib2_wrappers.HTTPAuthHandler()
1486
got_pass = handler.get_user_password(dict(
1493
self.assertEqual((user, password), got_pass)
1496
class TestAuth(http_utils.TestCaseWithWebserver):
1497
"""Test authentication scheme"""
1499
scenarios = multiply_scenarios(
1500
vary_by_http_client_implementation(),
1501
vary_by_http_protocol_version(),
1502
vary_by_http_auth_scheme(),
1506
super(TestAuth, self).setUp()
1507
self.server = self.get_readonly_server()
1508
self.build_tree_contents([('a', 'contents of a\n'),
1509
('b', 'contents of b\n'),])
1511
def create_transport_readonly_server(self):
1512
server = self._auth_server(protocol_version=self._protocol_version)
1513
server._url_protocol = self._url_protocol
1516
def get_user_url(self, user, password):
1517
"""Build an url embedding user and password"""
1518
url = '%s://' % self.server._url_protocol
1519
if user is not None:
1521
if password is not None:
1522
url += ':' + password
1524
url += '%s:%s/' % (self.server.host, self.server.port)
1527
def get_user_transport(self, user, password):
1528
t = transport.get_transport_from_url(
1529
self.get_user_url(user, password))
1532
def test_no_user(self):
1533
self.server.add_user('joe', 'foo')
1534
t = self.get_user_transport(None, None)
1535
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1536
# Only one 'Authentication Required' error should occur
1537
self.assertEqual(1, self.server.auth_required_errors)
1539
def test_empty_pass(self):
1540
self.server.add_user('joe', '')
1541
t = self.get_user_transport('joe', '')
1542
self.assertEqual('contents of a\n', t.get('a').read())
1543
# Only one 'Authentication Required' error should occur
1544
self.assertEqual(1, self.server.auth_required_errors)
1546
def test_user_pass(self):
1547
self.server.add_user('joe', 'foo')
1548
t = self.get_user_transport('joe', 'foo')
1549
self.assertEqual('contents of a\n', t.get('a').read())
1550
# Only one 'Authentication Required' error should occur
1551
self.assertEqual(1, self.server.auth_required_errors)
1553
def test_unknown_user(self):
1554
self.server.add_user('joe', 'foo')
1555
t = self.get_user_transport('bill', 'foo')
1556
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1557
# Two 'Authentication Required' errors should occur (the
1558
# initial 'who are you' and 'I don't know you, who are
1560
self.assertEqual(2, self.server.auth_required_errors)
1562
def test_wrong_pass(self):
1563
self.server.add_user('joe', 'foo')
1564
t = self.get_user_transport('joe', 'bar')
1565
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1566
# Two 'Authentication Required' errors should occur (the
1567
# initial 'who are you' and 'this is not you, who are you')
1568
self.assertEqual(2, self.server.auth_required_errors)
1570
def test_prompt_for_username(self):
1571
self.server.add_user('joe', 'foo')
1572
t = self.get_user_transport(None, None)
1573
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
1574
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1575
self.assertEqual('contents of a\n',t.get('a').read())
1576
# stdin should be empty
1577
self.assertEqual('', ui.ui_factory.stdin.readline())
1579
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1580
self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1581
self.assertEqual('', stdout.getvalue())
1582
self._check_password_prompt(t._unqualified_scheme, 'joe',
1585
def test_prompt_for_password(self):
1586
self.server.add_user('joe', 'foo')
1587
t = self.get_user_transport('joe', None)
1588
ui.ui_factory = tests.TestUIFactory(stdin='foo\n')
1589
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1590
self.assertEqual('contents of a\n', t.get('a').read())
1591
# stdin should be empty
1592
self.assertEqual('', ui.ui_factory.stdin.readline())
1593
self._check_password_prompt(t._unqualified_scheme, 'joe',
1595
self.assertEqual('', stdout.getvalue())
1596
# And we shouldn't prompt again for a different request
1597
# against the same transport.
1598
self.assertEqual('contents of b\n',t.get('b').read())
1600
# And neither against a clone
1601
self.assertEqual('contents of b\n',t2.get('b').read())
1602
# Only one 'Authentication Required' error should occur
1603
self.assertEqual(1, self.server.auth_required_errors)
1605
def _check_password_prompt(self, scheme, user, actual_prompt):
1606
expected_prompt = (self._password_prompt_prefix
1607
+ ("%s %s@%s:%d, Realm: '%s' password: "
1609
user, self.server.host, self.server.port,
1610
self.server.auth_realm)))
1611
self.assertEqual(expected_prompt, actual_prompt)
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))
1619
def test_no_prompt_for_password_when_using_auth_config(self):
1622
stdin_content = 'bar\n' # Not the right password
1623
self.server.add_user(user, password)
1624
t = self.get_user_transport(user, None)
1625
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1626
# Create a minimal config file with the right password
1627
_setup_authentication_config(scheme='http', port=self.server.port,
1628
user=user, password=password)
1629
# Issue a request to the server to connect
1630
self.assertEqual('contents of a\n',t.get('a').read())
1631
# stdin should have been left untouched
1632
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1633
# Only one 'Authentication Required' error should occur
1634
self.assertEqual(1, self.server.auth_required_errors)
1636
def test_changing_nonce(self):
1637
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1638
http_utils.ProxyDigestAuthServer):
1639
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1640
self.server.add_user('joe', 'foo')
1641
t = self.get_user_transport('joe', 'foo')
1642
self.assertEqual('contents of a\n', t.get('a').read())
1643
self.assertEqual('contents of b\n', t.get('b').read())
1644
# Only one 'Authentication Required' error should have
1646
self.assertEqual(1, self.server.auth_required_errors)
1647
# The server invalidates the current nonce
1648
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1649
self.assertEqual('contents of a\n', t.get('a').read())
1650
# Two 'Authentication Required' errors should occur (the
1651
# initial 'who are you' and a second 'who are you' with the new nonce)
1652
self.assertEqual(2, self.server.auth_required_errors)
1654
def test_user_from_auth_conf(self):
1657
self.server.add_user(user, password)
1658
_setup_authentication_config(scheme='http', port=self.server.port,
1659
user=user, password=password)
1660
t = self.get_user_transport(None, None)
1661
# Issue a request to the server to connect
1662
self.assertEqual('contents of a\n', t.get('a').read())
1663
# Only one 'Authentication Required' error should occur
1664
self.assertEqual(1, self.server.auth_required_errors)
1666
def test_no_credential_leaks_in_log(self):
1667
self.overrideAttr(debug, 'debug_flags', {'http'})
1669
password = 'very-sensitive-password'
1670
self.server.add_user(user, password)
1671
t = self.get_user_transport(user, password)
1672
# Capture the debug calls to mutter
1675
lines = args[0] % args[1:]
1676
# Some calls output multiple lines, just split them now since we
1677
# care about a single one later.
1678
self.mutters.extend(lines.splitlines())
1679
self.overrideAttr(trace, 'mutter', mutter)
1680
# Issue a request to the server to connect
1681
self.assertEqual(True, t.has('a'))
1682
# Only one 'Authentication Required' error should occur
1683
self.assertEqual(1, self.server.auth_required_errors)
1684
# Since the authentification succeeded, there should be a corresponding
1686
sent_auth_headers = [line for line in self.mutters
1687
if line.startswith('> %s' % (self._auth_header,))]
1688
self.assertLength(1, sent_auth_headers)
1689
self.assertStartsWith(sent_auth_headers[0],
1690
'> %s: <masked>' % (self._auth_header,))
1693
class TestProxyAuth(TestAuth):
1694
"""Test proxy authentication schemes.
1696
This inherits from TestAuth to tweak the setUp and filter some failing
1700
scenarios = multiply_scenarios(
1701
vary_by_http_client_implementation(),
1702
vary_by_http_protocol_version(),
1703
vary_by_http_proxy_auth_scheme(),
1707
super(TestProxyAuth, self).setUp()
1708
# Override the contents to avoid false positives
1709
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1710
('b', 'not proxied contents of b\n'),
1711
('a-proxied', 'contents of a\n'),
1712
('b-proxied', 'contents of b\n'),
1715
def get_user_transport(self, user, password):
1716
proxy_url = self.get_user_url(user, password)
1717
self.overrideEnv('all_proxy', proxy_url)
1718
return TestAuth.get_user_transport(self, user, password)
1721
class NonClosingBytesIO(io.BytesIO):
1724
"""Ignore and leave file open."""
1727
class SampleSocket(object):
1728
"""A socket-like object for use in testing the HTTP request handler."""
1730
def __init__(self, socket_read_content):
1731
"""Constructs a sample socket.
1733
:param socket_read_content: a byte sequence
1735
self.readfile = io.BytesIO(socket_read_content)
1736
self.writefile = NonClosingBytesIO()
1739
"""Ignore and leave files alone."""
1741
def makefile(self, mode='r', bufsize=None):
1743
return self.readfile
1745
return self.writefile
1748
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1750
scenarios = multiply_scenarios(
1751
vary_by_http_client_implementation(),
1752
vary_by_http_protocol_version(),
1756
super(SmartHTTPTunnellingTest, self).setUp()
1757
# We use the VFS layer as part of HTTP tunnelling tests.
1758
self.overrideEnv('BRZ_NO_SMART_VFS', None)
1759
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1760
self.http_server = self.get_readonly_server()
1762
def create_transport_readonly_server(self):
1763
server = http_utils.HTTPServerWithSmarts(
1764
protocol_version=self._protocol_version)
1765
server._url_protocol = self._url_protocol
1768
def test_open_controldir(self):
1769
branch = self.make_branch('relpath')
1770
url = self.http_server.get_url() + 'relpath'
1771
bd = controldir.ControlDir.open(url)
1772
self.addCleanup(bd.transport.disconnect)
1773
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1775
def test_bulk_data(self):
1776
# We should be able to send and receive bulk data in a single message.
1777
# The 'readv' command in the smart protocol both sends and receives
1778
# bulk data, so we use that.
1779
self.build_tree(['data-file'])
1780
http_transport = transport.get_transport_from_url(
1781
self.http_server.get_url())
1782
medium = http_transport.get_smart_medium()
1783
# Since we provide the medium, the url below will be mostly ignored
1784
# during the test, as long as the path is '/'.
1785
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1788
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1790
def test_http_send_smart_request(self):
1792
post_body = 'hello\n'
1793
expected_reply_body = 'ok\x012\n'
1795
http_transport = transport.get_transport_from_url(
1796
self.http_server.get_url())
1797
medium = http_transport.get_smart_medium()
1798
response = medium.send_http_smart_request(post_body)
1799
reply_body = response.read()
1800
self.assertEqual(expected_reply_body, reply_body)
1802
def test_smart_http_server_post_request_handler(self):
1803
httpd = self.http_server.server
1805
socket = SampleSocket(
1806
'POST /.bzr/smart %s \r\n' % self._protocol_version
1807
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1809
+ 'Content-Length: 6\r\n'
1812
# Beware: the ('localhost', 80) below is the
1813
# client_address parameter, but we don't have one because
1814
# we have defined a socket which is not bound to an
1815
# address. The test framework never uses this client
1816
# address, so far...
1817
request_handler = http_utils.SmartRequestHandler(socket,
1820
response = socket.writefile.getvalue()
1821
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1822
# This includes the end of the HTTP headers, and all the body.
1823
expected_end_of_response = '\r\n\r\nok\x012\n'
1824
self.assertEndsWith(response, expected_end_of_response)
1827
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1828
"""No smart server here request handler."""
1831
self.send_error(403, "Forbidden")
1834
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1835
"""Test smart client behaviour against an http server without smarts."""
1837
_req_handler_class = ForbiddenRequestHandler
1839
def test_probe_smart_server(self):
1840
"""Test error handling against server refusing smart requests."""
1841
t = self.get_readonly_transport()
1842
# No need to build a valid smart request here, the server will not even
1843
# try to interpret it.
1844
self.assertRaises(errors.SmartProtocolError,
1845
t.get_smart_medium().send_http_smart_request,
1849
class Test_redirected_to(tests.TestCase):
1851
scenarios = vary_by_http_client_implementation()
1853
def test_redirected_to_subdir(self):
1854
t = self._transport('http://www.example.com/foo')
1855
r = t._redirected_to('http://www.example.com/foo',
1856
'http://www.example.com/foo/subdir')
1857
self.assertIsInstance(r, type(t))
1858
# Both transports share the some connection
1859
self.assertEqual(t._get_connection(), r._get_connection())
1860
self.assertEqual('http://www.example.com/foo/subdir/', r.base)
1862
def test_redirected_to_self_with_slash(self):
1863
t = self._transport('http://www.example.com/foo')
1864
r = t._redirected_to('http://www.example.com/foo',
1865
'http://www.example.com/foo/')
1866
self.assertIsInstance(r, type(t))
1867
# Both transports share the some connection (one can argue that we
1868
# should return the exact same transport here, but that seems
1870
self.assertEqual(t._get_connection(), r._get_connection())
1872
def test_redirected_to_host(self):
1873
t = self._transport('http://www.example.com/foo')
1874
r = t._redirected_to('http://www.example.com/foo',
1875
'http://foo.example.com/foo/subdir')
1876
self.assertIsInstance(r, type(t))
1877
self.assertEqual('http://foo.example.com/foo/subdir/',
1880
def test_redirected_to_same_host_sibling_protocol(self):
1881
t = self._transport('http://www.example.com/foo')
1882
r = t._redirected_to('http://www.example.com/foo',
1883
'https://www.example.com/foo')
1884
self.assertIsInstance(r, type(t))
1885
self.assertEqual('https://www.example.com/foo/',
1888
def test_redirected_to_same_host_different_protocol(self):
1889
t = self._transport('http://www.example.com/foo')
1890
r = t._redirected_to('http://www.example.com/foo',
1891
'ftp://www.example.com/foo')
1892
self.assertNotEqual(type(r), type(t))
1893
self.assertEqual('ftp://www.example.com/foo/', r.external_url())
1895
def test_redirected_to_same_host_specific_implementation(self):
1896
t = self._transport('http://www.example.com/foo')
1897
r = t._redirected_to('http://www.example.com/foo',
1898
'https+urllib://www.example.com/foo')
1899
self.assertEqual('https://www.example.com/foo/', r.external_url())
1901
def test_redirected_to_different_host_same_user(self):
1902
t = self._transport('http://joe@www.example.com/foo')
1903
r = t._redirected_to('http://www.example.com/foo',
1904
'https://foo.example.com/foo')
1905
self.assertIsInstance(r, type(t))
1906
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
1907
self.assertEqual('https://joe@foo.example.com/foo/', r.external_url())
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 features.HTTPSServerFeature.available():
1967
from . 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
self.server = self._activity_server(self._protocol_version)
1981
self.server.start_server()
1982
self.addCleanup(self.server.stop_server)
1983
_activities = {} # Don't close over self and create a cycle
1984
def report_activity(t, bytes, direction):
1985
count = _activities.get(direction, 0)
1987
_activities[direction] = count
1988
self.activities = _activities
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)
1994
def get_transport(self):
1995
t = self._transport(self.server.get_url())
1996
# FIXME: Needs cleanup -- vila 20100611
1999
def assertActivitiesMatch(self):
2000
self.assertEqual(self.server.bytes_read,
2001
self.activities.get('write', 0), 'written bytes')
2002
self.assertEqual(self.server.bytes_written,
2003
self.activities.get('read', 0), 'read bytes')
2006
self.server.canned_response = '''HTTP/1.1 200 OK\r
2007
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2008
Server: Apache/2.0.54 (Fedora)\r
2009
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2010
ETag: "56691-23-38e9ae00"\r
2011
Accept-Ranges: bytes\r
2012
Content-Length: 35\r
2014
Content-Type: text/plain; charset=UTF-8\r
2016
Bazaar-NG meta directory, format 1
2018
t = self.get_transport()
2019
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2020
t.get('foo/bar').read())
2021
self.assertActivitiesMatch()
2024
self.server.canned_response = '''HTTP/1.1 200 OK\r
2025
Server: SimpleHTTP/0.6 Python/2.5.2\r
2026
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2027
Content-type: application/octet-stream\r
2028
Content-Length: 20\r
2029
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2032
t = self.get_transport()
2033
self.assertTrue(t.has('foo/bar'))
2034
self.assertActivitiesMatch()
2036
def test_readv(self):
2037
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2038
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2039
Server: Apache/2.0.54 (Fedora)\r
2040
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2041
ETag: "238a3c-16ec2-805c5540"\r
2042
Accept-Ranges: bytes\r
2043
Content-Length: 1534\r
2045
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2048
--418470f848b63279b\r
2049
Content-type: text/plain; charset=UTF-8\r
2050
Content-range: bytes 0-254/93890\r
2052
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2053
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2054
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2055
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2056
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2058
--418470f848b63279b\r
2059
Content-type: text/plain; charset=UTF-8\r
2060
Content-range: bytes 1000-2049/93890\r
2063
mbp@sourcefrog.net-20050311063625-07858525021f270b
2064
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2065
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2066
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2067
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2068
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2069
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2070
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2071
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2072
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2073
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2074
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2075
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2076
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2077
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2078
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2079
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2080
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2081
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2082
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2084
--418470f848b63279b--\r
2086
t = self.get_transport()
2087
# Remember that the request is ignored and that the ranges below
2088
# doesn't have to match the canned response.
2089
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2090
self.assertEqual(2, len(l))
2091
self.assertActivitiesMatch()
2093
def test_post(self):
2094
self.server.canned_response = '''HTTP/1.1 200 OK\r
2095
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2096
Server: Apache/2.0.54 (Fedora)\r
2097
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2098
ETag: "56691-23-38e9ae00"\r
2099
Accept-Ranges: bytes\r
2100
Content-Length: 35\r
2102
Content-Type: text/plain; charset=UTF-8\r
2104
lalala whatever as long as itsssss
2106
t = self.get_transport()
2107
# We must send a single line of body bytes, see
2108
# PredefinedRequestHandler._handle_one_request
2109
code, f = t._post('abc def end-of-body\n')
2110
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2111
self.assertActivitiesMatch()
2114
class TestActivity(tests.TestCase, TestActivityMixin):
2116
scenarios = multiply_scenarios(
2117
vary_by_http_activity(),
2118
vary_by_http_protocol_version(),
2122
super(TestActivity, self).setUp()
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
super(TestNoReportActivity, self).setUp()
2138
self._transport =_urllib.HttpTransport_urllib
2139
TestActivityMixin.setUp(self)
2141
def assertActivitiesMatch(self):
2142
# Nothing to check here
2146
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2147
"""Test authentication on the redirected http server."""
2149
scenarios = vary_by_http_protocol_version()
2151
_auth_header = 'Authorization'
2152
_password_prompt_prefix = ''
2153
_username_prompt_prefix = ''
2154
_auth_server = http_utils.HTTPBasicAuthServer
2155
_transport = _urllib.HttpTransport_urllib
2158
super(TestAuthOnRedirected, self).setUp()
2159
self.build_tree_contents([('a','a'),
2161
('1/a', 'redirected once'),
2163
new_prefix = 'http://%s:%s' % (self.new_server.host,
2164
self.new_server.port)
2165
self.old_server.redirections = [
2166
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2167
self.old_transport = self.get_old_transport()
2168
self.new_server.add_user('joe', 'foo')
2169
cleanup_http_redirection_connections(self)
2171
def create_transport_readonly_server(self):
2172
server = self._auth_server(protocol_version=self._protocol_version)
2173
server._url_protocol = self._url_protocol
2179
def test_auth_on_redirected_via_do_catching_redirections(self):
2180
self.redirections = 0
2182
def redirected(t, exception, redirection_notice):
2183
self.redirections += 1
2184
redirected_t = t._redirected_to(exception.source, exception.target)
2185
self.addCleanup(redirected_t.disconnect)
2188
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2189
self.assertEqual('redirected once',
2190
transport.do_catching_redirections(
2191
self.get_a, self.old_transport, redirected).read())
2192
self.assertEqual(1, self.redirections)
2193
# stdin should be empty
2194
self.assertEqual('', ui.ui_factory.stdin.readline())
2195
# stdout should be empty, stderr will contains the prompts
2196
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2198
def test_auth_on_redirected_via_following_redirections(self):
2199
self.new_server.add_user('joe', 'foo')
2200
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2201
t = self.old_transport
2202
req = RedirectedRequest('GET', t.abspath('a'))
2203
new_prefix = 'http://%s:%s' % (self.new_server.host,
2204
self.new_server.port)
2205
self.old_server.redirections = [
2206
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2207
self.assertEqual('redirected once', t._perform(req).read())
2208
# stdin should be empty
2209
self.assertEqual('', ui.ui_factory.stdin.readline())
2210
# stdout should be empty, stderr will contains the prompts
2211
self.assertEqual('', ui.ui_factory.stdout.getvalue())
added
removed
breezy/tests/test_http.py
