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
47
remote as _mod_remote,
55
from .scenarios import (
56
load_tests_apply_scenarios,
59
from ..transport import (
63
from ..transport.http import (
69
load_tests = load_tests_apply_scenarios
72
def vary_by_http_client_implementation():
73
"""Test the libraries we can use, currently just urllib."""
74
transport_scenarios = [
75
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
76
_server=http_server.HttpServer_urllib,
77
_url_protocol='http+urllib',)),
79
return transport_scenarios
82
def vary_by_http_protocol_version():
83
"""Test on http/1.0 and 1.1"""
85
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
86
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
90
def vary_by_http_auth_scheme():
92
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
93
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
95
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
97
# Add some attributes common to all scenarios
98
for scenario_id, scenario_dict in scenarios:
99
scenario_dict.update(_auth_header='Authorization',
100
_username_prompt_prefix='',
101
_password_prompt_prefix='')
105
def vary_by_http_proxy_auth_scheme():
107
('proxy-basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
108
('proxy-digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
109
('proxy-basicdigest',
110
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
112
# Add some attributes common to all scenarios
113
for scenario_id, scenario_dict in scenarios:
114
scenario_dict.update(_auth_header='Proxy-Authorization',
115
_username_prompt_prefix='Proxy ',
116
_password_prompt_prefix='Proxy ')
120
def vary_by_http_activity():
121
activity_scenarios = [
122
('urllib,http', dict(_activity_server=ActivityHTTPServer,
123
_transport=_urllib.HttpTransport_urllib,)),
125
if features.HTTPSServerFeature.available():
126
# FIXME: Until we have a better way to handle self-signed certificates
127
# (like allowing them in a test specific authentication.conf for
128
# example), we need some specialized urllib transport for tests.
133
class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
135
def __init__(self, base, _from_transport=None):
136
super(HTTPS_urllib_transport, self).__init__(
137
base, _from_transport=_from_transport,
138
ca_certs=ssl_certs.build_path('ca.crt'))
140
activity_scenarios.append(
141
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
142
_transport=HTTPS_urllib_transport,)),)
143
return activity_scenarios
146
class FakeManager(object):
149
self.credentials = []
151
def add_password(self, realm, host, username, password):
152
self.credentials.append([realm, host, username, password])
155
class RecordingServer(object):
156
"""A fake HTTP server.
158
It records the bytes sent to it, and replies with a 200.
161
def __init__(self, expect_body_tail=None, scheme=''):
164
:type expect_body_tail: str
165
:param expect_body_tail: a reply won't be sent until this string is
168
self._expect_body_tail = expect_body_tail
171
self.received_bytes = ''
175
return '%s://%s:%s/' % (self.scheme, self.host, self.port)
177
def start_server(self):
178
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
179
self._sock.bind(('127.0.0.1', 0))
180
self.host, self.port = self._sock.getsockname()
181
self._ready = threading.Event()
182
self._thread = test_server.TestThread(
183
sync_event=self._ready, target=self._accept_read_and_reply)
185
if 'threads' in tests.selftest_debug_flags:
186
sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
189
def _accept_read_and_reply(self):
192
conn, address = self._sock.accept()
193
if self._expect_body_tail is not None:
194
while not self.received_bytes.endswith(self._expect_body_tail):
195
self.received_bytes += conn.recv(4096)
196
conn.sendall('HTTP/1.1 200 OK\r\n')
200
# The client may have already closed the socket.
203
def stop_server(self):
205
# Issue a fake connection to wake up the server and allow it to
207
fake_conn = osutils.connect_socket((self.host, self.port))
210
# We might have already closed it. We don't care.
215
if 'threads' in tests.selftest_debug_flags:
216
sys.stderr.write('Thread joined: %s\n' % (self._thread.ident,))
219
class TestAuthHeader(tests.TestCase):
221
def parse_header(self, header, auth_handler_class=None):
222
if auth_handler_class is None:
223
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
224
self.auth_handler = auth_handler_class()
225
return self.auth_handler._parse_auth_header(header)
227
def test_empty_header(self):
228
scheme, remainder = self.parse_header('')
229
self.assertEqual('', scheme)
230
self.assertIs(None, remainder)
232
def test_negotiate_header(self):
233
scheme, remainder = self.parse_header('Negotiate')
234
self.assertEqual('negotiate', scheme)
235
self.assertIs(None, remainder)
237
def test_basic_header(self):
238
scheme, remainder = self.parse_header(
239
'Basic realm="Thou should not pass"')
240
self.assertEqual('basic', scheme)
241
self.assertEqual('realm="Thou should not pass"', remainder)
243
def test_build_basic_header_with_long_creds(self):
244
handler = _urllib2_wrappers.BasicAuthHandler()
245
user = 'user' * 10 # length 40
246
password = 'password' * 5 # length 40
247
header = handler.build_auth_header(
248
dict(user=user, password=password), None)
249
# https://bugs.launchpad.net/bzr/+bug/1606203 was caused by incorrectly
250
# creating a header value with an embedded '\n'
251
self.assertFalse('\n' in header)
253
def test_basic_extract_realm(self):
254
scheme, remainder = self.parse_header(
255
'Basic realm="Thou should not pass"',
256
_urllib2_wrappers.BasicAuthHandler)
257
match, realm = self.auth_handler.extract_realm(remainder)
258
self.assertTrue(match is not None)
259
self.assertEqual('Thou should not pass', realm)
261
def test_digest_header(self):
262
scheme, remainder = self.parse_header(
263
'Digest realm="Thou should not pass"')
264
self.assertEqual('digest', scheme)
265
self.assertEqual('realm="Thou should not pass"', remainder)
268
class TestHTTPRangeParsing(tests.TestCase):
271
super(TestHTTPRangeParsing, self).setUp()
272
# We focus on range parsing here and ignore everything else
273
class RequestHandler(http_server.TestingHTTPRequestHandler):
274
def setup(self): pass
275
def handle(self): pass
276
def finish(self): pass
278
self.req_handler = RequestHandler(None, None, None)
280
def assertRanges(self, ranges, header, file_size):
281
self.assertEqual(ranges,
282
self.req_handler._parse_ranges(header, file_size))
284
def test_simple_range(self):
285
self.assertRanges([(0,2)], 'bytes=0-2', 12)
288
self.assertRanges([(8, 11)], 'bytes=-4', 12)
290
def test_tail_bigger_than_file(self):
291
self.assertRanges([(0, 11)], 'bytes=-99', 12)
293
def test_range_without_end(self):
294
self.assertRanges([(4, 11)], 'bytes=4-', 12)
296
def test_invalid_ranges(self):
297
self.assertRanges(None, 'bytes=12-22', 12)
298
self.assertRanges(None, 'bytes=1-3,12-22', 12)
299
self.assertRanges(None, 'bytes=-', 12)
302
class TestHTTPServer(tests.TestCase):
303
"""Test the HTTP servers implementations."""
305
def test_invalid_protocol(self):
306
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
308
protocol_version = 'HTTP/0.1'
310
self.assertRaises(httplib.UnknownProtocol,
311
http_server.HttpServer, BogusRequestHandler)
313
def test_force_invalid_protocol(self):
314
self.assertRaises(httplib.UnknownProtocol,
315
http_server.HttpServer, protocol_version='HTTP/0.1')
317
def test_server_start_and_stop(self):
318
server = http_server.HttpServer()
319
self.addCleanup(server.stop_server)
320
server.start_server()
321
self.assertTrue(server.server is not None)
322
self.assertTrue(server.server.serving is not None)
323
self.assertTrue(server.server.serving)
325
def test_create_http_server_one_zero(self):
326
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
328
protocol_version = 'HTTP/1.0'
330
server = http_server.HttpServer(RequestHandlerOneZero)
331
self.start_server(server)
332
self.assertIsInstance(server.server, http_server.TestingHTTPServer)
334
def test_create_http_server_one_one(self):
335
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
337
protocol_version = 'HTTP/1.1'
339
server = http_server.HttpServer(RequestHandlerOneOne)
340
self.start_server(server)
341
self.assertIsInstance(server.server,
342
http_server.TestingThreadingHTTPServer)
344
def test_create_http_server_force_one_one(self):
345
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
347
protocol_version = 'HTTP/1.0'
349
server = http_server.HttpServer(RequestHandlerOneZero,
350
protocol_version='HTTP/1.1')
351
self.start_server(server)
352
self.assertIsInstance(server.server,
353
http_server.TestingThreadingHTTPServer)
355
def test_create_http_server_force_one_zero(self):
356
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
358
protocol_version = 'HTTP/1.1'
360
server = http_server.HttpServer(RequestHandlerOneOne,
361
protocol_version='HTTP/1.0')
362
self.start_server(server)
363
self.assertIsInstance(server.server,
364
http_server.TestingHTTPServer)
367
class TestHttpTransportUrls(tests.TestCase):
368
"""Test the http urls."""
370
scenarios = vary_by_http_client_implementation()
372
def test_abs_url(self):
373
"""Construction of absolute http URLs"""
374
t = self._transport('http://example.com/bzr/bzr.dev/')
375
eq = self.assertEqualDiff
376
eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
377
eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
378
eq(t.abspath('.bzr'), 'http://example.com/bzr/bzr.dev/.bzr')
379
eq(t.abspath('.bzr/1//2/./3'),
380
'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
382
def test_invalid_http_urls(self):
383
"""Trap invalid construction of urls"""
384
self._transport('http://example.com/bzr/bzr.dev/')
385
self.assertRaises(urlutils.InvalidURL,
387
'http://http://example.com/bzr/bzr.dev/')
389
def test_http_root_urls(self):
390
"""Construction of URLs from server root"""
391
t = self._transport('http://example.com/')
392
eq = self.assertEqualDiff
393
eq(t.abspath('.bzr/tree-version'),
394
'http://example.com/.bzr/tree-version')
396
def test_http_impl_urls(self):
397
"""There are servers which ask for particular clients to connect"""
398
server = self._server()
399
server.start_server()
401
url = server.get_url()
402
self.assertTrue(url.startswith('%s://' % self._url_protocol))
407
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
408
"""Test the http connections."""
410
scenarios = multiply_scenarios(
411
vary_by_http_client_implementation(),
412
vary_by_http_protocol_version(),
416
super(TestHTTPConnections, self).setUp()
417
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
418
transport=self.get_transport())
420
def test_http_has(self):
421
server = self.get_readonly_server()
422
t = self.get_readonly_transport()
423
self.assertEqual(t.has('foo/bar'), True)
424
self.assertEqual(len(server.logs), 1)
425
self.assertContainsRe(server.logs[0],
426
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
428
def test_http_has_not_found(self):
429
server = self.get_readonly_server()
430
t = self.get_readonly_transport()
431
self.assertEqual(t.has('not-found'), False)
432
self.assertContainsRe(server.logs[1],
433
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
435
def test_http_get(self):
436
server = self.get_readonly_server()
437
t = self.get_readonly_transport()
438
fp = t.get('foo/bar')
439
self.assertEqualDiff(
441
'contents of foo/bar\n')
442
self.assertEqual(len(server.logs), 1)
443
self.assertTrue(server.logs[0].find(
444
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
445
% breezy.__version__) > -1)
447
def test_has_on_bogus_host(self):
448
# Get a free address and don't 'accept' on it, so that we
449
# can be sure there is no http handler there, but set a
450
# reasonable timeout to not slow down tests too much.
451
default_timeout = socket.getdefaulttimeout()
453
socket.setdefaulttimeout(2)
455
s.bind(('localhost', 0))
456
t = self._transport('http://%s:%s/' % s.getsockname())
457
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
459
socket.setdefaulttimeout(default_timeout)
462
class TestHttpTransportRegistration(tests.TestCase):
463
"""Test registrations of various http implementations"""
465
scenarios = vary_by_http_client_implementation()
467
def test_http_registered(self):
468
t = transport.get_transport_from_url(
469
'%s://foo.com/' % self._url_protocol)
470
self.assertIsInstance(t, transport.Transport)
471
self.assertIsInstance(t, self._transport)
474
class TestPost(tests.TestCase):
476
scenarios = multiply_scenarios(
477
vary_by_http_client_implementation(),
478
vary_by_http_protocol_version(),
481
def test_post_body_is_received(self):
482
server = RecordingServer(expect_body_tail='end-of-body',
483
scheme=self._url_protocol)
484
self.start_server(server)
485
url = server.get_url()
486
# FIXME: needs a cleanup -- vila 20100611
487
http_transport = transport.get_transport_from_url(url)
488
code, response = http_transport._post('abc def end-of-body')
490
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
491
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
492
self.assertTrue('content-type: application/octet-stream\r'
493
in server.received_bytes.lower())
494
# The transport should not be assuming that the server can accept
495
# chunked encoding the first time it connects, because HTTP/1.1, so we
496
# check for the literal string.
498
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
501
class TestRangeHeader(tests.TestCase):
502
"""Test range_header method"""
504
def check_header(self, value, ranges=[], tail=0):
505
offsets = [ (start, end - start + 1) for start, end in ranges]
506
coalesce = transport.Transport._coalesce_offsets
507
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
508
range_header = http.HttpTransportBase._range_header
509
self.assertEqual(value, range_header(coalesced, tail))
511
def test_range_header_single(self):
512
self.check_header('0-9', ranges=[(0,9)])
513
self.check_header('100-109', ranges=[(100,109)])
515
def test_range_header_tail(self):
516
self.check_header('-10', tail=10)
517
self.check_header('-50', tail=50)
519
def test_range_header_multi(self):
520
self.check_header('0-9,100-200,300-5000',
521
ranges=[(0,9), (100, 200), (300,5000)])
523
def test_range_header_mixed(self):
524
self.check_header('0-9,300-5000,-50',
525
ranges=[(0,9), (300,5000)],
529
class TestSpecificRequestHandler(http_utils.TestCaseWithWebserver):
530
"""Tests a specific request handler.
532
Daughter classes are expected to override _req_handler_class
535
scenarios = multiply_scenarios(
536
vary_by_http_client_implementation(),
537
vary_by_http_protocol_version(),
540
# Provide a useful default
541
_req_handler_class = http_server.TestingHTTPRequestHandler
543
def create_transport_readonly_server(self):
544
server = http_server.HttpServer(self._req_handler_class,
545
protocol_version=self._protocol_version)
546
server._url_protocol = self._url_protocol
550
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
551
"""Whatever request comes in, close the connection"""
553
def _handle_one_request(self):
554
"""Handle a single HTTP request, by abruptly closing the connection"""
555
self.close_connection = 1
558
class TestWallServer(TestSpecificRequestHandler):
559
"""Tests exceptions during the connection phase"""
561
_req_handler_class = WallRequestHandler
563
def test_http_has(self):
564
t = self.get_readonly_transport()
565
# Unfortunately httplib (see HTTPResponse._read_status
566
# for details) make no distinction between a closed
567
# socket and badly formatted status line, so we can't
568
# just test for ConnectionError, we have to test
569
# InvalidHttpResponse too.
570
self.assertRaises((errors.ConnectionError,
571
errors.InvalidHttpResponse),
574
def test_http_get(self):
575
t = self.get_readonly_transport()
576
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
577
errors.InvalidHttpResponse),
581
class BadStatusRequestHandler(http_server.TestingHTTPRequestHandler):
582
"""Whatever request comes in, returns a bad status"""
584
def parse_request(self):
585
"""Fakes handling a single HTTP request, returns a bad status"""
586
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
587
self.send_response(0, "Bad status")
588
self.close_connection = 1
592
class TestBadStatusServer(TestSpecificRequestHandler):
593
"""Tests bad status from server."""
595
_req_handler_class = BadStatusRequestHandler
598
super(TestBadStatusServer, self).setUp()
599
# See https://bugs.launchpad.net/bzr/+bug/1451448 for details.
600
# TD;LR: Running both a TCP client and server in the same process and
601
# thread uncovers a race in python. The fix is to run the server in a
602
# different process. Trying to fix yet another race here is not worth
603
# the effort. -- vila 2015-09-06
604
if 'HTTP/1.0' in self.id():
605
raise tests.TestSkipped(
606
'Client/Server in the same process and thread can hang')
608
def test_http_has(self):
609
t = self.get_readonly_transport()
610
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
611
errors.InvalidHttpResponse),
614
def test_http_get(self):
615
t = self.get_readonly_transport()
616
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
617
errors.InvalidHttpResponse),
621
class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):
622
"""Whatever request comes in, returns an invalid status"""
624
def parse_request(self):
625
"""Fakes handling a single HTTP request, returns a bad status"""
626
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
627
self.wfile.write("Invalid status line\r\n")
628
# If we don't close the connection pycurl will hang. Since this is a
629
# stress test we don't *have* to respect the protocol, but we don't
630
# have to sabotage it too much either.
631
self.close_connection = True
635
class TestInvalidStatusServer(TestBadStatusServer):
636
"""Tests invalid status from server.
638
Both implementations raises the same error as for a bad status.
641
_req_handler_class = InvalidStatusRequestHandler
644
class BadProtocolRequestHandler(http_server.TestingHTTPRequestHandler):
645
"""Whatever request comes in, returns a bad protocol version"""
647
def parse_request(self):
648
"""Fakes handling a single HTTP request, returns a bad status"""
649
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
650
# Returns an invalid protocol version, but curl just
651
# ignores it and those cannot be tested.
652
self.wfile.write("%s %d %s\r\n" % ('HTTP/0.0',
654
'Look at my protocol version'))
658
class TestBadProtocolServer(TestSpecificRequestHandler):
659
"""Tests bad protocol from server."""
661
_req_handler_class = BadProtocolRequestHandler
663
def test_http_has(self):
664
t = self.get_readonly_transport()
665
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
667
def test_http_get(self):
668
t = self.get_readonly_transport()
669
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
672
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
673
"""Whatever request comes in, returns a 403 code"""
675
def parse_request(self):
676
"""Handle a single HTTP request, by replying we cannot handle it"""
677
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
682
class TestForbiddenServer(TestSpecificRequestHandler):
683
"""Tests forbidden server"""
685
_req_handler_class = ForbiddenRequestHandler
687
def test_http_has(self):
688
t = self.get_readonly_transport()
689
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
691
def test_http_get(self):
692
t = self.get_readonly_transport()
693
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
696
class TestRecordingServer(tests.TestCase):
698
def test_create(self):
699
server = RecordingServer(expect_body_tail=None)
700
self.assertEqual('', server.received_bytes)
701
self.assertEqual(None, server.host)
702
self.assertEqual(None, server.port)
704
def test_setUp_and_stop(self):
705
server = RecordingServer(expect_body_tail=None)
706
server.start_server()
708
self.assertNotEqual(None, server.host)
709
self.assertNotEqual(None, server.port)
712
self.assertEqual(None, server.host)
713
self.assertEqual(None, server.port)
715
def test_send_receive_bytes(self):
716
server = RecordingServer(expect_body_tail='c', scheme='http')
717
self.start_server(server)
718
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
719
sock.connect((server.host, server.port))
721
self.assertEqual('HTTP/1.1 200 OK\r\n',
722
osutils.recv_all(sock, 4096))
723
self.assertEqual('abc', server.received_bytes)
726
class TestRangeRequestServer(TestSpecificRequestHandler):
727
"""Tests readv requests against server.
729
We test against default "normal" server.
733
super(TestRangeRequestServer, self).setUp()
734
self.build_tree_contents([('a', '0123456789')],)
736
def test_readv(self):
737
t = self.get_readonly_transport()
738
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
739
self.assertEqual(l[0], (0, '0'))
740
self.assertEqual(l[1], (1, '1'))
741
self.assertEqual(l[2], (3, '34'))
742
self.assertEqual(l[3], (9, '9'))
744
def test_readv_out_of_order(self):
745
t = self.get_readonly_transport()
746
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
747
self.assertEqual(l[0], (1, '1'))
748
self.assertEqual(l[1], (9, '9'))
749
self.assertEqual(l[2], (0, '0'))
750
self.assertEqual(l[3], (3, '34'))
752
def test_readv_invalid_ranges(self):
753
t = self.get_readonly_transport()
755
# This is intentionally reading off the end of the file
756
# since we are sure that it cannot get there
757
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
758
t.readv, 'a', [(1,1), (8,10)])
760
# This is trying to seek past the end of the file, it should
761
# also raise a special error
762
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
763
t.readv, 'a', [(12,2)])
765
def test_readv_multiple_get_requests(self):
766
server = self.get_readonly_server()
767
t = self.get_readonly_transport()
768
# force transport to issue multiple requests
769
t._max_readv_combine = 1
770
t._max_get_ranges = 1
771
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
772
self.assertEqual(l[0], (0, '0'))
773
self.assertEqual(l[1], (1, '1'))
774
self.assertEqual(l[2], (3, '34'))
775
self.assertEqual(l[3], (9, '9'))
776
# The server should have issued 4 requests
777
self.assertEqual(4, server.GET_request_nb)
779
def test_readv_get_max_size(self):
780
server = self.get_readonly_server()
781
t = self.get_readonly_transport()
782
# force transport to issue multiple requests by limiting the number of
783
# bytes by request. Note that this apply to coalesced offsets only, a
784
# single range will keep its size even if bigger than the limit.
786
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
787
self.assertEqual(l[0], (0, '0'))
788
self.assertEqual(l[1], (1, '1'))
789
self.assertEqual(l[2], (2, '2345'))
790
self.assertEqual(l[3], (6, '6789'))
791
# The server should have issued 3 requests
792
self.assertEqual(3, server.GET_request_nb)
794
def test_complete_readv_leave_pipe_clean(self):
795
server = self.get_readonly_server()
796
t = self.get_readonly_transport()
797
# force transport to issue multiple requests
799
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
800
# The server should have issued 3 requests
801
self.assertEqual(3, server.GET_request_nb)
802
self.assertEqual('0123456789', t.get_bytes('a'))
803
self.assertEqual(4, server.GET_request_nb)
805
def test_incomplete_readv_leave_pipe_clean(self):
806
server = self.get_readonly_server()
807
t = self.get_readonly_transport()
808
# force transport to issue multiple requests
810
# Don't collapse readv results into a list so that we leave unread
811
# bytes on the socket
812
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
813
self.assertEqual((0, '0'), next(ireadv))
814
# The server should have issued one request so far
815
self.assertEqual(1, server.GET_request_nb)
816
self.assertEqual('0123456789', t.get_bytes('a'))
817
# get_bytes issued an additional request, the readv pending ones are
819
self.assertEqual(2, server.GET_request_nb)
822
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
823
"""Always reply to range request as if they were single.
825
Don't be explicit about it, just to annoy the clients.
828
def get_multiple_ranges(self, file, file_size, ranges):
829
"""Answer as if it was a single range request and ignores the rest"""
830
(start, end) = ranges[0]
831
return self.get_single_range(file, file_size, start, end)
834
class TestSingleRangeRequestServer(TestRangeRequestServer):
835
"""Test readv against a server which accept only single range requests"""
837
_req_handler_class = SingleRangeRequestHandler
840
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
841
"""Only reply to simple range requests, errors out on multiple"""
843
def get_multiple_ranges(self, file, file_size, ranges):
844
"""Refuses the multiple ranges request"""
847
self.send_error(416, "Requested range not satisfiable")
849
(start, end) = ranges[0]
850
return self.get_single_range(file, file_size, start, end)
853
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
854
"""Test readv against a server which only accept single range requests"""
856
_req_handler_class = SingleOnlyRangeRequestHandler
859
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
860
"""Ignore range requests without notice"""
863
# Update the statistics
864
self.server.test_case_server.GET_request_nb += 1
865
# Just bypass the range handling done by TestingHTTPRequestHandler
866
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
869
class TestNoRangeRequestServer(TestRangeRequestServer):
870
"""Test readv against a server which do not accept range requests"""
872
_req_handler_class = NoRangeRequestHandler
875
class MultipleRangeWithoutContentLengthRequestHandler(
876
http_server.TestingHTTPRequestHandler):
877
"""Reply to multiple range requests without content length header."""
879
def get_multiple_ranges(self, file, file_size, ranges):
880
self.send_response(206)
881
self.send_header('Accept-Ranges', 'bytes')
882
# XXX: this is strange; the 'random' name below seems undefined and
883
# yet the tests pass -- mbp 2010-10-11 bug 658773
884
boundary = "%d" % random.randint(0,0x7FFFFFFF)
885
self.send_header("Content-Type",
886
"multipart/byteranges; boundary=%s" % boundary)
888
for (start, end) in ranges:
889
self.wfile.write("--%s\r\n" % boundary)
890
self.send_header("Content-type", 'application/octet-stream')
891
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
895
self.send_range_content(file, start, end - start + 1)
897
self.wfile.write("--%s\r\n" % boundary)
900
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
902
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
905
class TruncatedMultipleRangeRequestHandler(
906
http_server.TestingHTTPRequestHandler):
907
"""Reply to multiple range requests truncating the last ones.
909
This server generates responses whose Content-Length describes all the
910
ranges, but fail to include the last ones leading to client short reads.
911
This has been observed randomly with lighttpd (bug #179368).
914
_truncated_ranges = 2
916
def get_multiple_ranges(self, file, file_size, ranges):
917
self.send_response(206)
918
self.send_header('Accept-Ranges', 'bytes')
920
self.send_header('Content-Type',
921
'multipart/byteranges; boundary=%s' % boundary)
922
boundary_line = '--%s\r\n' % boundary
923
# Calculate the Content-Length
925
for (start, end) in ranges:
926
content_length += len(boundary_line)
927
content_length += self._header_line_length(
928
'Content-type', 'application/octet-stream')
929
content_length += self._header_line_length(
930
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
931
content_length += len('\r\n') # end headers
932
content_length += end - start # + 1
933
content_length += len(boundary_line)
934
self.send_header('Content-length', content_length)
937
# Send the multipart body
939
for (start, end) in ranges:
940
self.wfile.write(boundary_line)
941
self.send_header('Content-type', 'application/octet-stream')
942
self.send_header('Content-Range', 'bytes %d-%d/%d'
943
% (start, end, file_size))
945
if cur + self._truncated_ranges >= len(ranges):
946
# Abruptly ends the response and close the connection
947
self.close_connection = 1
949
self.send_range_content(file, start, end - start + 1)
952
self.wfile.write(boundary_line)
955
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
957
_req_handler_class = TruncatedMultipleRangeRequestHandler
960
super(TestTruncatedMultipleRangeServer, self).setUp()
961
self.build_tree_contents([('a', '0123456789')],)
963
def test_readv_with_short_reads(self):
964
server = self.get_readonly_server()
965
t = self.get_readonly_transport()
966
# Force separate ranges for each offset
967
t._bytes_to_read_before_seek = 0
968
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
969
self.assertEqual((0, '0'), next(ireadv))
970
self.assertEqual((2, '2'), next(ireadv))
971
# Only one request have been issued so far
972
self.assertEqual(1, server.GET_request_nb)
973
self.assertEqual((4, '45'), next(ireadv))
974
self.assertEqual((9, '9'), next(ireadv))
975
# We issue 3 requests: two multiple (4 ranges, then 2 ranges) then a
977
self.assertEqual(3, server.GET_request_nb)
978
# Finally the client have tried a single range request and stays in
980
self.assertEqual('single', t._range_hint)
983
class TruncatedBeforeBoundaryRequestHandler(
984
http_server.TestingHTTPRequestHandler):
985
"""Truncation before a boundary, like in bug 198646"""
987
_truncated_ranges = 1
989
def get_multiple_ranges(self, file, file_size, ranges):
990
self.send_response(206)
991
self.send_header('Accept-Ranges', 'bytes')
993
self.send_header('Content-Type',
994
'multipart/byteranges; boundary=%s' % boundary)
995
boundary_line = '--%s\r\n' % boundary
996
# Calculate the Content-Length
998
for (start, end) in ranges:
999
content_length += len(boundary_line)
1000
content_length += self._header_line_length(
1001
'Content-type', 'application/octet-stream')
1002
content_length += self._header_line_length(
1003
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
1004
content_length += len('\r\n') # end headers
1005
content_length += end - start # + 1
1006
content_length += len(boundary_line)
1007
self.send_header('Content-length', content_length)
1010
# Send the multipart body
1012
for (start, end) in ranges:
1013
if cur + self._truncated_ranges >= len(ranges):
1014
# Abruptly ends the response and close the connection
1015
self.close_connection = 1
1017
self.wfile.write(boundary_line)
1018
self.send_header('Content-type', 'application/octet-stream')
1019
self.send_header('Content-Range', 'bytes %d-%d/%d'
1020
% (start, end, file_size))
1022
self.send_range_content(file, start, end - start + 1)
1025
self.wfile.write(boundary_line)
1028
class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):
1029
"""Tests the case of bug 198646, disconnecting before a boundary."""
1031
_req_handler_class = TruncatedBeforeBoundaryRequestHandler
1034
super(TestTruncatedBeforeBoundary, self).setUp()
1035
self.build_tree_contents([('a', '0123456789')],)
1037
def test_readv_with_short_reads(self):
1038
server = self.get_readonly_server()
1039
t = self.get_readonly_transport()
1040
# Force separate ranges for each offset
1041
t._bytes_to_read_before_seek = 0
1042
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1043
self.assertEqual((0, '0'), next(ireadv))
1044
self.assertEqual((2, '2'), next(ireadv))
1045
self.assertEqual((4, '45'), next(ireadv))
1046
self.assertEqual((9, '9'), next(ireadv))
1049
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1050
"""Errors out when range specifiers exceed the limit"""
1052
def get_multiple_ranges(self, file, file_size, ranges):
1053
"""Refuses the multiple ranges request"""
1054
tcs = self.server.test_case_server
1055
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1057
# Emulate apache behavior
1058
self.send_error(400, "Bad Request")
1060
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1061
self, file, file_size, ranges)
1064
class LimitedRangeHTTPServer(http_server.HttpServer):
1065
"""An HttpServer erroring out on requests with too much range specifiers"""
1067
def __init__(self, request_handler=LimitedRangeRequestHandler,
1068
protocol_version=None,
1070
http_server.HttpServer.__init__(self, request_handler,
1071
protocol_version=protocol_version)
1072
self.range_limit = range_limit
1075
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1076
"""Tests readv requests against a server erroring out on too much ranges."""
1078
scenarios = multiply_scenarios(
1079
vary_by_http_client_implementation(),
1080
vary_by_http_protocol_version(),
1083
# Requests with more range specifiers will error out
1086
def create_transport_readonly_server(self):
1087
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1088
protocol_version=self._protocol_version)
1091
super(TestLimitedRangeRequestServer, self).setUp()
1092
# We need to manipulate ranges that correspond to real chunks in the
1093
# response, so we build a content appropriately.
1094
filler = ''.join(['abcdefghij' for x in range(102)])
1095
content = ''.join(['%04d' % v + filler for v in range(16)])
1096
self.build_tree_contents([('a', content)],)
1098
def test_few_ranges(self):
1099
t = self.get_readonly_transport()
1100
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1101
self.assertEqual(l[0], (0, '0000'))
1102
self.assertEqual(l[1], (1024, '0001'))
1103
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1105
def test_more_ranges(self):
1106
t = self.get_readonly_transport()
1107
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1108
self.assertEqual(l[0], (0, '0000'))
1109
self.assertEqual(l[1], (1024, '0001'))
1110
self.assertEqual(l[2], (4096, '0004'))
1111
self.assertEqual(l[3], (8192, '0008'))
1112
# The server will refuse to serve the first request (too much ranges),
1113
# a second request will succeed.
1114
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1117
class TestHttpProxyWhiteBox(tests.TestCase):
1118
"""Whitebox test proxy http authorization.
1120
Only the urllib implementation is tested here.
1123
def _proxied_request(self):
1124
handler = _urllib2_wrappers.ProxyHandler()
1125
request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
1126
handler.set_proxy(request, 'http')
1129
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1130
handler = _urllib2_wrappers.ProxyHandler()
1131
self.assertEqual(expected,
1132
handler.evaluate_proxy_bypass(host, no_proxy))
1134
def test_empty_user(self):
1135
self.overrideEnv('http_proxy', 'http://bar.com')
1136
request = self._proxied_request()
1137
self.assertFalse('Proxy-authorization' in request.headers)
1139
def test_user_with_at(self):
1140
self.overrideEnv('http_proxy',
1141
'http://username@domain:password@proxy_host:1234')
1142
request = self._proxied_request()
1143
self.assertFalse('Proxy-authorization' in request.headers)
1145
def test_invalid_proxy(self):
1146
"""A proxy env variable without scheme"""
1147
self.overrideEnv('http_proxy', 'host:1234')
1148
self.assertRaises(urlutils.InvalidURL, self._proxied_request)
1150
def test_evaluate_proxy_bypass_true(self):
1151
"""The host is not proxied"""
1152
self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
1153
self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
1155
def test_evaluate_proxy_bypass_false(self):
1156
"""The host is proxied"""
1157
self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
1159
def test_evaluate_proxy_bypass_unknown(self):
1160
"""The host is not explicitly proxied"""
1161
self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
1162
self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
1164
def test_evaluate_proxy_bypass_empty_entries(self):
1165
"""Ignore empty entries"""
1166
self.assertEvaluateProxyBypass(None, 'example.com', '')
1167
self.assertEvaluateProxyBypass(None, 'example.com', ',')
1168
self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
1171
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
1172
"""Tests proxy server.
1174
Be aware that we do not setup a real proxy here. Instead, we
1175
check that the *connection* goes through the proxy by serving
1176
different content (the faked proxy server append '-proxied'
1180
scenarios = multiply_scenarios(
1181
vary_by_http_client_implementation(),
1182
vary_by_http_protocol_version(),
1185
# FIXME: We don't have an https server available, so we don't
1186
# test https connections. --vila toolongago
1189
super(TestProxyHttpServer, self).setUp()
1190
self.transport_secondary_server = http_utils.ProxyServer
1191
self.build_tree_contents([('foo', 'contents of foo\n'),
1192
('foo-proxied', 'proxied contents of foo\n')])
1193
# Let's setup some attributes for tests
1194
server = self.get_readonly_server()
1195
self.server_host_port = '%s:%d' % (server.host, server.port)
1196
self.no_proxy_host = self.server_host_port
1197
# The secondary server is the proxy
1198
self.proxy_url = self.get_secondary_url()
1200
def assertProxied(self):
1201
t = self.get_readonly_transport()
1202
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1204
def assertNotProxied(self):
1205
t = self.get_readonly_transport()
1206
self.assertEqual('contents of foo\n', t.get('foo').read())
1208
def test_http_proxy(self):
1209
self.overrideEnv('http_proxy', self.proxy_url)
1210
self.assertProxied()
1212
def test_HTTP_PROXY(self):
1213
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1214
self.assertProxied()
1216
def test_all_proxy(self):
1217
self.overrideEnv('all_proxy', self.proxy_url)
1218
self.assertProxied()
1220
def test_ALL_PROXY(self):
1221
self.overrideEnv('ALL_PROXY', self.proxy_url)
1222
self.assertProxied()
1224
def test_http_proxy_with_no_proxy(self):
1225
self.overrideEnv('no_proxy', self.no_proxy_host)
1226
self.overrideEnv('http_proxy', self.proxy_url)
1227
self.assertNotProxied()
1229
def test_HTTP_PROXY_with_NO_PROXY(self):
1230
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1231
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1232
self.assertNotProxied()
1234
def test_all_proxy_with_no_proxy(self):
1235
self.overrideEnv('no_proxy', self.no_proxy_host)
1236
self.overrideEnv('all_proxy', self.proxy_url)
1237
self.assertNotProxied()
1239
def test_ALL_PROXY_with_NO_PROXY(self):
1240
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1241
self.overrideEnv('ALL_PROXY', self.proxy_url)
1242
self.assertNotProxied()
1244
def test_http_proxy_without_scheme(self):
1245
self.overrideEnv('http_proxy', self.server_host_port)
1246
self.assertRaises(urlutils.InvalidURL, self.assertProxied)
1249
class TestRanges(http_utils.TestCaseWithWebserver):
1250
"""Test the Range header in GET methods."""
1252
scenarios = multiply_scenarios(
1253
vary_by_http_client_implementation(),
1254
vary_by_http_protocol_version(),
1258
super(TestRanges, self).setUp()
1259
self.build_tree_contents([('a', '0123456789')],)
1261
def create_transport_readonly_server(self):
1262
return http_server.HttpServer(protocol_version=self._protocol_version)
1264
def _file_contents(self, relpath, ranges):
1265
t = self.get_readonly_transport()
1266
offsets = [ (start, end - start + 1) for start, end in ranges]
1267
coalesce = t._coalesce_offsets
1268
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1269
code, data = t._get(relpath, coalesced)
1270
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1271
for start, end in ranges:
1273
yield data.read(end - start + 1)
1275
def _file_tail(self, relpath, tail_amount):
1276
t = self.get_readonly_transport()
1277
code, data = t._get(relpath, [], tail_amount)
1278
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1279
data.seek(-tail_amount, 2)
1280
return data.read(tail_amount)
1282
def test_range_header(self):
1285
['0', '234'], list(self._file_contents('a', [(0,0), (2,4)])))
1287
def test_range_header_tail(self):
1288
self.assertEqual('789', self._file_tail('a', 3))
1290
def test_syntactically_invalid_range_header(self):
1291
self.assertListRaises(errors.InvalidHttpRange,
1292
self._file_contents, 'a', [(4, 3)])
1294
def test_semantically_invalid_range_header(self):
1295
self.assertListRaises(errors.InvalidHttpRange,
1296
self._file_contents, 'a', [(42, 128)])
1299
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1300
"""Test redirection between http servers."""
1302
scenarios = multiply_scenarios(
1303
vary_by_http_client_implementation(),
1304
vary_by_http_protocol_version(),
1308
super(TestHTTPRedirections, self).setUp()
1309
self.build_tree_contents([('a', '0123456789'),
1311
'# Bazaar revision bundle v0.9\n#\n')
1314
def test_redirected(self):
1315
self.assertRaises(errors.RedirectRequested,
1316
self.get_old_transport().get, 'a')
1317
self.assertEqual('0123456789', self.get_new_transport().get('a').read())
1320
class RedirectedRequest(_urllib2_wrappers.Request):
1321
"""Request following redirections. """
1323
init_orig = _urllib2_wrappers.Request.__init__
1325
def __init__(self, method, url, *args, **kwargs):
1329
# Since the tests using this class will replace
1330
# _urllib2_wrappers.Request, we can't just call the base class __init__
1332
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1333
self.follow_redirections = True
1336
def install_redirected_request(test):
1337
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1340
def cleanup_http_redirection_connections(test):
1341
# Some sockets are opened but never seen by _urllib, so we trap them at
1342
# the _urllib2_wrappers level to be able to clean them up.
1343
def socket_disconnect(sock):
1345
sock.shutdown(socket.SHUT_RDWR)
1347
except socket.error:
1349
def connect(connection):
1350
test.http_connect_orig(connection)
1351
test.addCleanup(socket_disconnect, connection.sock)
1352
test.http_connect_orig = test.overrideAttr(
1353
_urllib2_wrappers.HTTPConnection, 'connect', connect)
1354
def connect(connection):
1355
test.https_connect_orig(connection)
1356
test.addCleanup(socket_disconnect, connection.sock)
1357
test.https_connect_orig = test.overrideAttr(
1358
_urllib2_wrappers.HTTPSConnection, 'connect', connect)
1361
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1362
"""Test redirections.
1364
http implementations do not redirect silently anymore (they
1365
do not redirect at all in fact). The mechanism is still in
1366
place at the _urllib2_wrappers.Request level and these tests
1370
scenarios = multiply_scenarios(
1371
vary_by_http_client_implementation(),
1372
vary_by_http_protocol_version(),
1376
super(TestHTTPSilentRedirections, self).setUp()
1377
install_redirected_request(self)
1378
cleanup_http_redirection_connections(self)
1379
self.build_tree_contents([('a','a'),
1381
('1/a', 'redirected once'),
1383
('2/a', 'redirected twice'),
1385
('3/a', 'redirected thrice'),
1387
('4/a', 'redirected 4 times'),
1389
('5/a', 'redirected 5 times'),
1392
def test_one_redirection(self):
1393
t = self.get_old_transport()
1394
req = RedirectedRequest('GET', t._remote_path('a'))
1395
new_prefix = 'http://%s:%s' % (self.new_server.host,
1396
self.new_server.port)
1397
self.old_server.redirections = \
1398
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1399
self.assertEqual('redirected once', t._perform(req).read())
1401
def test_five_redirections(self):
1402
t = self.get_old_transport()
1403
req = RedirectedRequest('GET', t._remote_path('a'))
1404
old_prefix = 'http://%s:%s' % (self.old_server.host,
1405
self.old_server.port)
1406
new_prefix = 'http://%s:%s' % (self.new_server.host,
1407
self.new_server.port)
1408
self.old_server.redirections = [
1409
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1410
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1411
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1412
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1413
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1415
self.assertEqual('redirected 5 times', t._perform(req).read())
1418
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1419
"""Test transport.do_catching_redirections."""
1421
scenarios = multiply_scenarios(
1422
vary_by_http_client_implementation(),
1423
vary_by_http_protocol_version(),
1427
super(TestDoCatchRedirections, self).setUp()
1428
self.build_tree_contents([('a', '0123456789'),],)
1429
cleanup_http_redirection_connections(self)
1431
self.old_transport = self.get_old_transport()
1436
def test_no_redirection(self):
1437
t = self.get_new_transport()
1439
# We use None for redirected so that we fail if redirected
1440
self.assertEqual('0123456789',
1441
transport.do_catching_redirections(
1442
self.get_a, t, None).read())
1444
def test_one_redirection(self):
1445
self.redirections = 0
1447
def redirected(t, exception, redirection_notice):
1448
self.redirections += 1
1449
redirected_t = t._redirected_to(exception.source, exception.target)
1452
self.assertEqual('0123456789',
1453
transport.do_catching_redirections(
1454
self.get_a, self.old_transport, redirected).read())
1455
self.assertEqual(1, self.redirections)
1457
def test_redirection_loop(self):
1459
def redirected(transport, exception, redirection_notice):
1460
# By using the redirected url as a base dir for the
1461
# *old* transport, we create a loop: a => a/a =>
1463
return self.old_transport.clone(exception.target)
1465
self.assertRaises(errors.TooManyRedirections,
1466
transport.do_catching_redirections,
1467
self.get_a, self.old_transport, redirected)
1470
def _setup_authentication_config(**kwargs):
1471
conf = config.AuthenticationConfig()
1472
conf._get_config().update({'httptest': kwargs})
1476
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1477
"""Unit tests for glue by which urllib2 asks us for authentication"""
1479
def test_get_user_password_without_port(self):
1480
"""We cope if urllib2 doesn't tell us the port.
1482
See https://bugs.launchpad.net/bzr/+bug/654684
1486
_setup_authentication_config(scheme='http', host='localhost',
1487
user=user, password=password)
1488
handler = _urllib2_wrappers.HTTPAuthHandler()
1489
got_pass = handler.get_user_password(dict(
1496
self.assertEqual((user, password), got_pass)
1499
class TestAuth(http_utils.TestCaseWithWebserver):
1500
"""Test authentication scheme"""
1502
scenarios = multiply_scenarios(
1503
vary_by_http_client_implementation(),
1504
vary_by_http_protocol_version(),
1505
vary_by_http_auth_scheme(),
1509
super(TestAuth, self).setUp()
1510
self.server = self.get_readonly_server()
1511
self.build_tree_contents([('a', 'contents of a\n'),
1512
('b', 'contents of b\n'),])
1514
def create_transport_readonly_server(self):
1515
server = self._auth_server(protocol_version=self._protocol_version)
1516
server._url_protocol = self._url_protocol
1519
def get_user_url(self, user, password):
1520
"""Build an url embedding user and password"""
1521
url = '%s://' % self.server._url_protocol
1522
if user is not None:
1524
if password is not None:
1525
url += ':' + password
1527
url += '%s:%s/' % (self.server.host, self.server.port)
1530
def get_user_transport(self, user, password):
1531
t = transport.get_transport_from_url(
1532
self.get_user_url(user, password))
1535
def test_no_user(self):
1536
self.server.add_user('joe', 'foo')
1537
t = self.get_user_transport(None, None)
1538
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1539
# Only one 'Authentication Required' error should occur
1540
self.assertEqual(1, self.server.auth_required_errors)
1542
def test_empty_pass(self):
1543
self.server.add_user('joe', '')
1544
t = self.get_user_transport('joe', '')
1545
self.assertEqual('contents of a\n', t.get('a').read())
1546
# Only one 'Authentication Required' error should occur
1547
self.assertEqual(1, self.server.auth_required_errors)
1549
def test_user_pass(self):
1550
self.server.add_user('joe', 'foo')
1551
t = self.get_user_transport('joe', 'foo')
1552
self.assertEqual('contents of a\n', t.get('a').read())
1553
# Only one 'Authentication Required' error should occur
1554
self.assertEqual(1, self.server.auth_required_errors)
1556
def test_unknown_user(self):
1557
self.server.add_user('joe', 'foo')
1558
t = self.get_user_transport('bill', 'foo')
1559
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1560
# Two 'Authentication Required' errors should occur (the
1561
# initial 'who are you' and 'I don't know you, who are
1563
self.assertEqual(2, self.server.auth_required_errors)
1565
def test_wrong_pass(self):
1566
self.server.add_user('joe', 'foo')
1567
t = self.get_user_transport('joe', 'bar')
1568
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1569
# Two 'Authentication Required' errors should occur (the
1570
# initial 'who are you' and 'this is not you, who are you')
1571
self.assertEqual(2, self.server.auth_required_errors)
1573
def test_prompt_for_username(self):
1574
self.server.add_user('joe', 'foo')
1575
t = self.get_user_transport(None, None)
1576
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
1577
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1578
self.assertEqual('contents of a\n',t.get('a').read())
1579
# stdin should be empty
1580
self.assertEqual('', ui.ui_factory.stdin.readline())
1582
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1583
self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1584
self.assertEqual('', stdout.getvalue())
1585
self._check_password_prompt(t._unqualified_scheme, 'joe',
1588
def test_prompt_for_password(self):
1589
self.server.add_user('joe', 'foo')
1590
t = self.get_user_transport('joe', None)
1591
ui.ui_factory = tests.TestUIFactory(stdin='foo\n')
1592
stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1593
self.assertEqual('contents of a\n', t.get('a').read())
1594
# stdin should be empty
1595
self.assertEqual('', ui.ui_factory.stdin.readline())
1596
self._check_password_prompt(t._unqualified_scheme, 'joe',
1598
self.assertEqual('', stdout.getvalue())
1599
# And we shouldn't prompt again for a different request
1600
# against the same transport.
1601
self.assertEqual('contents of b\n',t.get('b').read())
1603
# And neither against a clone
1604
self.assertEqual('contents of b\n',t2.get('b').read())
1605
# Only one 'Authentication Required' error should occur
1606
self.assertEqual(1, self.server.auth_required_errors)
1608
def _check_password_prompt(self, scheme, user, actual_prompt):
1609
expected_prompt = (self._password_prompt_prefix
1610
+ ("%s %s@%s:%d, Realm: '%s' password: "
1612
user, self.server.host, self.server.port,
1613
self.server.auth_realm)))
1614
self.assertEqual(expected_prompt, actual_prompt)
1616
def _expected_username_prompt(self, scheme):
1617
return (self._username_prompt_prefix
1618
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1619
self.server.host, self.server.port,
1620
self.server.auth_realm))
1622
def test_no_prompt_for_password_when_using_auth_config(self):
1625
stdin_content = 'bar\n' # Not the right password
1626
self.server.add_user(user, password)
1627
t = self.get_user_transport(user, None)
1628
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1629
# Create a minimal config file with the right password
1630
_setup_authentication_config(scheme='http', port=self.server.port,
1631
user=user, password=password)
1632
# Issue a request to the server to connect
1633
self.assertEqual('contents of a\n',t.get('a').read())
1634
# stdin should have been left untouched
1635
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1636
# Only one 'Authentication Required' error should occur
1637
self.assertEqual(1, self.server.auth_required_errors)
1639
def test_changing_nonce(self):
1640
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1641
http_utils.ProxyDigestAuthServer):
1642
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1643
self.server.add_user('joe', 'foo')
1644
t = self.get_user_transport('joe', 'foo')
1645
self.assertEqual('contents of a\n', t.get('a').read())
1646
self.assertEqual('contents of b\n', t.get('b').read())
1647
# Only one 'Authentication Required' error should have
1649
self.assertEqual(1, self.server.auth_required_errors)
1650
# The server invalidates the current nonce
1651
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1652
self.assertEqual('contents of a\n', t.get('a').read())
1653
# Two 'Authentication Required' errors should occur (the
1654
# initial 'who are you' and a second 'who are you' with the new nonce)
1655
self.assertEqual(2, self.server.auth_required_errors)
1657
def test_user_from_auth_conf(self):
1660
self.server.add_user(user, password)
1661
_setup_authentication_config(scheme='http', port=self.server.port,
1662
user=user, password=password)
1663
t = self.get_user_transport(None, None)
1664
# Issue a request to the server to connect
1665
self.assertEqual('contents of a\n', t.get('a').read())
1666
# Only one 'Authentication Required' error should occur
1667
self.assertEqual(1, self.server.auth_required_errors)
1669
def test_no_credential_leaks_in_log(self):
1670
self.overrideAttr(debug, 'debug_flags', {'http'})
1672
password = 'very-sensitive-password'
1673
self.server.add_user(user, password)
1674
t = self.get_user_transport(user, password)
1675
# Capture the debug calls to mutter
1678
lines = args[0] % args[1:]
1679
# Some calls output multiple lines, just split them now since we
1680
# care about a single one later.
1681
self.mutters.extend(lines.splitlines())
1682
self.overrideAttr(trace, 'mutter', mutter)
1683
# Issue a request to the server to connect
1684
self.assertEqual(True, t.has('a'))
1685
# Only one 'Authentication Required' error should occur
1686
self.assertEqual(1, self.server.auth_required_errors)
1687
# Since the authentification succeeded, there should be a corresponding
1689
sent_auth_headers = [line for line in self.mutters
1690
if line.startswith('> %s' % (self._auth_header,))]
1691
self.assertLength(1, sent_auth_headers)
1692
self.assertStartsWith(sent_auth_headers[0],
1693
'> %s: <masked>' % (self._auth_header,))
1696
class TestProxyAuth(TestAuth):
1697
"""Test proxy authentication schemes.
1699
This inherits from TestAuth to tweak the setUp and filter some failing
1703
scenarios = multiply_scenarios(
1704
vary_by_http_client_implementation(),
1705
vary_by_http_protocol_version(),
1706
vary_by_http_proxy_auth_scheme(),
1710
super(TestProxyAuth, self).setUp()
1711
# Override the contents to avoid false positives
1712
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1713
('b', 'not proxied contents of b\n'),
1714
('a-proxied', 'contents of a\n'),
1715
('b-proxied', 'contents of b\n'),
1718
def get_user_transport(self, user, password):
1719
proxy_url = self.get_user_url(user, password)
1720
self.overrideEnv('all_proxy', proxy_url)
1721
return TestAuth.get_user_transport(self, user, password)
1724
class NonClosingBytesIO(io.BytesIO):
1727
"""Ignore and leave file open."""
1730
class SampleSocket(object):
1731
"""A socket-like object for use in testing the HTTP request handler."""
1733
def __init__(self, socket_read_content):
1734
"""Constructs a sample socket.
1736
:param socket_read_content: a byte sequence
1738
self.readfile = io.BytesIO(socket_read_content)
1739
self.writefile = NonClosingBytesIO()
1742
"""Ignore and leave files alone."""
1744
def makefile(self, mode='r', bufsize=None):
1746
return self.readfile
1748
return self.writefile
1751
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1753
scenarios = multiply_scenarios(
1754
vary_by_http_client_implementation(),
1755
vary_by_http_protocol_version(),
1759
super(SmartHTTPTunnellingTest, self).setUp()
1760
# We use the VFS layer as part of HTTP tunnelling tests.
1761
self.overrideEnv('BRZ_NO_SMART_VFS', None)
1762
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1763
self.http_server = self.get_readonly_server()
1765
def create_transport_readonly_server(self):
1766
server = http_utils.HTTPServerWithSmarts(
1767
protocol_version=self._protocol_version)
1768
server._url_protocol = self._url_protocol
1771
def test_open_controldir(self):
1772
branch = self.make_branch('relpath')
1773
url = self.http_server.get_url() + 'relpath'
1774
bd = controldir.ControlDir.open(url)
1775
self.addCleanup(bd.transport.disconnect)
1776
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1778
def test_bulk_data(self):
1779
# We should be able to send and receive bulk data in a single message.
1780
# The 'readv' command in the smart protocol both sends and receives
1781
# bulk data, so we use that.
1782
self.build_tree(['data-file'])
1783
http_transport = transport.get_transport_from_url(
1784
self.http_server.get_url())
1785
medium = http_transport.get_smart_medium()
1786
# Since we provide the medium, the url below will be mostly ignored
1787
# during the test, as long as the path is '/'.
1788
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1791
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1793
def test_http_send_smart_request(self):
1795
post_body = 'hello\n'
1796
expected_reply_body = 'ok\x012\n'
1798
http_transport = transport.get_transport_from_url(
1799
self.http_server.get_url())
1800
medium = http_transport.get_smart_medium()
1801
response = medium.send_http_smart_request(post_body)
1802
reply_body = response.read()
1803
self.assertEqual(expected_reply_body, reply_body)
1805
def test_smart_http_server_post_request_handler(self):
1806
httpd = self.http_server.server
1808
socket = SampleSocket(
1809
'POST /.bzr/smart %s \r\n' % self._protocol_version
1810
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1812
+ 'Content-Length: 6\r\n'
1815
# Beware: the ('localhost', 80) below is the
1816
# client_address parameter, but we don't have one because
1817
# we have defined a socket which is not bound to an
1818
# address. The test framework never uses this client
1819
# address, so far...
1820
request_handler = http_utils.SmartRequestHandler(socket,
1823
response = socket.writefile.getvalue()
1824
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1825
# This includes the end of the HTTP headers, and all the body.
1826
expected_end_of_response = '\r\n\r\nok\x012\n'
1827
self.assertEndsWith(response, expected_end_of_response)
1830
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1831
"""No smart server here request handler."""
1834
self.send_error(403, "Forbidden")
1837
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1838
"""Test smart client behaviour against an http server without smarts."""
1840
_req_handler_class = ForbiddenRequestHandler
1842
def test_probe_smart_server(self):
1843
"""Test error handling against server refusing smart requests."""
1844
t = self.get_readonly_transport()
1845
# No need to build a valid smart request here, the server will not even
1846
# try to interpret it.
1847
self.assertRaises(errors.SmartProtocolError,
1848
t.get_smart_medium().send_http_smart_request,
1852
class Test_redirected_to(tests.TestCase):
1854
scenarios = vary_by_http_client_implementation()
1856
def test_redirected_to_subdir(self):
1857
t = self._transport('http://www.example.com/foo')
1858
r = t._redirected_to('http://www.example.com/foo',
1859
'http://www.example.com/foo/subdir')
1860
self.assertIsInstance(r, type(t))
1861
# Both transports share the some connection
1862
self.assertEqual(t._get_connection(), r._get_connection())
1863
self.assertEqual('http://www.example.com/foo/subdir/', r.base)
1865
def test_redirected_to_self_with_slash(self):
1866
t = self._transport('http://www.example.com/foo')
1867
r = t._redirected_to('http://www.example.com/foo',
1868
'http://www.example.com/foo/')
1869
self.assertIsInstance(r, type(t))
1870
# Both transports share the some connection (one can argue that we
1871
# should return the exact same transport here, but that seems
1873
self.assertEqual(t._get_connection(), r._get_connection())
1875
def test_redirected_to_host(self):
1876
t = self._transport('http://www.example.com/foo')
1877
r = t._redirected_to('http://www.example.com/foo',
1878
'http://foo.example.com/foo/subdir')
1879
self.assertIsInstance(r, type(t))
1880
self.assertEqual('http://foo.example.com/foo/subdir/',
1883
def test_redirected_to_same_host_sibling_protocol(self):
1884
t = self._transport('http://www.example.com/foo')
1885
r = t._redirected_to('http://www.example.com/foo',
1886
'https://www.example.com/foo')
1887
self.assertIsInstance(r, type(t))
1888
self.assertEqual('https://www.example.com/foo/',
1891
def test_redirected_to_same_host_different_protocol(self):
1892
t = self._transport('http://www.example.com/foo')
1893
r = t._redirected_to('http://www.example.com/foo',
1894
'ftp://www.example.com/foo')
1895
self.assertNotEqual(type(r), type(t))
1896
self.assertEqual('ftp://www.example.com/foo/', r.external_url())
1898
def test_redirected_to_same_host_specific_implementation(self):
1899
t = self._transport('http://www.example.com/foo')
1900
r = t._redirected_to('http://www.example.com/foo',
1901
'https+urllib://www.example.com/foo')
1902
self.assertEqual('https://www.example.com/foo/', r.external_url())
1904
def test_redirected_to_different_host_same_user(self):
1905
t = self._transport('http://joe@www.example.com/foo')
1906
r = t._redirected_to('http://www.example.com/foo',
1907
'https://foo.example.com/foo')
1908
self.assertIsInstance(r, type(t))
1909
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
1910
self.assertEqual('https://joe@foo.example.com/foo/', r.external_url())
1913
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1914
"""Request handler for a unique and pre-defined request.
1916
The only thing we care about here is how many bytes travel on the wire. But
1917
since we want to measure it for a real http client, we have to send it
1920
We expect to receive a *single* request nothing more (and we won't even
1921
check what request it is, we just measure the bytes read until an empty
1925
def _handle_one_request(self):
1926
tcs = self.server.test_case_server
1927
requestline = self.rfile.readline()
1928
headers = self.MessageClass(self.rfile, 0)
1929
# We just read: the request, the headers, an empty line indicating the
1930
# end of the headers.
1931
bytes_read = len(requestline)
1932
for line in headers.headers:
1933
bytes_read += len(line)
1934
bytes_read += len('\r\n')
1935
if requestline.startswith('POST'):
1936
# The body should be a single line (or we don't know where it ends
1937
# and we don't want to issue a blocking read)
1938
body = self.rfile.readline()
1939
bytes_read += len(body)
1940
tcs.bytes_read = bytes_read
1942
# We set the bytes written *before* issuing the write, the client is
1943
# supposed to consume every produced byte *before* checking that value.
1945
# Doing the oppposite may lead to test failure: we may be interrupted
1946
# after the write but before updating the value. The client can then
1947
# continue and read the value *before* we can update it. And yes,
1948
# this has been observed -- vila 20090129
1949
tcs.bytes_written = len(tcs.canned_response)
1950
self.wfile.write(tcs.canned_response)
1953
class ActivityServerMixin(object):
1955
def __init__(self, protocol_version):
1956
super(ActivityServerMixin, self).__init__(
1957
request_handler=PredefinedRequestHandler,
1958
protocol_version=protocol_version)
1959
# Bytes read and written by the server
1961
self.bytes_written = 0
1962
self.canned_response = None
1965
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1969
if features.HTTPSServerFeature.available():
1970
from . import https_server
1971
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1975
class TestActivityMixin(object):
1976
"""Test socket activity reporting.
1978
We use a special purpose server to control the bytes sent and received and
1979
be able to predict the activity on the client socket.
1983
self.server = self._activity_server(self._protocol_version)
1984
self.server.start_server()
1985
self.addCleanup(self.server.stop_server)
1986
_activities = {} # Don't close over self and create a cycle
1987
def report_activity(t, bytes, direction):
1988
count = _activities.get(direction, 0)
1990
_activities[direction] = count
1991
self.activities = _activities
1992
# We override at class level because constructors may propagate the
1993
# bound method and render instance overriding ineffective (an
1994
# alternative would be to define a specific ui factory instead...)
1995
self.overrideAttr(self._transport, '_report_activity', report_activity)
1997
def get_transport(self):
1998
t = self._transport(self.server.get_url())
1999
# FIXME: Needs cleanup -- vila 20100611
2002
def assertActivitiesMatch(self):
2003
self.assertEqual(self.server.bytes_read,
2004
self.activities.get('write', 0), 'written bytes')
2005
self.assertEqual(self.server.bytes_written,
2006
self.activities.get('read', 0), 'read bytes')
2009
self.server.canned_response = '''HTTP/1.1 200 OK\r
2010
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2011
Server: Apache/2.0.54 (Fedora)\r
2012
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2013
ETag: "56691-23-38e9ae00"\r
2014
Accept-Ranges: bytes\r
2015
Content-Length: 35\r
2017
Content-Type: text/plain; charset=UTF-8\r
2019
Bazaar-NG meta directory, format 1
2021
t = self.get_transport()
2022
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2023
t.get('foo/bar').read())
2024
self.assertActivitiesMatch()
2027
self.server.canned_response = '''HTTP/1.1 200 OK\r
2028
Server: SimpleHTTP/0.6 Python/2.5.2\r
2029
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2030
Content-type: application/octet-stream\r
2031
Content-Length: 20\r
2032
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2035
t = self.get_transport()
2036
self.assertTrue(t.has('foo/bar'))
2037
self.assertActivitiesMatch()
2039
def test_readv(self):
2040
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2041
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2042
Server: Apache/2.0.54 (Fedora)\r
2043
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2044
ETag: "238a3c-16ec2-805c5540"\r
2045
Accept-Ranges: bytes\r
2046
Content-Length: 1534\r
2048
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2051
--418470f848b63279b\r
2052
Content-type: text/plain; charset=UTF-8\r
2053
Content-range: bytes 0-254/93890\r
2055
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2056
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2057
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2058
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2059
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2061
--418470f848b63279b\r
2062
Content-type: text/plain; charset=UTF-8\r
2063
Content-range: bytes 1000-2049/93890\r
2066
mbp@sourcefrog.net-20050311063625-07858525021f270b
2067
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2068
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2069
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2070
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2071
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2072
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2073
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2074
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2075
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2076
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2077
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2078
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2079
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2080
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2081
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2082
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2083
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2084
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2085
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2087
--418470f848b63279b--\r
2089
t = self.get_transport()
2090
# Remember that the request is ignored and that the ranges below
2091
# doesn't have to match the canned response.
2092
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2093
self.assertEqual(2, len(l))
2094
self.assertActivitiesMatch()
2096
def test_post(self):
2097
self.server.canned_response = '''HTTP/1.1 200 OK\r
2098
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2099
Server: Apache/2.0.54 (Fedora)\r
2100
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2101
ETag: "56691-23-38e9ae00"\r
2102
Accept-Ranges: bytes\r
2103
Content-Length: 35\r
2105
Content-Type: text/plain; charset=UTF-8\r
2107
lalala whatever as long as itsssss
2109
t = self.get_transport()
2110
# We must send a single line of body bytes, see
2111
# PredefinedRequestHandler._handle_one_request
2112
code, f = t._post('abc def end-of-body\n')
2113
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2114
self.assertActivitiesMatch()
2117
class TestActivity(tests.TestCase, TestActivityMixin):
2119
scenarios = multiply_scenarios(
2120
vary_by_http_activity(),
2121
vary_by_http_protocol_version(),
2125
super(TestActivity, self).setUp()
2126
TestActivityMixin.setUp(self)
2129
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2131
# Unlike TestActivity, we are really testing ReportingFileSocket and
2132
# ReportingSocket, so we don't need all the parametrization. Since
2133
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2134
# test them through their use by the transport than directly (that's a
2135
# bit less clean but far more simpler and effective).
2136
_activity_server = ActivityHTTPServer
2137
_protocol_version = 'HTTP/1.1'
2140
super(TestNoReportActivity, self).setUp()
2141
self._transport =_urllib.HttpTransport_urllib
2142
TestActivityMixin.setUp(self)
2144
def assertActivitiesMatch(self):
2145
# Nothing to check here
2149
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2150
"""Test authentication on the redirected http server."""
2152
scenarios = vary_by_http_protocol_version()
2154
_auth_header = 'Authorization'
2155
_password_prompt_prefix = ''
2156
_username_prompt_prefix = ''
2157
_auth_server = http_utils.HTTPBasicAuthServer
2158
_transport = _urllib.HttpTransport_urllib
2161
super(TestAuthOnRedirected, self).setUp()
2162
self.build_tree_contents([('a','a'),
2164
('1/a', 'redirected once'),
2166
new_prefix = 'http://%s:%s' % (self.new_server.host,
2167
self.new_server.port)
2168
self.old_server.redirections = [
2169
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2170
self.old_transport = self.get_old_transport()
2171
self.new_server.add_user('joe', 'foo')
2172
cleanup_http_redirection_connections(self)
2174
def create_transport_readonly_server(self):
2175
server = self._auth_server(protocol_version=self._protocol_version)
2176
server._url_protocol = self._url_protocol
2182
def test_auth_on_redirected_via_do_catching_redirections(self):
2183
self.redirections = 0
2185
def redirected(t, exception, redirection_notice):
2186
self.redirections += 1
2187
redirected_t = t._redirected_to(exception.source, exception.target)
2188
self.addCleanup(redirected_t.disconnect)
2191
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2192
self.assertEqual('redirected once',
2193
transport.do_catching_redirections(
2194
self.get_a, self.old_transport, redirected).read())
2195
self.assertEqual(1, self.redirections)
2196
# stdin should be empty
2197
self.assertEqual('', ui.ui_factory.stdin.readline())
2198
# stdout should be empty, stderr will contains the prompts
2199
self.assertEqual('', ui.ui_factory.stdout.getvalue())
2201
def test_auth_on_redirected_via_following_redirections(self):
2202
self.new_server.add_user('joe', 'foo')
2203
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2204
t = self.old_transport
2205
req = RedirectedRequest('GET', t.abspath('a'))
2206
new_prefix = 'http://%s:%s' % (self.new_server.host,
2207
self.new_server.port)
2208
self.old_server.redirections = [
2209
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2210
self.assertEqual('redirected once', t._perform(req).read())
2211
# stdin should be empty
2212
self.assertEqual('', ui.ui_factory.stdin.readline())
2213
# stdout should be empty, stderr will contains the prompts
2214
self.assertEqual('', ui.ui_factory.stdout.getvalue())
added
removed
breezy/tests/test_http.py
