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 ?
27
from http.client import UnknownProtocol, parse_headers
28
from http.server import SimpleHTTPRequestHandler
29
except ImportError: # python < 3
30
from httplib import UnknownProtocol
31
from SimpleHTTPServer import SimpleHTTPRequestHandler
51
remote as _mod_remote,
53
from ..sixish import PY3
60
from .scenarios import (
61
load_tests_apply_scenarios,
64
from ..transport import (
68
from ..transport.http import (
73
load_tests = load_tests_apply_scenarios
76
def vary_by_http_client_implementation():
77
"""Test the libraries we can use, currently just urllib."""
78
transport_scenarios = [
79
('urllib', dict(_transport=HttpTransport,
80
_server=http_server.HttpServer,
81
_url_protocol='http',)),
83
return transport_scenarios
86
def vary_by_http_protocol_version():
87
"""Test on http/1.0 and 1.1"""
89
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
90
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
94
def vary_by_http_auth_scheme():
96
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
97
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
99
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
101
# Add some attributes common to all scenarios
102
for scenario_id, scenario_dict in scenarios:
103
scenario_dict.update(_auth_header='Authorization',
104
_username_prompt_prefix='',
105
_password_prompt_prefix='')
109
def vary_by_http_proxy_auth_scheme():
111
('proxy-basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
112
('proxy-digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
113
('proxy-basicdigest',
114
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
116
# Add some attributes common to all scenarios
117
for scenario_id, scenario_dict in scenarios:
118
scenario_dict.update(_auth_header='Proxy-Authorization',
119
_username_prompt_prefix='Proxy ',
120
_password_prompt_prefix='Proxy ')
124
def vary_by_http_activity():
125
activity_scenarios = [
126
('urllib,http', dict(_activity_server=ActivityHTTPServer,
127
_transport=HttpTransport,)),
129
if features.HTTPSServerFeature.available():
130
# FIXME: Until we have a better way to handle self-signed certificates
131
# (like allowing them in a test specific authentication.conf for
132
# example), we need some specialized urllib transport for tests.
138
class HTTPS_transport(HttpTransport):
140
def __init__(self, base, _from_transport=None):
141
super(HTTPS_transport, self).__init__(
142
base, _from_transport=_from_transport,
143
ca_certs=ssl_certs.build_path('ca.crt'))
145
activity_scenarios.append(
146
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
147
_transport=HTTPS_transport,)),)
148
return activity_scenarios
151
class FakeManager(object):
154
self.credentials = []
156
def add_password(self, realm, host, username, password):
157
self.credentials.append([realm, host, username, password])
160
class RecordingServer(object):
161
"""A fake HTTP server.
163
It records the bytes sent to it, and replies with a 200.
166
def __init__(self, expect_body_tail=None, scheme=''):
169
:type expect_body_tail: str
170
:param expect_body_tail: a reply won't be sent until this string is
173
self._expect_body_tail = expect_body_tail
176
self.received_bytes = b''
180
return '%s://%s:%s/' % (self.scheme, self.host, self.port)
182
def start_server(self):
183
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
184
self._sock.bind(('127.0.0.1', 0))
185
self.host, self.port = self._sock.getsockname()
186
self._ready = threading.Event()
187
self._thread = test_server.TestThread(
188
sync_event=self._ready, target=self._accept_read_and_reply)
190
if 'threads' in tests.selftest_debug_flags:
191
sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
194
def _accept_read_and_reply(self):
197
conn, address = self._sock.accept()
198
if self._expect_body_tail is not None:
199
while not self.received_bytes.endswith(self._expect_body_tail):
200
self.received_bytes += conn.recv(4096)
201
conn.sendall(b'HTTP/1.1 200 OK\r\n')
205
# The client may have already closed the socket.
208
def stop_server(self):
210
# Issue a fake connection to wake up the server and allow it to
212
fake_conn = osutils.connect_socket((self.host, self.port))
215
# We might have already closed it. We don't care.
220
if 'threads' in tests.selftest_debug_flags:
221
sys.stderr.write('Thread joined: %s\n' % (self._thread.ident,))
224
class TestAuthHeader(tests.TestCase):
226
def parse_header(self, header, auth_handler_class=None):
227
if auth_handler_class is None:
228
auth_handler_class = http.AbstractAuthHandler
229
self.auth_handler = auth_handler_class()
230
return self.auth_handler._parse_auth_header(header)
232
def test_empty_header(self):
233
scheme, remainder = self.parse_header('')
234
self.assertEqual('', scheme)
235
self.assertIs(None, remainder)
237
def test_negotiate_header(self):
238
scheme, remainder = self.parse_header('Negotiate')
239
self.assertEqual('negotiate', scheme)
240
self.assertIs(None, remainder)
242
def test_basic_header(self):
243
scheme, remainder = self.parse_header(
244
'Basic realm="Thou should not pass"')
245
self.assertEqual('basic', scheme)
246
self.assertEqual('realm="Thou should not pass"', remainder)
248
def test_build_basic_header_with_long_creds(self):
249
handler = http.BasicAuthHandler()
250
user = 'user' * 10 # length 40
251
password = 'password' * 5 # length 40
252
header = handler.build_auth_header(
253
dict(user=user, password=password), None)
254
# https://bugs.launchpad.net/bzr/+bug/1606203 was caused by incorrectly
255
# creating a header value with an embedded '\n'
256
self.assertFalse('\n' in header)
258
def test_basic_extract_realm(self):
259
scheme, remainder = self.parse_header(
260
'Basic realm="Thou should not pass"',
261
http.BasicAuthHandler)
262
match, realm = self.auth_handler.extract_realm(remainder)
263
self.assertTrue(match is not None)
264
self.assertEqual(u'Thou should not pass', realm)
266
def test_digest_header(self):
267
scheme, remainder = self.parse_header(
268
'Digest realm="Thou should not pass"')
269
self.assertEqual('digest', scheme)
270
self.assertEqual('realm="Thou should not pass"', remainder)
273
class TestHTTPRangeParsing(tests.TestCase):
276
super(TestHTTPRangeParsing, self).setUp()
277
# We focus on range parsing here and ignore everything else
279
class RequestHandler(http_server.TestingHTTPRequestHandler):
280
def setup(self): pass
282
def handle(self): pass
284
def finish(self): pass
286
self.req_handler = RequestHandler(None, None, None)
288
def assertRanges(self, ranges, header, file_size):
289
self.assertEqual(ranges,
290
self.req_handler._parse_ranges(header, file_size))
292
def test_simple_range(self):
293
self.assertRanges([(0, 2)], 'bytes=0-2', 12)
296
self.assertRanges([(8, 11)], 'bytes=-4', 12)
298
def test_tail_bigger_than_file(self):
299
self.assertRanges([(0, 11)], 'bytes=-99', 12)
301
def test_range_without_end(self):
302
self.assertRanges([(4, 11)], 'bytes=4-', 12)
304
def test_invalid_ranges(self):
305
self.assertRanges(None, 'bytes=12-22', 12)
306
self.assertRanges(None, 'bytes=1-3,12-22', 12)
307
self.assertRanges(None, 'bytes=-', 12)
310
class TestHTTPServer(tests.TestCase):
311
"""Test the HTTP servers implementations."""
313
def test_invalid_protocol(self):
314
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
316
protocol_version = 'HTTP/0.1'
318
self.assertRaises(UnknownProtocol,
319
http_server.HttpServer, BogusRequestHandler)
321
def test_force_invalid_protocol(self):
322
self.assertRaises(UnknownProtocol,
323
http_server.HttpServer, protocol_version='HTTP/0.1')
325
def test_server_start_and_stop(self):
326
server = http_server.HttpServer()
327
self.addCleanup(server.stop_server)
328
server.start_server()
329
self.assertTrue(server.server is not None)
330
self.assertTrue(server.server.serving is not None)
331
self.assertTrue(server.server.serving)
333
def test_create_http_server_one_zero(self):
334
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
336
protocol_version = 'HTTP/1.0'
338
server = http_server.HttpServer(RequestHandlerOneZero)
339
self.start_server(server)
340
self.assertIsInstance(server.server, http_server.TestingHTTPServer)
342
def test_create_http_server_one_one(self):
343
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
345
protocol_version = 'HTTP/1.1'
347
server = http_server.HttpServer(RequestHandlerOneOne)
348
self.start_server(server)
349
self.assertIsInstance(server.server,
350
http_server.TestingThreadingHTTPServer)
352
def test_create_http_server_force_one_one(self):
353
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
355
protocol_version = 'HTTP/1.0'
357
server = http_server.HttpServer(RequestHandlerOneZero,
358
protocol_version='HTTP/1.1')
359
self.start_server(server)
360
self.assertIsInstance(server.server,
361
http_server.TestingThreadingHTTPServer)
363
def test_create_http_server_force_one_zero(self):
364
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
366
protocol_version = 'HTTP/1.1'
368
server = http_server.HttpServer(RequestHandlerOneOne,
369
protocol_version='HTTP/1.0')
370
self.start_server(server)
371
self.assertIsInstance(server.server,
372
http_server.TestingHTTPServer)
375
class TestHttpTransportUrls(tests.TestCase):
376
"""Test the http urls."""
378
scenarios = vary_by_http_client_implementation()
380
def test_abs_url(self):
381
"""Construction of absolute http URLs"""
382
t = self._transport('http://example.com/bzr/bzr.dev/')
383
eq = self.assertEqualDiff
384
eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
385
eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
386
eq(t.abspath('.bzr'), 'http://example.com/bzr/bzr.dev/.bzr')
387
eq(t.abspath('.bzr/1//2/./3'),
388
'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
390
def test_invalid_http_urls(self):
391
"""Trap invalid construction of urls"""
392
self._transport('http://example.com/bzr/bzr.dev/')
393
self.assertRaises(urlutils.InvalidURL,
395
'http://example.com:port/bzr/bzr.dev/')
397
def test_http_root_urls(self):
398
"""Construction of URLs from server root"""
399
t = self._transport('http://example.com/')
400
eq = self.assertEqualDiff
401
eq(t.abspath('.bzr/tree-version'),
402
'http://example.com/.bzr/tree-version')
404
def test_http_impl_urls(self):
405
"""There are servers which ask for particular clients to connect"""
406
server = self._server()
407
server.start_server()
409
url = server.get_url()
410
self.assertTrue(url.startswith('%s://' % self._url_protocol))
415
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
416
"""Test the http connections."""
418
scenarios = multiply_scenarios(
419
vary_by_http_client_implementation(),
420
vary_by_http_protocol_version(),
424
super(TestHTTPConnections, self).setUp()
425
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
426
transport=self.get_transport())
428
def test_http_has(self):
429
server = self.get_readonly_server()
430
t = self.get_readonly_transport()
431
self.assertEqual(t.has('foo/bar'), True)
432
self.assertEqual(len(server.logs), 1)
433
self.assertContainsRe(server.logs[0],
434
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "Breezy/')
436
def test_http_has_not_found(self):
437
server = self.get_readonly_server()
438
t = self.get_readonly_transport()
439
self.assertEqual(t.has('not-found'), False)
440
self.assertContainsRe(server.logs[1],
441
r'"HEAD /not-found HTTP/1.." 404 - "-" "Breezy/')
443
def test_http_get(self):
444
server = self.get_readonly_server()
445
t = self.get_readonly_transport()
446
fp = t.get('foo/bar')
447
self.assertEqualDiff(
449
b'contents of foo/bar\n')
450
self.assertEqual(len(server.logs), 1)
451
self.assertTrue(server.logs[0].find(
452
'"GET /foo/bar HTTP/1.1" 200 - "-" "Breezy/%s'
453
% breezy.__version__) > -1)
455
def test_has_on_bogus_host(self):
456
# Get a free address and don't 'accept' on it, so that we
457
# can be sure there is no http handler there, but set a
458
# reasonable timeout to not slow down tests too much.
459
default_timeout = socket.getdefaulttimeout()
461
socket.setdefaulttimeout(2)
463
s.bind(('localhost', 0))
464
t = self._transport('http://%s:%s/' % s.getsockname())
465
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
467
socket.setdefaulttimeout(default_timeout)
470
class TestHttpTransportRegistration(tests.TestCase):
471
"""Test registrations of various http implementations"""
473
scenarios = vary_by_http_client_implementation()
475
def test_http_registered(self):
476
t = transport.get_transport_from_url(
477
'%s://foo.com/' % self._url_protocol)
478
self.assertIsInstance(t, transport.Transport)
479
self.assertIsInstance(t, self._transport)
482
class TestPost(tests.TestCase):
484
scenarios = multiply_scenarios(
485
vary_by_http_client_implementation(),
486
vary_by_http_protocol_version(),
489
def test_post_body_is_received(self):
490
server = RecordingServer(expect_body_tail=b'end-of-body',
491
scheme=self._url_protocol)
492
self.start_server(server)
493
url = server.get_url()
494
# FIXME: needs a cleanup -- vila 20100611
495
http_transport = transport.get_transport_from_url(url)
496
code, response = http_transport._post(b'abc def end-of-body')
498
server.received_bytes.startswith(b'POST /.bzr/smart HTTP/1.'))
500
b'content-length: 19\r' in server.received_bytes.lower())
501
self.assertTrue(b'content-type: application/octet-stream\r'
502
in server.received_bytes.lower())
503
# The transport should not be assuming that the server can accept
504
# chunked encoding the first time it connects, because HTTP/1.1, so we
505
# check for the literal string.
507
server.received_bytes.endswith(b'\r\n\r\nabc def end-of-body'))
510
class TestRangeHeader(tests.TestCase):
511
"""Test range_header method"""
513
def check_header(self, value, ranges=[], tail=0):
514
offsets = [(start, end - start + 1) for start, end in ranges]
515
coalesce = transport.Transport._coalesce_offsets
516
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
517
range_header = http.HttpTransport._range_header
518
self.assertEqual(value, range_header(coalesced, tail))
520
def test_range_header_single(self):
521
self.check_header('0-9', ranges=[(0, 9)])
522
self.check_header('100-109', ranges=[(100, 109)])
524
def test_range_header_tail(self):
525
self.check_header('-10', tail=10)
526
self.check_header('-50', tail=50)
528
def test_range_header_multi(self):
529
self.check_header('0-9,100-200,300-5000',
530
ranges=[(0, 9), (100, 200), (300, 5000)])
532
def test_range_header_mixed(self):
533
self.check_header('0-9,300-5000,-50',
534
ranges=[(0, 9), (300, 5000)],
538
class TestSpecificRequestHandler(http_utils.TestCaseWithWebserver):
539
"""Tests a specific request handler.
541
Daughter classes are expected to override _req_handler_class
544
scenarios = multiply_scenarios(
545
vary_by_http_client_implementation(),
546
vary_by_http_protocol_version(),
549
# Provide a useful default
550
_req_handler_class = http_server.TestingHTTPRequestHandler
552
def create_transport_readonly_server(self):
553
server = http_server.HttpServer(self._req_handler_class,
554
protocol_version=self._protocol_version)
555
server._url_protocol = self._url_protocol
559
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
560
"""Whatever request comes in, close the connection"""
562
def _handle_one_request(self):
563
"""Handle a single HTTP request, by abruptly closing the connection"""
564
self.close_connection = 1
567
class TestWallServer(TestSpecificRequestHandler):
568
"""Tests exceptions during the connection phase"""
570
_req_handler_class = WallRequestHandler
572
def test_http_has(self):
573
t = self.get_readonly_transport()
574
# Unfortunately httplib (see HTTPResponse._read_status
575
# for details) make no distinction between a closed
576
# socket and badly formatted status line, so we can't
577
# just test for ConnectionError, we have to test
578
# InvalidHttpResponse too.
579
self.assertRaises((errors.ConnectionError,
580
errors.InvalidHttpResponse),
583
def test_http_get(self):
584
t = self.get_readonly_transport()
585
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
586
errors.InvalidHttpResponse),
590
class BadStatusRequestHandler(http_server.TestingHTTPRequestHandler):
591
"""Whatever request comes in, returns a bad status"""
593
def parse_request(self):
594
"""Fakes handling a single HTTP request, returns a bad status"""
595
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
596
self.send_response(0, "Bad status")
597
self.close_connection = 1
601
class TestBadStatusServer(TestSpecificRequestHandler):
602
"""Tests bad status from server."""
604
_req_handler_class = BadStatusRequestHandler
607
super(TestBadStatusServer, self).setUp()
608
# See https://bugs.launchpad.net/bzr/+bug/1451448 for details.
609
# TD;LR: Running both a TCP client and server in the same process and
610
# thread uncovers a race in python. The fix is to run the server in a
611
# different process. Trying to fix yet another race here is not worth
612
# the effort. -- vila 2015-09-06
613
if 'HTTP/1.0' in self.id():
614
raise tests.TestSkipped(
615
'Client/Server in the same process and thread can hang')
617
def test_http_has(self):
618
t = self.get_readonly_transport()
619
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
620
errors.InvalidHttpResponse),
623
def test_http_get(self):
624
t = self.get_readonly_transport()
625
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
626
errors.InvalidHttpResponse),
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(b"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(b"%s %d %s\r\n" % (
662
b'HTTP/0.0', 404, b'Look at my protocol version'))
666
class TestBadProtocolServer(TestSpecificRequestHandler):
667
"""Tests bad protocol from server."""
669
_req_handler_class = BadProtocolRequestHandler
671
def test_http_has(self):
672
t = self.get_readonly_transport()
673
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
675
def test_http_get(self):
676
t = self.get_readonly_transport()
677
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
680
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
681
"""Whatever request comes in, returns a 403 code"""
683
def parse_request(self):
684
"""Handle a single HTTP request, by replying we cannot handle it"""
685
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
690
class TestForbiddenServer(TestSpecificRequestHandler):
691
"""Tests forbidden server"""
693
_req_handler_class = ForbiddenRequestHandler
695
def test_http_has(self):
696
t = self.get_readonly_transport()
697
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
699
def test_http_get(self):
700
t = self.get_readonly_transport()
701
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
704
class TestRecordingServer(tests.TestCase):
706
def test_create(self):
707
server = RecordingServer(expect_body_tail=None)
708
self.assertEqual(b'', server.received_bytes)
709
self.assertEqual(None, server.host)
710
self.assertEqual(None, server.port)
712
def test_setUp_and_stop(self):
713
server = RecordingServer(expect_body_tail=None)
714
server.start_server()
716
self.assertNotEqual(None, server.host)
717
self.assertNotEqual(None, server.port)
720
self.assertEqual(None, server.host)
721
self.assertEqual(None, server.port)
723
def test_send_receive_bytes(self):
724
server = RecordingServer(expect_body_tail=b'c', scheme='http')
725
self.start_server(server)
726
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
727
sock.connect((server.host, server.port))
729
self.assertEqual(b'HTTP/1.1 200 OK\r\n',
730
osutils.recv_all(sock, 4096))
731
self.assertEqual(b'abc', server.received_bytes)
734
class TestRangeRequestServer(TestSpecificRequestHandler):
735
"""Tests readv requests against server.
737
We test against default "normal" server.
741
super(TestRangeRequestServer, self).setUp()
742
self.build_tree_contents([('a', b'0123456789')],)
744
def test_readv(self):
745
t = self.get_readonly_transport()
746
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
747
self.assertEqual(l[0], (0, b'0'))
748
self.assertEqual(l[1], (1, b'1'))
749
self.assertEqual(l[2], (3, b'34'))
750
self.assertEqual(l[3], (9, b'9'))
752
def test_readv_out_of_order(self):
753
t = self.get_readonly_transport()
754
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
755
self.assertEqual(l[0], (1, b'1'))
756
self.assertEqual(l[1], (9, b'9'))
757
self.assertEqual(l[2], (0, b'0'))
758
self.assertEqual(l[3], (3, b'34'))
760
def test_readv_invalid_ranges(self):
761
t = self.get_readonly_transport()
763
# This is intentionally reading off the end of the file
764
# since we are sure that it cannot get there
765
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
766
t.readv, 'a', [(1, 1), (8, 10)])
768
# This is trying to seek past the end of the file, it should
769
# also raise a special error
770
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
771
t.readv, 'a', [(12, 2)])
773
def test_readv_multiple_get_requests(self):
774
server = self.get_readonly_server()
775
t = self.get_readonly_transport()
776
# force transport to issue multiple requests
777
t._max_readv_combine = 1
778
t._max_get_ranges = 1
779
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
780
self.assertEqual(l[0], (0, b'0'))
781
self.assertEqual(l[1], (1, b'1'))
782
self.assertEqual(l[2], (3, b'34'))
783
self.assertEqual(l[3], (9, b'9'))
784
# The server should have issued 4 requests
785
self.assertEqual(4, server.GET_request_nb)
787
def test_readv_get_max_size(self):
788
server = self.get_readonly_server()
789
t = self.get_readonly_transport()
790
# force transport to issue multiple requests by limiting the number of
791
# bytes by request. Note that this apply to coalesced offsets only, a
792
# single range will keep its size even if bigger than the limit.
794
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
795
self.assertEqual(l[0], (0, b'0'))
796
self.assertEqual(l[1], (1, b'1'))
797
self.assertEqual(l[2], (2, b'2345'))
798
self.assertEqual(l[3], (6, b'6789'))
799
# The server should have issued 3 requests
800
self.assertEqual(3, server.GET_request_nb)
802
def test_complete_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
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
808
# The server should have issued 3 requests
809
self.assertEqual(3, server.GET_request_nb)
810
self.assertEqual(b'0123456789', t.get_bytes('a'))
811
self.assertEqual(4, server.GET_request_nb)
813
def test_incomplete_readv_leave_pipe_clean(self):
814
server = self.get_readonly_server()
815
t = self.get_readonly_transport()
816
# force transport to issue multiple requests
818
# Don't collapse readv results into a list so that we leave unread
819
# bytes on the socket
820
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
821
self.assertEqual((0, b'0'), next(ireadv))
822
# The server should have issued one request so far
823
self.assertEqual(1, server.GET_request_nb)
824
self.assertEqual(b'0123456789', t.get_bytes('a'))
825
# get_bytes issued an additional request, the readv pending ones are
827
self.assertEqual(2, server.GET_request_nb)
830
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
831
"""Always reply to range request as if they were single.
833
Don't be explicit about it, just to annoy the clients.
836
def get_multiple_ranges(self, file, file_size, ranges):
837
"""Answer as if it was a single range request and ignores the rest"""
838
(start, end) = ranges[0]
839
return self.get_single_range(file, file_size, start, end)
842
class TestSingleRangeRequestServer(TestRangeRequestServer):
843
"""Test readv against a server which accept only single range requests"""
845
_req_handler_class = SingleRangeRequestHandler
848
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
849
"""Only reply to simple range requests, errors out on multiple"""
851
def get_multiple_ranges(self, file, file_size, ranges):
852
"""Refuses the multiple ranges request"""
855
self.send_error(416, "Requested range not satisfiable")
857
(start, end) = ranges[0]
858
return self.get_single_range(file, file_size, start, end)
861
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
862
"""Test readv against a server which only accept single range requests"""
864
_req_handler_class = SingleOnlyRangeRequestHandler
867
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
868
"""Ignore range requests without notice"""
871
# Update the statistics
872
self.server.test_case_server.GET_request_nb += 1
873
# Just bypass the range handling done by TestingHTTPRequestHandler
874
return SimpleHTTPRequestHandler.do_GET(self)
877
class TestNoRangeRequestServer(TestRangeRequestServer):
878
"""Test readv against a server which do not accept range requests"""
880
_req_handler_class = NoRangeRequestHandler
883
class MultipleRangeWithoutContentLengthRequestHandler(
884
http_server.TestingHTTPRequestHandler):
885
"""Reply to multiple range requests without content length header."""
887
def get_multiple_ranges(self, file, file_size, ranges):
888
self.send_response(206)
889
self.send_header('Accept-Ranges', 'bytes')
890
# XXX: this is strange; the 'random' name below seems undefined and
891
# yet the tests pass -- mbp 2010-10-11 bug 658773
892
boundary = "%d" % random.randint(0, 0x7FFFFFFF)
893
self.send_header("Content-Type",
894
"multipart/byteranges; boundary=%s" % boundary)
896
for (start, end) in ranges:
897
self.wfile.write(b"--%s\r\n" % boundary.encode('ascii'))
898
self.send_header("Content-type", 'application/octet-stream')
899
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
903
self.send_range_content(file, start, end - start + 1)
905
self.wfile.write(b"--%s\r\n" % boundary)
908
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
910
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
913
class TruncatedMultipleRangeRequestHandler(
914
http_server.TestingHTTPRequestHandler):
915
"""Reply to multiple range requests truncating the last ones.
917
This server generates responses whose Content-Length describes all the
918
ranges, but fail to include the last ones leading to client short reads.
919
This has been observed randomly with lighttpd (bug #179368).
922
_truncated_ranges = 2
924
def get_multiple_ranges(self, file, file_size, ranges):
925
self.send_response(206)
926
self.send_header('Accept-Ranges', 'bytes')
928
self.send_header('Content-Type',
929
'multipart/byteranges; boundary=%s' % boundary)
930
boundary_line = b'--%s\r\n' % boundary.encode('ascii')
931
# Calculate the Content-Length
933
for (start, end) in ranges:
934
content_length += len(boundary_line)
935
content_length += self._header_line_length(
936
'Content-type', 'application/octet-stream')
937
content_length += self._header_line_length(
938
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
939
content_length += len('\r\n') # end headers
940
content_length += end - start # + 1
941
content_length += len(boundary_line)
942
self.send_header('Content-length', content_length)
945
# Send the multipart body
947
for (start, end) in ranges:
948
self.wfile.write(boundary_line)
949
self.send_header('Content-type', 'application/octet-stream')
950
self.send_header('Content-Range', 'bytes %d-%d/%d'
951
% (start, end, file_size))
953
if cur + self._truncated_ranges >= len(ranges):
954
# Abruptly ends the response and close the connection
955
self.close_connection = 1
957
self.send_range_content(file, start, end - start + 1)
960
self.wfile.write(boundary_line)
963
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
965
_req_handler_class = TruncatedMultipleRangeRequestHandler
968
super(TestTruncatedMultipleRangeServer, self).setUp()
969
self.build_tree_contents([('a', b'0123456789')],)
971
def test_readv_with_short_reads(self):
972
server = self.get_readonly_server()
973
t = self.get_readonly_transport()
974
# Force separate ranges for each offset
975
t._bytes_to_read_before_seek = 0
976
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
977
self.assertEqual((0, b'0'), next(ireadv))
978
self.assertEqual((2, b'2'), next(ireadv))
979
# Only one request have been issued so far
980
self.assertEqual(1, server.GET_request_nb)
981
self.assertEqual((4, b'45'), next(ireadv))
982
self.assertEqual((9, b'9'), next(ireadv))
983
# We issue 3 requests: two multiple (4 ranges, then 2 ranges) then a
985
self.assertEqual(3, server.GET_request_nb)
986
# Finally the client have tried a single range request and stays in
988
self.assertEqual('single', t._range_hint)
991
class TruncatedBeforeBoundaryRequestHandler(
992
http_server.TestingHTTPRequestHandler):
993
"""Truncation before a boundary, like in bug 198646"""
995
_truncated_ranges = 1
997
def get_multiple_ranges(self, file, file_size, ranges):
998
self.send_response(206)
999
self.send_header('Accept-Ranges', 'bytes')
1001
self.send_header('Content-Type',
1002
'multipart/byteranges; boundary=%s' % boundary)
1003
boundary_line = b'--%s\r\n' % boundary.encode('ascii')
1004
# Calculate the Content-Length
1006
for (start, end) in ranges:
1007
content_length += len(boundary_line)
1008
content_length += self._header_line_length(
1009
'Content-type', 'application/octet-stream')
1010
content_length += self._header_line_length(
1011
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
1012
content_length += len('\r\n') # end headers
1013
content_length += end - start # + 1
1014
content_length += len(boundary_line)
1015
self.send_header('Content-length', content_length)
1018
# Send the multipart body
1020
for (start, end) in ranges:
1021
if cur + self._truncated_ranges >= len(ranges):
1022
# Abruptly ends the response and close the connection
1023
self.close_connection = 1
1025
self.wfile.write(boundary_line)
1026
self.send_header('Content-type', 'application/octet-stream')
1027
self.send_header('Content-Range', 'bytes %d-%d/%d'
1028
% (start, end, file_size))
1030
self.send_range_content(file, start, end - start + 1)
1033
self.wfile.write(boundary_line)
1036
class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):
1037
"""Tests the case of bug 198646, disconnecting before a boundary."""
1039
_req_handler_class = TruncatedBeforeBoundaryRequestHandler
1042
super(TestTruncatedBeforeBoundary, self).setUp()
1043
self.build_tree_contents([('a', b'0123456789')],)
1045
def test_readv_with_short_reads(self):
1046
server = self.get_readonly_server()
1047
t = self.get_readonly_transport()
1048
# Force separate ranges for each offset
1049
t._bytes_to_read_before_seek = 0
1050
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1051
self.assertEqual((0, b'0'), next(ireadv))
1052
self.assertEqual((2, b'2'), next(ireadv))
1053
self.assertEqual((4, b'45'), next(ireadv))
1054
self.assertEqual((9, b'9'), next(ireadv))
1057
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1058
"""Errors out when range specifiers exceed the limit"""
1060
def get_multiple_ranges(self, file, file_size, ranges):
1061
"""Refuses the multiple ranges request"""
1062
tcs = self.server.test_case_server
1063
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1065
# Emulate apache behavior
1066
self.send_error(400, "Bad Request")
1068
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1069
self, file, file_size, ranges)
1072
class LimitedRangeHTTPServer(http_server.HttpServer):
1073
"""An HttpServer erroring out on requests with too much range specifiers"""
1075
def __init__(self, request_handler=LimitedRangeRequestHandler,
1076
protocol_version=None,
1078
http_server.HttpServer.__init__(self, request_handler,
1079
protocol_version=protocol_version)
1080
self.range_limit = range_limit
1083
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1084
"""Tests readv requests against a server erroring out on too much ranges."""
1086
scenarios = multiply_scenarios(
1087
vary_by_http_client_implementation(),
1088
vary_by_http_protocol_version(),
1091
# Requests with more range specifiers will error out
1094
def create_transport_readonly_server(self):
1095
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1096
protocol_version=self._protocol_version)
1099
super(TestLimitedRangeRequestServer, self).setUp()
1100
# We need to manipulate ranges that correspond to real chunks in the
1101
# response, so we build a content appropriately.
1102
filler = b''.join([b'abcdefghij' for x in range(102)])
1103
content = b''.join([b'%04d' % v + filler for v in range(16)])
1104
self.build_tree_contents([('a', content)],)
1106
def test_few_ranges(self):
1107
t = self.get_readonly_transport()
1108
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1109
self.assertEqual(l[0], (0, b'0000'))
1110
self.assertEqual(l[1], (1024, b'0001'))
1111
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1113
def test_more_ranges(self):
1114
t = self.get_readonly_transport()
1115
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1116
self.assertEqual(l[0], (0, b'0000'))
1117
self.assertEqual(l[1], (1024, b'0001'))
1118
self.assertEqual(l[2], (4096, b'0004'))
1119
self.assertEqual(l[3], (8192, b'0008'))
1120
# The server will refuse to serve the first request (too much ranges),
1121
# a second request will succeed.
1122
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1125
class TestHttpProxyWhiteBox(tests.TestCase):
1126
"""Whitebox test proxy http authorization.
1128
Only the urllib implementation is tested here.
1131
def _proxied_request(self):
1132
handler = http.ProxyHandler()
1133
request = http.Request('GET', 'http://baz/buzzle')
1134
handler.set_proxy(request, 'http')
1137
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1138
handler = http.ProxyHandler()
1139
self.assertEqual(expected,
1140
handler.evaluate_proxy_bypass(host, no_proxy))
1142
def test_empty_user(self):
1143
self.overrideEnv('http_proxy', 'http://bar.com')
1144
request = self._proxied_request()
1145
self.assertFalse('Proxy-authorization' in request.headers)
1147
def test_user_with_at(self):
1148
self.overrideEnv('http_proxy',
1149
'http://username@domain:password@proxy_host:1234')
1150
request = self._proxied_request()
1151
self.assertFalse('Proxy-authorization' in request.headers)
1153
def test_invalid_proxy(self):
1154
"""A proxy env variable without scheme"""
1155
self.overrideEnv('http_proxy', 'host:1234')
1156
self.assertRaises(urlutils.InvalidURL, self._proxied_request)
1158
def test_evaluate_proxy_bypass_true(self):
1159
"""The host is not proxied"""
1160
self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
1161
self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
1163
def test_evaluate_proxy_bypass_false(self):
1164
"""The host is proxied"""
1165
self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
1167
def test_evaluate_proxy_bypass_unknown(self):
1168
"""The host is not explicitly proxied"""
1169
self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
1170
self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
1172
def test_evaluate_proxy_bypass_empty_entries(self):
1173
"""Ignore empty entries"""
1174
self.assertEvaluateProxyBypass(None, 'example.com', '')
1175
self.assertEvaluateProxyBypass(None, 'example.com', ',')
1176
self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
1179
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1180
"""Tests proxy server.
1182
Be aware that we do not setup a real proxy here. Instead, we
1183
check that the *connection* goes through the proxy by serving
1184
different content (the faked proxy server append '-proxied'
1188
scenarios = multiply_scenarios(
1189
vary_by_http_client_implementation(),
1190
vary_by_http_protocol_version(),
1193
# FIXME: We don't have an https server available, so we don't
1194
# test https connections. --vila toolongago
1197
super(TestProxyHttpServer, self).setUp()
1198
self.transport_secondary_server = http_utils.ProxyServer
1199
self.build_tree_contents([('foo', b'contents of foo\n'),
1200
('foo-proxied', b'proxied contents of foo\n')])
1201
# Let's setup some attributes for tests
1202
server = self.get_readonly_server()
1203
self.server_host_port = '%s:%d' % (server.host, server.port)
1204
self.no_proxy_host = self.server_host_port
1205
# The secondary server is the proxy
1206
self.proxy_url = self.get_secondary_url()
1208
def assertProxied(self):
1209
t = self.get_readonly_transport()
1210
self.assertEqual(b'proxied contents of foo\n', t.get('foo').read())
1212
def assertNotProxied(self):
1213
t = self.get_readonly_transport()
1214
self.assertEqual(b'contents of foo\n', t.get('foo').read())
1216
def test_http_proxy(self):
1217
self.overrideEnv('http_proxy', self.proxy_url)
1218
self.assertProxied()
1220
def test_HTTP_PROXY(self):
1221
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1222
self.assertProxied()
1224
def test_all_proxy(self):
1225
self.overrideEnv('all_proxy', self.proxy_url)
1226
self.assertProxied()
1228
def test_ALL_PROXY(self):
1229
self.overrideEnv('ALL_PROXY', self.proxy_url)
1230
self.assertProxied()
1232
def test_http_proxy_with_no_proxy(self):
1233
self.overrideEnv('no_proxy', self.no_proxy_host)
1234
self.overrideEnv('http_proxy', self.proxy_url)
1235
self.assertNotProxied()
1237
def test_HTTP_PROXY_with_NO_PROXY(self):
1238
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1239
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1240
self.assertNotProxied()
1242
def test_all_proxy_with_no_proxy(self):
1243
self.overrideEnv('no_proxy', self.no_proxy_host)
1244
self.overrideEnv('all_proxy', self.proxy_url)
1245
self.assertNotProxied()
1247
def test_ALL_PROXY_with_NO_PROXY(self):
1248
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1249
self.overrideEnv('ALL_PROXY', self.proxy_url)
1250
self.assertNotProxied()
1252
def test_http_proxy_without_scheme(self):
1253
self.overrideEnv('http_proxy', self.server_host_port)
1254
self.assertRaises(urlutils.InvalidURL, self.assertProxied)
1257
class TestRanges(http_utils.TestCaseWithWebserver):
1258
"""Test the Range header in GET methods."""
1260
scenarios = multiply_scenarios(
1261
vary_by_http_client_implementation(),
1262
vary_by_http_protocol_version(),
1266
super(TestRanges, self).setUp()
1267
self.build_tree_contents([('a', b'0123456789')],)
1269
def create_transport_readonly_server(self):
1270
return http_server.HttpServer(protocol_version=self._protocol_version)
1272
def _file_contents(self, relpath, ranges):
1273
t = self.get_readonly_transport()
1274
offsets = [(start, end - start + 1) for start, end in ranges]
1275
coalesce = t._coalesce_offsets
1276
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1277
code, data = t._get(relpath, coalesced)
1278
self.assertTrue(code in (200, 206), '_get returns: %d' % code)
1279
for start, end in ranges:
1281
yield data.read(end - start + 1)
1283
def _file_tail(self, relpath, tail_amount):
1284
t = self.get_readonly_transport()
1285
code, data = t._get(relpath, [], tail_amount)
1286
self.assertTrue(code in (200, 206), '_get returns: %d' % code)
1287
data.seek(-tail_amount, 2)
1288
return data.read(tail_amount)
1290
def test_range_header(self):
1293
[b'0', b'234'], list(self._file_contents('a', [(0, 0), (2, 4)])))
1295
def test_range_header_tail(self):
1296
self.assertEqual(b'789', self._file_tail('a', 3))
1298
def test_syntactically_invalid_range_header(self):
1299
self.assertListRaises(errors.InvalidHttpRange,
1300
self._file_contents, 'a', [(4, 3)])
1302
def test_semantically_invalid_range_header(self):
1303
self.assertListRaises(errors.InvalidHttpRange,
1304
self._file_contents, 'a', [(42, 128)])
1307
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1308
"""Test redirection between http servers."""
1310
scenarios = multiply_scenarios(
1311
vary_by_http_client_implementation(),
1312
vary_by_http_protocol_version(),
1316
super(TestHTTPRedirections, self).setUp()
1317
self.build_tree_contents([('a', b'0123456789'),
1319
b'# Bazaar revision bundle v0.9\n#\n')
1322
def test_redirected(self):
1323
self.assertRaises(errors.RedirectRequested,
1324
self.get_old_transport().get, 'a')
1327
self.get_new_transport().get('a').read())
1330
class RedirectedRequest(http.Request):
1331
"""Request following redirections. """
1333
init_orig = http.Request.__init__
1335
def __init__(self, method, url, *args, **kwargs):
1339
# Since the tests using this class will replace
1340
# http.Request, we can't just call the base class __init__
1342
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1343
self.follow_redirections = True
1346
def install_redirected_request(test):
1347
test.overrideAttr(http, 'Request', RedirectedRequest)
1350
def cleanup_http_redirection_connections(test):
1351
# Some sockets are opened but never seen by _urllib, so we trap them at
1352
# the http level to be able to clean them up.
1353
def socket_disconnect(sock):
1355
sock.shutdown(socket.SHUT_RDWR)
1357
except socket.error:
1360
def connect(connection):
1361
test.http_connect_orig(connection)
1362
test.addCleanup(socket_disconnect, connection.sock)
1363
test.http_connect_orig = test.overrideAttr(
1364
http.HTTPConnection, 'connect', connect)
1366
def connect(connection):
1367
test.https_connect_orig(connection)
1368
test.addCleanup(socket_disconnect, connection.sock)
1369
test.https_connect_orig = test.overrideAttr(
1370
http.HTTPSConnection, 'connect', connect)
1373
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1374
"""Test redirections.
1376
http implementations do not redirect silently anymore (they
1377
do not redirect at all in fact). The mechanism is still in
1378
place at the http.Request level and these tests
1382
scenarios = multiply_scenarios(
1383
vary_by_http_client_implementation(),
1384
vary_by_http_protocol_version(),
1388
super(TestHTTPSilentRedirections, self).setUp()
1389
install_redirected_request(self)
1390
cleanup_http_redirection_connections(self)
1391
self.build_tree_contents([('a', b'a'),
1393
('1/a', b'redirected once'),
1395
('2/a', b'redirected twice'),
1397
('3/a', b'redirected thrice'),
1399
('4/a', b'redirected 4 times'),
1401
('5/a', b'redirected 5 times'),
1404
def test_one_redirection(self):
1405
t = self.get_old_transport()
1406
new_prefix = 'http://%s:%s' % (self.new_server.host,
1407
self.new_server.port)
1408
self.old_server.redirections = \
1409
[('(.*)', r'%s/1\1' % (new_prefix), 301), ]
1412
t.request('GET', t._remote_path('a'), retries=1).read())
1414
def test_five_redirections(self):
1415
t = self.get_old_transport()
1416
old_prefix = 'http://%s:%s' % (self.old_server.host,
1417
self.old_server.port)
1418
new_prefix = 'http://%s:%s' % (self.new_server.host,
1419
self.new_server.port)
1420
self.old_server.redirections = [
1421
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1422
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1423
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1424
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1425
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1428
b'redirected 5 times',
1429
t.request('GET', t._remote_path('a'), retries=6).read())
1432
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1433
"""Test transport.do_catching_redirections."""
1435
scenarios = multiply_scenarios(
1436
vary_by_http_client_implementation(),
1437
vary_by_http_protocol_version(),
1441
super(TestDoCatchRedirections, self).setUp()
1442
self.build_tree_contents([('a', b'0123456789'), ],)
1443
cleanup_http_redirection_connections(self)
1445
self.old_transport = self.get_old_transport()
1450
def test_no_redirection(self):
1451
t = self.get_new_transport()
1453
# We use None for redirected so that we fail if redirected
1454
self.assertEqual(b'0123456789',
1455
transport.do_catching_redirections(
1456
self.get_a, t, None).read())
1458
def test_one_redirection(self):
1459
self.redirections = 0
1461
def redirected(t, exception, redirection_notice):
1462
self.redirections += 1
1463
redirected_t = t._redirected_to(exception.source, exception.target)
1466
self.assertEqual(b'0123456789',
1467
transport.do_catching_redirections(
1468
self.get_a, self.old_transport, redirected).read())
1469
self.assertEqual(1, self.redirections)
1471
def test_redirection_loop(self):
1473
def redirected(transport, exception, redirection_notice):
1474
# By using the redirected url as a base dir for the
1475
# *old* transport, we create a loop: a => a/a =>
1477
return self.old_transport.clone(exception.target)
1479
self.assertRaises(errors.TooManyRedirections,
1480
transport.do_catching_redirections,
1481
self.get_a, self.old_transport, redirected)
1484
def _setup_authentication_config(**kwargs):
1485
conf = config.AuthenticationConfig()
1486
conf._get_config().update({'httptest': kwargs})
1490
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1491
"""Unit tests for glue by which urllib2 asks us for authentication"""
1493
def test_get_user_password_without_port(self):
1494
"""We cope if urllib2 doesn't tell us the port.
1496
See https://bugs.launchpad.net/bzr/+bug/654684
1500
_setup_authentication_config(scheme='http', host='localhost',
1501
user=user, password=password)
1502
handler = http.HTTPAuthHandler()
1503
got_pass = handler.get_user_password(dict(
1510
self.assertEqual((user, password), got_pass)
1513
class TestAuth(http_utils.TestCaseWithWebserver):
1514
"""Test authentication scheme"""
1516
scenarios = multiply_scenarios(
1517
vary_by_http_client_implementation(),
1518
vary_by_http_protocol_version(),
1519
vary_by_http_auth_scheme(),
1523
super(TestAuth, self).setUp()
1524
self.server = self.get_readonly_server()
1525
self.build_tree_contents([('a', b'contents of a\n'),
1526
('b', b'contents of b\n'), ])
1528
def create_transport_readonly_server(self):
1529
server = self._auth_server(protocol_version=self._protocol_version)
1530
server._url_protocol = self._url_protocol
1533
def get_user_url(self, user, password):
1534
"""Build an url embedding user and password"""
1535
url = '%s://' % self.server._url_protocol
1536
if user is not None:
1538
if password is not None:
1539
url += ':' + password
1541
url += '%s:%s/' % (self.server.host, self.server.port)
1544
def get_user_transport(self, user, password):
1545
t = transport.get_transport_from_url(
1546
self.get_user_url(user, password))
1549
def test_no_user(self):
1550
self.server.add_user('joe', 'foo')
1551
t = self.get_user_transport(None, None)
1552
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1553
# Only one 'Authentication Required' error should occur
1554
self.assertEqual(1, self.server.auth_required_errors)
1556
def test_empty_pass(self):
1557
self.server.add_user('joe', '')
1558
t = self.get_user_transport('joe', '')
1559
self.assertEqual(b'contents of a\n', t.get('a').read())
1560
# Only one 'Authentication Required' error should occur
1561
self.assertEqual(1, self.server.auth_required_errors)
1563
def test_user_pass(self):
1564
self.server.add_user('joe', 'foo')
1565
t = self.get_user_transport('joe', 'foo')
1566
self.assertEqual(b'contents of a\n', t.get('a').read())
1567
# Only one 'Authentication Required' error should occur
1568
self.assertEqual(1, self.server.auth_required_errors)
1570
def test_unknown_user(self):
1571
self.server.add_user('joe', 'foo')
1572
t = self.get_user_transport('bill', 'foo')
1573
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1574
# Two 'Authentication Required' errors should occur (the
1575
# initial 'who are you' and 'I don't know you, who are
1577
self.assertEqual(2, self.server.auth_required_errors)
1579
def test_wrong_pass(self):
1580
self.server.add_user('joe', 'foo')
1581
t = self.get_user_transport('joe', 'bar')
1582
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1583
# Two 'Authentication Required' errors should occur (the
1584
# initial 'who are you' and 'this is not you, who are you')
1585
self.assertEqual(2, self.server.auth_required_errors)
1587
def test_prompt_for_username(self):
1588
self.server.add_user('joe', 'foo')
1589
t = self.get_user_transport(None, None)
1590
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
1591
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1592
self.assertEqual(b'contents of a\n', t.get('a').read())
1593
# stdin should be empty
1594
self.assertEqual('', ui.ui_factory.stdin.readline())
1596
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1597
self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1598
self.assertEqual('', stdout.getvalue())
1599
self._check_password_prompt(t._unqualified_scheme, 'joe',
1602
def test_prompt_for_password(self):
1603
self.server.add_user('joe', 'foo')
1604
t = self.get_user_transport('joe', None)
1605
ui.ui_factory = tests.TestUIFactory(stdin='foo\n')
1606
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1607
self.assertEqual(b'contents of a\n', t.get('a').read())
1608
# stdin should be empty
1609
self.assertEqual('', ui.ui_factory.stdin.readline())
1610
self._check_password_prompt(t._unqualified_scheme, 'joe',
1612
self.assertEqual('', stdout.getvalue())
1613
# And we shouldn't prompt again for a different request
1614
# against the same transport.
1615
self.assertEqual(b'contents of b\n', t.get('b').read())
1617
# And neither against a clone
1618
self.assertEqual(b'contents of b\n', t2.get('b').read())
1619
# Only one 'Authentication Required' error should occur
1620
self.assertEqual(1, self.server.auth_required_errors)
1622
def _check_password_prompt(self, scheme, user, actual_prompt):
1623
expected_prompt = (self._password_prompt_prefix
1624
+ ("%s %s@%s:%d, Realm: '%s' password: "
1626
user, self.server.host, self.server.port,
1627
self.server.auth_realm)))
1628
self.assertEqual(expected_prompt, actual_prompt)
1630
def _expected_username_prompt(self, scheme):
1631
return (self._username_prompt_prefix
1632
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1633
self.server.host, self.server.port,
1634
self.server.auth_realm))
1636
def test_no_prompt_for_password_when_using_auth_config(self):
1639
stdin_content = 'bar\n' # Not the right password
1640
self.server.add_user(user, password)
1641
t = self.get_user_transport(user, None)
1642
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1643
# Create a minimal config file with the right password
1644
_setup_authentication_config(scheme='http', port=self.server.port,
1645
user=user, password=password)
1646
# Issue a request to the server to connect
1647
with t.get('a') as f:
1648
self.assertEqual(b'contents of a\n', f.read())
1649
# stdin should have been left untouched
1650
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1651
# Only one 'Authentication Required' error should occur
1652
self.assertEqual(1, self.server.auth_required_errors)
1654
def test_changing_nonce(self):
1655
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1656
http_utils.ProxyDigestAuthServer):
1657
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1658
self.server.add_user('joe', 'foo')
1659
t = self.get_user_transport('joe', 'foo')
1660
with t.get('a') as f:
1661
self.assertEqual(b'contents of a\n', f.read())
1662
with t.get('b') as f:
1663
self.assertEqual(b'contents of b\n', f.read())
1664
# Only one 'Authentication Required' error should have
1666
self.assertEqual(1, self.server.auth_required_errors)
1667
# The server invalidates the current nonce
1668
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1669
self.assertEqual(b'contents of a\n', t.get('a').read())
1670
# Two 'Authentication Required' errors should occur (the
1671
# initial 'who are you' and a second 'who are you' with the new nonce)
1672
self.assertEqual(2, self.server.auth_required_errors)
1674
def test_user_from_auth_conf(self):
1677
self.server.add_user(user, password)
1678
_setup_authentication_config(scheme='http', port=self.server.port,
1679
user=user, password=password)
1680
t = self.get_user_transport(None, None)
1681
# Issue a request to the server to connect
1682
with t.get('a') as f:
1683
self.assertEqual(b'contents of a\n', f.read())
1684
# Only one 'Authentication Required' error should occur
1685
self.assertEqual(1, self.server.auth_required_errors)
1687
def test_no_credential_leaks_in_log(self):
1688
self.overrideAttr(debug, 'debug_flags', {'http'})
1690
password = 'very-sensitive-password'
1691
self.server.add_user(user, password)
1692
t = self.get_user_transport(user, password)
1693
# Capture the debug calls to mutter
1697
lines = args[0] % args[1:]
1698
# Some calls output multiple lines, just split them now since we
1699
# care about a single one later.
1700
self.mutters.extend(lines.splitlines())
1701
self.overrideAttr(trace, 'mutter', mutter)
1702
# Issue a request to the server to connect
1703
self.assertEqual(True, t.has('a'))
1704
# Only one 'Authentication Required' error should occur
1705
self.assertEqual(1, self.server.auth_required_errors)
1706
# Since the authentification succeeded, there should be a corresponding
1708
sent_auth_headers = [line for line in self.mutters
1709
if line.startswith('> %s' % (self._auth_header,))]
1710
self.assertLength(1, sent_auth_headers)
1711
self.assertStartsWith(sent_auth_headers[0],
1712
'> %s: <masked>' % (self._auth_header,))
1715
class TestProxyAuth(TestAuth):
1716
"""Test proxy authentication schemes.
1718
This inherits from TestAuth to tweak the setUp and filter some failing
1722
scenarios = multiply_scenarios(
1723
vary_by_http_client_implementation(),
1724
vary_by_http_protocol_version(),
1725
vary_by_http_proxy_auth_scheme(),
1729
super(TestProxyAuth, self).setUp()
1730
# Override the contents to avoid false positives
1731
self.build_tree_contents([('a', b'not proxied contents of a\n'),
1732
('b', b'not proxied contents of b\n'),
1733
('a-proxied', b'contents of a\n'),
1734
('b-proxied', b'contents of b\n'),
1737
def get_user_transport(self, user, password):
1738
proxy_url = self.get_user_url(user, password)
1739
self.overrideEnv('all_proxy', proxy_url)
1740
return TestAuth.get_user_transport(self, user, password)
1743
class NonClosingBytesIO(io.BytesIO):
1746
"""Ignore and leave file open."""
1749
class SampleSocket(object):
1750
"""A socket-like object for use in testing the HTTP request handler."""
1752
def __init__(self, socket_read_content):
1753
"""Constructs a sample socket.
1755
:param socket_read_content: a byte sequence
1757
self.readfile = io.BytesIO(socket_read_content)
1758
self.writefile = NonClosingBytesIO()
1761
"""Ignore and leave files alone."""
1763
def sendall(self, bytes):
1764
self.writefile.write(bytes)
1766
def makefile(self, mode='r', bufsize=None):
1768
return self.readfile
1770
return self.writefile
1773
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1775
scenarios = multiply_scenarios(
1776
vary_by_http_client_implementation(),
1777
vary_by_http_protocol_version(),
1781
super(SmartHTTPTunnellingTest, self).setUp()
1782
# We use the VFS layer as part of HTTP tunnelling tests.
1783
self.overrideEnv('BRZ_NO_SMART_VFS', None)
1784
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1785
self.http_server = self.get_readonly_server()
1787
def create_transport_readonly_server(self):
1788
server = http_utils.HTTPServerWithSmarts(
1789
protocol_version=self._protocol_version)
1790
server._url_protocol = self._url_protocol
1793
def test_open_controldir(self):
1794
branch = self.make_branch('relpath')
1795
url = self.http_server.get_url() + 'relpath'
1796
bd = controldir.ControlDir.open(url)
1797
self.addCleanup(bd.transport.disconnect)
1798
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1800
def test_bulk_data(self):
1801
# We should be able to send and receive bulk data in a single message.
1802
# The 'readv' command in the smart protocol both sends and receives
1803
# bulk data, so we use that.
1804
self.build_tree(['data-file'])
1805
http_transport = transport.get_transport_from_url(
1806
self.http_server.get_url())
1807
medium = http_transport.get_smart_medium()
1808
# Since we provide the medium, the url below will be mostly ignored
1809
# during the test, as long as the path is '/'.
1810
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1813
[(0, b"c")], list(remote_transport.readv("data-file", [(0, 1)])))
1815
def test_http_send_smart_request(self):
1817
post_body = b'hello\n'
1818
expected_reply_body = b'ok\x012\n'
1820
http_transport = transport.get_transport_from_url(
1821
self.http_server.get_url())
1822
medium = http_transport.get_smart_medium()
1823
response = medium.send_http_smart_request(post_body)
1824
reply_body = response.read()
1825
self.assertEqual(expected_reply_body, reply_body)
1827
def test_smart_http_server_post_request_handler(self):
1828
httpd = self.http_server.server
1830
socket = SampleSocket(
1831
b'POST /.bzr/smart %s \r\n' % self._protocol_version.encode('ascii') +
1832
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1834
b'Content-Length: 6\r\n'
1837
# Beware: the ('localhost', 80) below is the
1838
# client_address parameter, but we don't have one because
1839
# we have defined a socket which is not bound to an
1840
# address. The test framework never uses this client
1841
# address, so far...
1842
request_handler = http_utils.SmartRequestHandler(socket,
1845
response = socket.writefile.getvalue()
1846
self.assertStartsWith(
1848
b'%s 200 ' % self._protocol_version.encode('ascii'))
1849
# This includes the end of the HTTP headers, and all the body.
1850
expected_end_of_response = b'\r\n\r\nok\x012\n'
1851
self.assertEndsWith(response, expected_end_of_response)
1854
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1855
"""No smart server here request handler."""
1858
self.send_error(403, "Forbidden")
1861
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1862
"""Test smart client behaviour against an http server without smarts."""
1864
_req_handler_class = ForbiddenRequestHandler
1866
def test_probe_smart_server(self):
1867
"""Test error handling against server refusing smart requests."""
1868
t = self.get_readonly_transport()
1869
# No need to build a valid smart request here, the server will not even
1870
# try to interpret it.
1871
self.assertRaises(errors.SmartProtocolError,
1872
t.get_smart_medium().send_http_smart_request,
1876
class Test_redirected_to(tests.TestCase):
1878
scenarios = vary_by_http_client_implementation()
1880
def test_redirected_to_subdir(self):
1881
t = self._transport('http://www.example.com/foo')
1882
r = t._redirected_to('http://www.example.com/foo',
1883
'http://www.example.com/foo/subdir')
1884
self.assertIsInstance(r, type(t))
1885
# Both transports share the some connection
1886
self.assertEqual(t._get_connection(), r._get_connection())
1887
self.assertEqual('http://www.example.com/foo/subdir/', r.base)
1889
def test_redirected_to_self_with_slash(self):
1890
t = self._transport('http://www.example.com/foo')
1891
r = t._redirected_to('http://www.example.com/foo',
1892
'http://www.example.com/foo/')
1893
self.assertIsInstance(r, type(t))
1894
# Both transports share the some connection (one can argue that we
1895
# should return the exact same transport here, but that seems
1897
self.assertEqual(t._get_connection(), r._get_connection())
1899
def test_redirected_to_host(self):
1900
t = self._transport('http://www.example.com/foo')
1901
r = t._redirected_to('http://www.example.com/foo',
1902
'http://foo.example.com/foo/subdir')
1903
self.assertIsInstance(r, type(t))
1904
self.assertEqual('http://foo.example.com/foo/subdir/',
1907
def test_redirected_to_same_host_sibling_protocol(self):
1908
t = self._transport('http://www.example.com/foo')
1909
r = t._redirected_to('http://www.example.com/foo',
1910
'https://www.example.com/foo')
1911
self.assertIsInstance(r, type(t))
1912
self.assertEqual('https://www.example.com/foo/',
1915
def test_redirected_to_same_host_different_protocol(self):
1916
t = self._transport('http://www.example.com/foo')
1917
r = t._redirected_to('http://www.example.com/foo',
1918
'bzr://www.example.com/foo')
1919
self.assertNotEqual(type(r), type(t))
1920
self.assertEqual('bzr://www.example.com/foo/', r.external_url())
1922
def test_redirected_to_same_host_specific_implementation(self):
1923
t = self._transport('http://www.example.com/foo')
1924
r = t._redirected_to('http://www.example.com/foo',
1925
'https+urllib://www.example.com/foo')
1926
self.assertEqual('https://www.example.com/foo/', r.external_url())
1928
def test_redirected_to_different_host_same_user(self):
1929
t = self._transport('http://joe@www.example.com/foo')
1930
r = t._redirected_to('http://www.example.com/foo',
1931
'https://foo.example.com/foo')
1932
self.assertIsInstance(r, type(t))
1933
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
1934
self.assertEqual('https://joe@foo.example.com/foo/', r.external_url())
1937
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1938
"""Request handler for a unique and pre-defined request.
1940
The only thing we care about here is how many bytes travel on the wire. But
1941
since we want to measure it for a real http client, we have to send it
1944
We expect to receive a *single* request nothing more (and we won't even
1945
check what request it is, we just measure the bytes read until an empty
1949
def _handle_one_request(self):
1950
tcs = self.server.test_case_server
1951
requestline = self.rfile.readline()
1953
headers = parse_headers(self.rfile)
1954
bytes_read = len(headers.as_bytes())
1955
bytes_read += headers.as_bytes().count(b'\n')
1956
bytes_read += len(requestline)
1958
headers = self.MessageClass(self.rfile, 0)
1959
# We just read: the request, the headers, an empty line indicating the
1960
# end of the headers.
1961
bytes_read = len(requestline)
1962
for line in headers.headers:
1963
bytes_read += len(line)
1964
bytes_read += len(b'\r\n')
1965
if requestline.startswith(b'POST'):
1966
# The body should be a single line (or we don't know where it ends
1967
# and we don't want to issue a blocking read)
1968
body = self.rfile.readline()
1969
bytes_read += len(body)
1970
tcs.bytes_read = bytes_read
1972
# We set the bytes written *before* issuing the write, the client is
1973
# supposed to consume every produced byte *before* checking that value.
1975
# Doing the oppposite may lead to test failure: we may be interrupted
1976
# after the write but before updating the value. The client can then
1977
# continue and read the value *before* we can update it. And yes,
1978
# this has been observed -- vila 20090129
1979
tcs.bytes_written = len(tcs.canned_response)
1980
self.wfile.write(tcs.canned_response)
1983
class ActivityServerMixin(object):
1985
def __init__(self, protocol_version):
1986
super(ActivityServerMixin, self).__init__(
1987
request_handler=PredefinedRequestHandler,
1988
protocol_version=protocol_version)
1989
# Bytes read and written by the server
1991
self.bytes_written = 0
1992
self.canned_response = None
1995
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1999
if features.HTTPSServerFeature.available():
2000
from . import https_server
2002
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
2006
class TestActivityMixin(object):
2007
"""Test socket activity reporting.
2009
We use a special purpose server to control the bytes sent and received and
2010
be able to predict the activity on the client socket.
2014
self.server = self._activity_server(self._protocol_version)
2015
self.server.start_server()
2016
self.addCleanup(self.server.stop_server)
2017
_activities = {} # Don't close over self and create a cycle
2019
def report_activity(t, bytes, direction):
2020
count = _activities.get(direction, 0)
2022
_activities[direction] = count
2023
self.activities = _activities
2024
# We override at class level because constructors may propagate the
2025
# bound method and render instance overriding ineffective (an
2026
# alternative would be to define a specific ui factory instead...)
2027
self.overrideAttr(self._transport, '_report_activity', report_activity)
2029
def get_transport(self):
2030
t = self._transport(self.server.get_url())
2031
# FIXME: Needs cleanup -- vila 20100611
2034
def assertActivitiesMatch(self):
2035
self.assertEqual(self.server.bytes_read,
2036
self.activities.get('write', 0), 'written bytes')
2037
self.assertEqual(self.server.bytes_written,
2038
self.activities.get('read', 0), 'read bytes')
2041
self.server.canned_response = b'''HTTP/1.1 200 OK\r
2042
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2043
Server: Apache/2.0.54 (Fedora)\r
2044
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2045
ETag: "56691-23-38e9ae00"\r
2046
Accept-Ranges: bytes\r
2047
Content-Length: 35\r
2049
Content-Type: text/plain; charset=UTF-8\r
2051
Bazaar-NG meta directory, format 1
2053
t = self.get_transport()
2054
self.assertEqual(b'Bazaar-NG meta directory, format 1\n',
2055
t.get('foo/bar').read())
2056
self.assertActivitiesMatch()
2059
self.server.canned_response = b'''HTTP/1.1 200 OK\r
2060
Server: SimpleHTTP/0.6 Python/2.5.2\r
2061
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2062
Content-type: application/octet-stream\r
2063
Content-Length: 20\r
2064
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2067
t = self.get_transport()
2068
self.assertTrue(t.has('foo/bar'))
2069
self.assertActivitiesMatch()
2071
def test_readv(self):
2072
self.server.canned_response = b'''HTTP/1.1 206 Partial Content\r
2073
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2074
Server: Apache/2.0.54 (Fedora)\r
2075
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2076
ETag: "238a3c-16ec2-805c5540"\r
2077
Accept-Ranges: bytes\r
2078
Content-Length: 1534\r
2080
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2083
--418470f848b63279b\r
2084
Content-type: text/plain; charset=UTF-8\r
2085
Content-range: bytes 0-254/93890\r
2087
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2088
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2089
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2090
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2091
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2093
--418470f848b63279b\r
2094
Content-type: text/plain; charset=UTF-8\r
2095
Content-range: bytes 1000-2049/93890\r
2098
mbp@sourcefrog.net-20050311063625-07858525021f270b
2099
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2100
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2101
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2102
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2103
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2104
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2105
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2106
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2107
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2108
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2109
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2110
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2111
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2112
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2113
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2114
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2115
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2116
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2117
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2119
--418470f848b63279b--\r
2121
t = self.get_transport()
2122
# Remember that the request is ignored and that the ranges below
2123
# doesn't have to match the canned response.
2124
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2125
# Force consumption of the last bytesrange boundary
2126
t._get_connection().cleanup_pipe()
2127
self.assertEqual(2, len(l))
2128
self.assertActivitiesMatch()
2130
def test_post(self):
2131
self.server.canned_response = b'''HTTP/1.1 200 OK\r
2132
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2133
Server: Apache/2.0.54 (Fedora)\r
2134
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2135
ETag: "56691-23-38e9ae00"\r
2136
Accept-Ranges: bytes\r
2137
Content-Length: 35\r
2139
Content-Type: text/plain; charset=UTF-8\r
2141
lalala whatever as long as itsssss
2143
t = self.get_transport()
2144
# We must send a single line of body bytes, see
2145
# PredefinedRequestHandler._handle_one_request
2146
code, f = t._post(b'abc def end-of-body\n')
2147
self.assertEqual(b'lalala whatever as long as itsssss\n', f.read())
2148
self.assertActivitiesMatch()
2151
class TestActivity(tests.TestCase, TestActivityMixin):
2153
scenarios = multiply_scenarios(
2154
vary_by_http_activity(),
2155
vary_by_http_protocol_version(),
2159
super(TestActivity, self).setUp()
2160
TestActivityMixin.setUp(self)
2163
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2165
# Unlike TestActivity, we are really testing ReportingFileSocket and
2166
# ReportingSocket, so we don't need all the parametrization. Since
2167
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2168
# test them through their use by the transport than directly (that's a
2169
# bit less clean but far more simpler and effective).
2170
_activity_server = ActivityHTTPServer
2171
_protocol_version = 'HTTP/1.1'
2174
super(TestNoReportActivity, self).setUp()
2175
self._transport = HttpTransport
2176
TestActivityMixin.setUp(self)
2178
def assertActivitiesMatch(self):
2179
# Nothing to check here
2183
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2184
"""Test authentication on the redirected http server."""
2186
scenarios = vary_by_http_protocol_version()
2188
_auth_header = 'Authorization'
2189
_password_prompt_prefix = ''
2190
_username_prompt_prefix = ''
2191
_auth_server = http_utils.HTTPBasicAuthServer
2192
_transport = HttpTransport
2195
super(TestAuthOnRedirected, self).setUp()
2196
self.build_tree_contents([('a', b'a'),
2198
('1/a', b'redirected once'),
2200
new_prefix = 'http://%s:%s' % (self.new_server.host,
2201
self.new_server.port)
2202
self.old_server.redirections = [
2203
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2204
self.old_transport = self.get_old_transport()
2205
self.new_server.add_user('joe', 'foo')
2206
cleanup_http_redirection_connections(self)
2208
def create_transport_readonly_server(self):
2209
server = self._auth_server(protocol_version=self._protocol_version)
2210
server._url_protocol = self._url_protocol
2216
def test_auth_on_redirected_via_do_catching_redirections(self):
2217
self.redirections = 0
2219
def redirected(t, exception, redirection_notice):
2220
self.redirections += 1
2221
redirected_t = t._redirected_to(exception.source, exception.target)
2222
self.addCleanup(redirected_t.disconnect)
2225
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2226
self.assertEqual(b'redirected once',
2227
transport.do_catching_redirections(
2228
self.get_a, self.old_transport, redirected).read())
2229
self.assertEqual(1, self.redirections)
2230
# stdin should be empty
2231
self.assertEqual('', ui.ui_factory.stdin.readline())
2232
# stdout should be empty, stderr will contains the prompts
2233
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2235
def test_auth_on_redirected_via_following_redirections(self):
2236
self.new_server.add_user('joe', 'foo')
2237
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2238
t = self.old_transport
2239
new_prefix = 'http://%s:%s' % (self.new_server.host,
2240
self.new_server.port)
2241
self.old_server.redirections = [
2242
('(.*)', r'%s/1\1' % (new_prefix), 301), ]
2245
t.request('GET', t.abspath('a'), retries=3).read())
2246
# stdin should be empty
2247
self.assertEqual('', ui.ui_factory.stdin.readline())
2248
# stdout should be empty, stderr will contains the prompts
2249
self.assertEqual('', ui.ui_factory.stdout.getvalue())
added
removed
breezy/tests/test_http.py
