88
171
self.received_bytes = ''
175
return '%s://%s:%s/' % (self.scheme, self.host, self.port)
177
def start_server(self):
91
178
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
92
179
self._sock.bind(('127.0.0.1', 0))
93
180
self.host, self.port = self._sock.getsockname()
94
181
self._ready = threading.Event()
95
self._thread = threading.Thread(target=self._accept_read_and_reply)
96
self._thread.setDaemon(True)
182
self._thread = test_server.TestThread(
183
sync_event=self._ready, target=self._accept_read_and_reply)
97
184
self._thread.start()
185
if 'threads' in tests.selftest_debug_flags:
186
sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
100
189
def _accept_read_and_reply(self):
101
190
self._sock.listen(1)
102
191
self._ready.set()
103
self._sock.settimeout(5)
105
conn, address = self._sock.accept()
106
# On win32, the accepted connection will be non-blocking to start
107
# with because we're using settimeout.
108
conn.setblocking(True)
192
conn, address = self._sock.accept()
193
if self._expect_body_tail is not None:
109
194
while not self.received_bytes.endswith(self._expect_body_tail):
110
195
self.received_bytes += conn.recv(4096)
111
196
conn.sendall('HTTP/1.1 200 OK\r\n')
112
except socket.timeout:
113
# Make sure the client isn't stuck waiting for us to e.g. accept.
114
198
self._sock.close()
115
199
except socket.error:
116
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))
122
209
except socket.error:
123
210
# We might have already closed it. We don't care.
129
class TestHttpUrls(TestCase):
131
# FIXME: Some of these tests should be done for both
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_basic_extract_realm(self):
244
scheme, remainder = self.parse_header(
245
'Basic realm="Thou should not pass"',
246
_urllib2_wrappers.BasicAuthHandler)
247
match, realm = self.auth_handler.extract_realm(remainder)
248
self.assertTrue(match is not None)
249
self.assertEqual('Thou should not pass', realm)
251
def test_digest_header(self):
252
scheme, remainder = self.parse_header(
253
'Digest realm="Thou should not pass"')
254
self.assertEqual('digest', scheme)
255
self.assertEqual('realm="Thou should not pass"', remainder)
258
class TestHTTPRangeParsing(tests.TestCase):
261
super(TestHTTPRangeParsing, self).setUp()
262
# We focus on range parsing here and ignore everything else
263
class RequestHandler(http_server.TestingHTTPRequestHandler):
264
def setup(self): pass
265
def handle(self): pass
266
def finish(self): pass
268
self.req_handler = RequestHandler(None, None, None)
270
def assertRanges(self, ranges, header, file_size):
271
self.assertEquals(ranges,
272
self.req_handler._parse_ranges(header, file_size))
274
def test_simple_range(self):
275
self.assertRanges([(0,2)], 'bytes=0-2', 12)
278
self.assertRanges([(8, 11)], 'bytes=-4', 12)
280
def test_tail_bigger_than_file(self):
281
self.assertRanges([(0, 11)], 'bytes=-99', 12)
283
def test_range_without_end(self):
284
self.assertRanges([(4, 11)], 'bytes=4-', 12)
286
def test_invalid_ranges(self):
287
self.assertRanges(None, 'bytes=12-22', 12)
288
self.assertRanges(None, 'bytes=1-3,12-22', 12)
289
self.assertRanges(None, 'bytes=-', 12)
292
class TestHTTPServer(tests.TestCase):
293
"""Test the HTTP servers implementations."""
295
def test_invalid_protocol(self):
296
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
298
protocol_version = 'HTTP/0.1'
300
self.assertRaises(httplib.UnknownProtocol,
301
http_server.HttpServer, BogusRequestHandler)
303
def test_force_invalid_protocol(self):
304
self.assertRaises(httplib.UnknownProtocol,
305
http_server.HttpServer, protocol_version='HTTP/0.1')
307
def test_server_start_and_stop(self):
308
server = http_server.HttpServer()
309
self.addCleanup(server.stop_server)
310
server.start_server()
311
self.assertTrue(server.server is not None)
312
self.assertTrue(server.server.serving is not None)
313
self.assertTrue(server.server.serving)
315
def test_create_http_server_one_zero(self):
316
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
318
protocol_version = 'HTTP/1.0'
320
server = http_server.HttpServer(RequestHandlerOneZero)
321
self.start_server(server)
322
self.assertIsInstance(server.server, http_server.TestingHTTPServer)
324
def test_create_http_server_one_one(self):
325
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
327
protocol_version = 'HTTP/1.1'
329
server = http_server.HttpServer(RequestHandlerOneOne)
330
self.start_server(server)
331
self.assertIsInstance(server.server,
332
http_server.TestingThreadingHTTPServer)
334
def test_create_http_server_force_one_one(self):
335
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
337
protocol_version = 'HTTP/1.0'
339
server = http_server.HttpServer(RequestHandlerOneZero,
340
protocol_version='HTTP/1.1')
341
self.start_server(server)
342
self.assertIsInstance(server.server,
343
http_server.TestingThreadingHTTPServer)
345
def test_create_http_server_force_one_zero(self):
346
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
348
protocol_version = 'HTTP/1.1'
350
server = http_server.HttpServer(RequestHandlerOneOne,
351
protocol_version='HTTP/1.0')
352
self.start_server(server)
353
self.assertIsInstance(server.server,
354
http_server.TestingHTTPServer)
357
class TestWithTransport_pycurl(object):
358
"""Test case to inherit from if pycurl is present"""
360
def _get_pycurl_maybe(self):
361
self.requireFeature(features.pycurl)
362
return PyCurlTransport
364
_transport = property(_get_pycurl_maybe)
367
class TestHttpUrls(tests.TestCase):
369
# TODO: This should be moved to authorization tests once they
134
372
def test_url_parsing(self):
135
373
f = FakeManager()
136
url = extract_auth('http://example.com', f)
137
self.assertEquals('http://example.com', url)
138
self.assertEquals(0, len(f.credentials))
139
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
140
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
141
self.assertEquals(1, len(f.credentials))
142
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
374
url = http.extract_auth('http://example.com', f)
375
self.assertEqual('http://example.com', url)
376
self.assertEqual(0, len(f.credentials))
377
url = http.extract_auth(
378
'http://user:pass@example.com/bzr/bzr.dev', f)
379
self.assertEqual('http://example.com/bzr/bzr.dev', url)
380
self.assertEqual(1, len(f.credentials))
381
self.assertEqual([None, 'example.com', 'user', 'pass'],
385
class TestHttpTransportUrls(tests.TestCase):
386
"""Test the http urls."""
388
scenarios = vary_by_http_client_implementation()
145
390
def test_abs_url(self):
146
391
"""Construction of absolute http URLs"""
147
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
392
t = self._transport('http://example.com/bzr/bzr.dev/')
148
393
eq = self.assertEqualDiff
150
'http://bazaar-vcs.org/bzr/bzr.dev')
151
eq(t.abspath('foo/bar'),
152
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
153
eq(t.abspath('.bzr'),
154
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
394
eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')
395
eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')
396
eq(t.abspath('.bzr'), 'http://example.com/bzr/bzr.dev/.bzr')
155
397
eq(t.abspath('.bzr/1//2/./3'),
156
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
398
'http://example.com/bzr/bzr.dev/.bzr/1/2/3')
158
400
def test_invalid_http_urls(self):
159
401
"""Trap invalid construction of urls"""
160
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
161
self.assertRaises(ValueError,
164
t = HttpTransport_urllib('http://http://bazaar-vcs.org/bzr/bzr.dev/')
165
self.assertRaises(errors.InvalidURL, t.has, 'foo/bar')
402
self._transport('http://example.com/bzr/bzr.dev/')
403
self.assertRaises(errors.InvalidURL,
405
'http://http://example.com/bzr/bzr.dev/')
167
407
def test_http_root_urls(self):
168
408
"""Construction of URLs from server root"""
169
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
409
t = self._transport('http://example.com/')
170
410
eq = self.assertEqualDiff
171
411
eq(t.abspath('.bzr/tree-version'),
172
'http://bzr.ozlabs.org/.bzr/tree-version')
412
'http://example.com/.bzr/tree-version')
174
414
def test_http_impl_urls(self):
175
415
"""There are servers which ask for particular clients to connect"""
176
server = HttpServer_PyCurl()
416
server = self._server()
417
server.start_server()
179
419
url = server.get_url()
180
self.assertTrue(url.startswith('http+pycurl://'))
420
self.assertTrue(url.startswith('%s://' % self._url_protocol))
185
class TestHttpConnections(object):
186
"""Test the http connections.
188
This MUST be used by daughter classes that also inherit from
189
TestCaseWithWebserver.
191
We can't inherit directly from TestCaseWithWebserver or the
192
test framework will try to create an instance which cannot
193
run, its implementation being incomplete.
425
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
427
# TODO: This should really be moved into another pycurl
428
# specific test. When https tests will be implemented, take
429
# this one into account.
430
def test_pycurl_without_https_support(self):
431
"""Test that pycurl without SSL do not fail with a traceback.
433
For the purpose of the test, we force pycurl to ignore
434
https by supplying a fake version_info that do not
437
self.requireFeature(features.pycurl)
438
# Import the module locally now that we now it's available.
439
pycurl = features.pycurl.module
441
self.overrideAttr(pycurl, 'version_info',
442
# Fake the pycurl version_info This was taken from
443
# a windows pycurl without SSL (thanks to bialix)
452
('ftp', 'gopher', 'telnet',
453
'dict', 'ldap', 'http', 'file'),
457
self.assertRaises(errors.DependencyNotPresent, self._transport,
458
'https://launchpad.net')
461
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
462
"""Test the http connections."""
464
scenarios = multiply_scenarios(
465
vary_by_http_client_implementation(),
466
vary_by_http_protocol_version(),
197
TestCaseWithWebserver.setUp(self)
198
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
470
http_utils.TestCaseWithWebserver.setUp(self)
471
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
199
472
transport=self.get_transport())
201
474
def test_http_has(self):
202
475
server = self.get_readonly_server()
203
t = self._transport(server.get_url())
476
t = self.get_readonly_transport()
204
477
self.assertEqual(t.has('foo/bar'), True)
205
478
self.assertEqual(len(server.logs), 1)
206
479
self.assertContainsRe(server.logs[0],
595
812
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
596
813
t.readv, 'a', [(12,2)])
815
def test_readv_multiple_get_requests(self):
816
server = self.get_readonly_server()
817
t = self.get_readonly_transport()
818
# force transport to issue multiple requests
819
t._max_readv_combine = 1
820
t._max_get_ranges = 1
821
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
822
self.assertEqual(l[0], (0, '0'))
823
self.assertEqual(l[1], (1, '1'))
824
self.assertEqual(l[2], (3, '34'))
825
self.assertEqual(l[3], (9, '9'))
826
# The server should have issued 4 requests
827
self.assertEqual(4, server.GET_request_nb)
829
def test_readv_get_max_size(self):
830
server = self.get_readonly_server()
831
t = self.get_readonly_transport()
832
# force transport to issue multiple requests by limiting the number of
833
# bytes by request. Note that this apply to coalesced offsets only, a
834
# single range will keep its size even if bigger than the limit.
836
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
837
self.assertEqual(l[0], (0, '0'))
838
self.assertEqual(l[1], (1, '1'))
839
self.assertEqual(l[2], (2, '2345'))
840
self.assertEqual(l[3], (6, '6789'))
841
# The server should have issued 3 requests
842
self.assertEqual(3, server.GET_request_nb)
844
def test_complete_readv_leave_pipe_clean(self):
845
server = self.get_readonly_server()
846
t = self.get_readonly_transport()
847
# force transport to issue multiple requests
849
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
850
# The server should have issued 3 requests
851
self.assertEqual(3, server.GET_request_nb)
852
self.assertEqual('0123456789', t.get_bytes('a'))
853
self.assertEqual(4, server.GET_request_nb)
855
def test_incomplete_readv_leave_pipe_clean(self):
856
server = self.get_readonly_server()
857
t = self.get_readonly_transport()
858
# force transport to issue multiple requests
860
# Don't collapse readv results into a list so that we leave unread
861
# bytes on the socket
862
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
863
self.assertEqual((0, '0'), ireadv.next())
864
# The server should have issued one request so far
865
self.assertEqual(1, server.GET_request_nb)
866
self.assertEqual('0123456789', t.get_bytes('a'))
867
# get_bytes issued an additional request, the readv pending ones are
869
self.assertEqual(2, server.GET_request_nb)
872
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
873
"""Always reply to range request as if they were single.
875
Don't be explicit about it, just to annoy the clients.
878
def get_multiple_ranges(self, file, file_size, ranges):
879
"""Answer as if it was a single range request and ignores the rest"""
880
(start, end) = ranges[0]
881
return self.get_single_range(file, file_size, start, end)
599
884
class TestSingleRangeRequestServer(TestRangeRequestServer):
600
885
"""Test readv against a server which accept only single range requests"""
602
def create_transport_readonly_server(self):
603
return HttpServer(SingleRangeRequestHandler)
606
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
607
TestCaseWithWebserver):
608
"""Tests single range requests accepting server for urllib implementation"""
610
_transport = HttpTransport_urllib
613
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
614
TestSingleRangeRequestServer,
615
TestCaseWithWebserver):
616
"""Tests single range requests accepting server for pycurl implementation"""
887
_req_handler_class = SingleRangeRequestHandler
890
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
891
"""Only reply to simple range requests, errors out on multiple"""
893
def get_multiple_ranges(self, file, file_size, ranges):
894
"""Refuses the multiple ranges request"""
897
self.send_error(416, "Requested range not satisfiable")
899
(start, end) = ranges[0]
900
return self.get_single_range(file, file_size, start, end)
903
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
904
"""Test readv against a server which only accept single range requests"""
906
_req_handler_class = SingleOnlyRangeRequestHandler
909
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
910
"""Ignore range requests without notice"""
913
# Update the statistics
914
self.server.test_case_server.GET_request_nb += 1
915
# Just bypass the range handling done by TestingHTTPRequestHandler
916
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
619
919
class TestNoRangeRequestServer(TestRangeRequestServer):
620
920
"""Test readv against a server which do not accept range requests"""
922
_req_handler_class = NoRangeRequestHandler
925
class MultipleRangeWithoutContentLengthRequestHandler(
926
http_server.TestingHTTPRequestHandler):
927
"""Reply to multiple range requests without content length header."""
929
def get_multiple_ranges(self, file, file_size, ranges):
930
self.send_response(206)
931
self.send_header('Accept-Ranges', 'bytes')
932
# XXX: this is strange; the 'random' name below seems undefined and
933
# yet the tests pass -- mbp 2010-10-11 bug 658773
934
boundary = "%d" % random.randint(0,0x7FFFFFFF)
935
self.send_header("Content-Type",
936
"multipart/byteranges; boundary=%s" % boundary)
938
for (start, end) in ranges:
939
self.wfile.write("--%s\r\n" % boundary)
940
self.send_header("Content-type", 'application/octet-stream')
941
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
945
self.send_range_content(file, start, end - start + 1)
947
self.wfile.write("--%s\r\n" % boundary)
950
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
952
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
955
class TruncatedMultipleRangeRequestHandler(
956
http_server.TestingHTTPRequestHandler):
957
"""Reply to multiple range requests truncating the last ones.
959
This server generates responses whose Content-Length describes all the
960
ranges, but fail to include the last ones leading to client short reads.
961
This has been observed randomly with lighttpd (bug #179368).
964
_truncated_ranges = 2
966
def get_multiple_ranges(self, file, file_size, ranges):
967
self.send_response(206)
968
self.send_header('Accept-Ranges', 'bytes')
970
self.send_header('Content-Type',
971
'multipart/byteranges; boundary=%s' % boundary)
972
boundary_line = '--%s\r\n' % boundary
973
# Calculate the Content-Length
975
for (start, end) in ranges:
976
content_length += len(boundary_line)
977
content_length += self._header_line_length(
978
'Content-type', 'application/octet-stream')
979
content_length += self._header_line_length(
980
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
981
content_length += len('\r\n') # end headers
982
content_length += end - start # + 1
983
content_length += len(boundary_line)
984
self.send_header('Content-length', content_length)
987
# Send the multipart body
989
for (start, end) in ranges:
990
self.wfile.write(boundary_line)
991
self.send_header('Content-type', 'application/octet-stream')
992
self.send_header('Content-Range', 'bytes %d-%d/%d'
993
% (start, end, file_size))
995
if cur + self._truncated_ranges >= len(ranges):
996
# Abruptly ends the response and close the connection
997
self.close_connection = 1
999
self.send_range_content(file, start, end - start + 1)
1002
self.wfile.write(boundary_line)
1005
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
1007
_req_handler_class = TruncatedMultipleRangeRequestHandler
1010
super(TestTruncatedMultipleRangeServer, self).setUp()
1011
self.build_tree_contents([('a', '0123456789')],)
1013
def test_readv_with_short_reads(self):
1014
server = self.get_readonly_server()
1015
t = self.get_readonly_transport()
1016
# Force separate ranges for each offset
1017
t._bytes_to_read_before_seek = 0
1018
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1019
self.assertEqual((0, '0'), ireadv.next())
1020
self.assertEqual((2, '2'), ireadv.next())
1021
if not self._testing_pycurl():
1022
# Only one request have been issued so far (except for pycurl that
1023
# try to read the whole response at once)
1024
self.assertEqual(1, server.GET_request_nb)
1025
self.assertEqual((4, '45'), ireadv.next())
1026
self.assertEqual((9, '9'), ireadv.next())
1027
# Both implementations issue 3 requests but:
1028
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1030
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
1031
self.assertEqual(3, server.GET_request_nb)
1032
# Finally the client have tried a single range request and stays in
1034
self.assertEqual('single', t._range_hint)
1037
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1038
"""Errors out when range specifiers exceed the limit"""
1040
def get_multiple_ranges(self, file, file_size, ranges):
1041
"""Refuses the multiple ranges request"""
1042
tcs = self.server.test_case_server
1043
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1045
# Emulate apache behavior
1046
self.send_error(400, "Bad Request")
1048
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1049
self, file, file_size, ranges)
1052
class LimitedRangeHTTPServer(http_server.HttpServer):
1053
"""An HttpServer erroring out on requests with too much range specifiers"""
1055
def __init__(self, request_handler=LimitedRangeRequestHandler,
1056
protocol_version=None,
1058
http_server.HttpServer.__init__(self, request_handler,
1059
protocol_version=protocol_version)
1060
self.range_limit = range_limit
1063
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1064
"""Tests readv requests against a server erroring out on too much ranges."""
1066
scenarios = multiply_scenarios(
1067
vary_by_http_client_implementation(),
1068
vary_by_http_protocol_version(),
1071
# Requests with more range specifiers will error out
622
1074
def create_transport_readonly_server(self):
623
return HttpServer(NoRangeRequestHandler)
626
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
627
TestCaseWithWebserver):
628
"""Tests range requests refusing server for urllib implementation"""
630
_transport = HttpTransport_urllib
633
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
634
TestNoRangeRequestServer,
635
TestCaseWithWebserver):
636
"""Tests range requests refusing server for pycurl implementation"""
639
class TestHttpProxyWhiteBox(TestCase):
640
"""Whitebox test proxy http authorization."""
1075
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1076
protocol_version=self._protocol_version)
642
1078
def setUp(self):
649
def _set_and_capture_env_var(self, name, new_value):
650
"""Set an environment variable, and reset it when finished."""
651
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
653
def _install_env(self, env):
654
for name, value in env.iteritems():
655
self._set_and_capture_env_var(name, value)
657
def _restore_env(self):
658
for name, value in self._old_env.iteritems():
659
osutils.set_or_unset_env(name, value)
1079
http_utils.TestCaseWithWebserver.setUp(self)
1080
# We need to manipulate ranges that correspond to real chunks in the
1081
# response, so we build a content appropriately.
1082
filler = ''.join(['abcdefghij' for x in range(102)])
1083
content = ''.join(['%04d' % v + filler for v in range(16)])
1084
self.build_tree_contents([('a', content)],)
1086
def test_few_ranges(self):
1087
t = self.get_readonly_transport()
1088
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1089
self.assertEqual(l[0], (0, '0000'))
1090
self.assertEqual(l[1], (1024, '0001'))
1091
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1093
def test_more_ranges(self):
1094
t = self.get_readonly_transport()
1095
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1096
self.assertEqual(l[0], (0, '0000'))
1097
self.assertEqual(l[1], (1024, '0001'))
1098
self.assertEqual(l[2], (4096, '0004'))
1099
self.assertEqual(l[3], (8192, '0008'))
1100
# The server will refuse to serve the first request (too much ranges),
1101
# a second request will succeed.
1102
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1105
class TestHttpProxyWhiteBox(tests.TestCase):
1106
"""Whitebox test proxy http authorization.
1108
Only the urllib implementation is tested here.
661
1111
def _proxied_request(self):
662
from bzrlib.transport.http._urllib2_wrappers import (
667
handler = ProxyHandler()
668
request = Request('GET','http://baz/buzzle')
1112
handler = _urllib2_wrappers.ProxyHandler()
1113
request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
669
1114
handler.set_proxy(request, 'http')
1117
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1118
handler = _urllib2_wrappers.ProxyHandler()
1119
self.assertEquals(expected,
1120
handler.evaluate_proxy_bypass(host, no_proxy))
672
1122
def test_empty_user(self):
673
self._install_env({'http_proxy': 'http://bar.com'})
674
request = self._proxied_request()
675
self.assertFalse(request.headers.has_key('Proxy-authorization'))
677
def test_empty_pass(self):
678
self._install_env({'http_proxy': 'http://joe@bar.com'})
679
request = self._proxied_request()
680
self.assertEqual('Basic ' + 'joe:'.encode('base64').strip(),
681
request.headers['Proxy-authorization'])
682
def test_user_pass(self):
683
self._install_env({'http_proxy': 'http://joe:foo@bar.com'})
684
request = self._proxied_request()
685
self.assertEqual('Basic ' + 'joe:foo'.encode('base64').strip(),
686
request.headers['Proxy-authorization'])
690
class TestProxyHttpServer(object):
1123
self.overrideEnv('http_proxy', 'http://bar.com')
1124
request = self._proxied_request()
1125
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1127
def test_user_with_at(self):
1128
self.overrideEnv('http_proxy',
1129
'http://username@domain:password@proxy_host:1234')
1130
request = self._proxied_request()
1131
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1133
def test_invalid_proxy(self):
1134
"""A proxy env variable without scheme"""
1135
self.overrideEnv('http_proxy', 'host:1234')
1136
self.assertRaises(errors.InvalidURL, self._proxied_request)
1138
def test_evaluate_proxy_bypass_true(self):
1139
"""The host is not proxied"""
1140
self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
1141
self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
1143
def test_evaluate_proxy_bypass_false(self):
1144
"""The host is proxied"""
1145
self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
1147
def test_evaluate_proxy_bypass_unknown(self):
1148
"""The host is not explicitly proxied"""
1149
self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
1150
self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
1152
def test_evaluate_proxy_bypass_empty_entries(self):
1153
"""Ignore empty entries"""
1154
self.assertEvaluateProxyBypass(None, 'example.com', '')
1155
self.assertEvaluateProxyBypass(None, 'example.com', ',')
1156
self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
1159
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
691
1160
"""Tests proxy server.
693
This MUST be used by daughter classes that also inherit from
694
TestCaseWithTwoWebservers.
696
We can't inherit directly from TestCaseWithTwoWebservers or
697
the test framework will try to create an instance which
698
cannot run, its implementation being incomplete.
700
1162
Be aware that we do not setup a real proxy here. Instead, we
701
1163
check that the *connection* goes through the proxy by serving
702
1164
different content (the faked proxy server append '-proxied'
703
1165
to the file names).
1168
scenarios = multiply_scenarios(
1169
vary_by_http_client_implementation(),
1170
vary_by_http_protocol_version(),
706
1173
# FIXME: We don't have an https server available, so we don't
707
# test https connections.
709
# FIXME: Once the test suite is better fitted to test
710
# authorization schemes, test proxy authorizations too (see
1174
# test https connections. --vila toolongago
713
1176
def setUp(self):
714
TestCaseWithTwoWebservers.setUp(self)
1177
super(TestProxyHttpServer, self).setUp()
1178
self.transport_secondary_server = http_utils.ProxyServer
715
1179
self.build_tree_contents([('foo', 'contents of foo\n'),
716
1180
('foo-proxied', 'proxied contents of foo\n')])
717
1181
# Let's setup some attributes for tests
718
self.server = self.get_readonly_server()
719
self.no_proxy_host = 'localhost:%d' % self.server.port
1182
server = self.get_readonly_server()
1183
self.server_host_port = '%s:%d' % (server.host, server.port)
1184
if self._testing_pycurl():
1185
# Oh my ! pycurl does not check for the port as part of
1186
# no_proxy :-( So we just test the host part
1187
self.no_proxy_host = server.host
1189
self.no_proxy_host = self.server_host_port
720
1190
# The secondary server is the proxy
721
self.proxy = self.get_secondary_server()
722
self.proxy_url = self.proxy.get_url()
725
def create_transport_secondary_server(self):
726
"""Creates an http server that will serve files with
727
'-proxied' appended to their names.
729
return HttpServer(FakeProxyRequestHandler)
731
def _set_and_capture_env_var(self, name, new_value):
732
"""Set an environment variable, and reset it when finished."""
733
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
735
def _install_env(self, env):
736
for name, value in env.iteritems():
737
self._set_and_capture_env_var(name, value)
739
def _restore_env(self):
740
for name, value in self._old_env.iteritems():
741
osutils.set_or_unset_env(name, value)
743
def proxied_in_env(self, env):
744
self._install_env(env)
745
url = self.server.get_url()
746
t = self._transport(url)
748
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
752
def not_proxied_in_env(self, env):
753
self._install_env(env)
754
url = self.server.get_url()
755
t = self._transport(url)
757
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
1191
self.proxy_url = self.get_secondary_url()
1193
def _testing_pycurl(self):
1194
# TODO: This is duplicated for lots of the classes in this file
1195
return (features.pycurl.available()
1196
and self._transport == PyCurlTransport)
1198
def assertProxied(self):
1199
t = self.get_readonly_transport()
1200
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1202
def assertNotProxied(self):
1203
t = self.get_readonly_transport()
1204
self.assertEqual('contents of foo\n', t.get('foo').read())
761
1206
def test_http_proxy(self):
762
self.proxied_in_env({'http_proxy': self.proxy_url})
1207
self.overrideEnv('http_proxy', self.proxy_url)
1208
self.assertProxied()
764
1210
def test_HTTP_PROXY(self):
765
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
1211
if self._testing_pycurl():
1212
# pycurl does not check HTTP_PROXY for security reasons
1213
# (for use in a CGI context that we do not care
1214
# about. Should we ?)
1215
raise tests.TestNotApplicable(
1216
'pycurl does not check HTTP_PROXY for security reasons')
1217
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1218
self.assertProxied()
767
1220
def test_all_proxy(self):
768
self.proxied_in_env({'all_proxy': self.proxy_url})
1221
self.overrideEnv('all_proxy', self.proxy_url)
1222
self.assertProxied()
770
1224
def test_ALL_PROXY(self):
771
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1225
self.overrideEnv('ALL_PROXY', self.proxy_url)
1226
self.assertProxied()
773
1228
def test_http_proxy_with_no_proxy(self):
774
self.not_proxied_in_env({'http_proxy': self.proxy_url,
775
'no_proxy': self.no_proxy_host})
1229
self.overrideEnv('no_proxy', self.no_proxy_host)
1230
self.overrideEnv('http_proxy', self.proxy_url)
1231
self.assertNotProxied()
777
1233
def test_HTTP_PROXY_with_NO_PROXY(self):
778
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
779
'NO_PROXY': self.no_proxy_host})
1234
if self._testing_pycurl():
1235
raise tests.TestNotApplicable(
1236
'pycurl does not check HTTP_PROXY for security reasons')
1237
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1238
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1239
self.assertNotProxied()
781
1241
def test_all_proxy_with_no_proxy(self):
782
self.not_proxied_in_env({'all_proxy': self.proxy_url,
783
'no_proxy': self.no_proxy_host})
1242
self.overrideEnv('no_proxy', self.no_proxy_host)
1243
self.overrideEnv('all_proxy', self.proxy_url)
1244
self.assertNotProxied()
785
1246
def test_ALL_PROXY_with_NO_PROXY(self):
786
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
787
'NO_PROXY': self.no_proxy_host})
790
class TestProxyHttpServer_urllib(TestProxyHttpServer,
791
TestCaseWithTwoWebservers):
792
"""Tests proxy server for urllib implementation"""
794
_transport = HttpTransport_urllib
797
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
799
TestCaseWithTwoWebservers):
800
"""Tests proxy server for pycurl implementation"""
803
TestProxyHttpServer.setUp(self)
804
# Oh my ! pycurl does not check for the port as part of
805
# no_proxy :-( So we just test the host part
806
self.no_proxy_host = 'localhost'
808
def test_HTTP_PROXY(self):
809
# pycurl do not check HTTP_PROXY for security reasons
810
# (for use in a CGI context that we do not care
811
# about. Should we ?)
814
def test_HTTP_PROXY_with_NO_PROXY(self):
818
class TestRanges(object):
819
"""Test the Range header in GET methods..
821
This MUST be used by daughter classes that also inherit from
822
TestCaseWithWebserver.
824
We can't inherit directly from TestCaseWithWebserver or the
825
test framework will try to create an instance which cannot
826
run, its implementation being incomplete.
830
TestCaseWithWebserver.setUp(self)
1247
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1248
self.overrideEnv('ALL_PROXY', self.proxy_url)
1249
self.assertNotProxied()
1251
def test_http_proxy_without_scheme(self):
1252
self.overrideEnv('http_proxy', self.server_host_port)
1253
if self._testing_pycurl():
1254
# pycurl *ignores* invalid proxy env variables. If that ever change
1255
# in the future, this test will fail indicating that pycurl do not
1256
# ignore anymore such variables.
1257
self.assertNotProxied()
1259
self.assertRaises(errors.InvalidURL, self.assertProxied)
1262
class TestRanges(http_utils.TestCaseWithWebserver):
1263
"""Test the Range header in GET methods."""
1265
scenarios = multiply_scenarios(
1266
vary_by_http_client_implementation(),
1267
vary_by_http_protocol_version(),
1271
http_utils.TestCaseWithWebserver.setUp(self)
831
1272
self.build_tree_contents([('a', '0123456789')],)
832
server = self.get_readonly_server()
833
self.transport = self._transport(server.get_url())
835
def _file_contents(self, relpath, ranges, tail_amount=0):
836
code, data = self.transport._get(relpath, ranges)
837
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
838
for start, end in ranges:
840
yield data.read(end - start + 1)
1274
def create_transport_readonly_server(self):
1275
return http_server.HttpServer(protocol_version=self._protocol_version)
1277
def _file_contents(self, relpath, ranges):
1278
t = self.get_readonly_transport()
1279
offsets = [ (start, end - start + 1) for start, end in ranges]
1280
coalesce = t._coalesce_offsets
1281
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1282
code, data = t._get(relpath, coalesced)
1283
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1284
for start, end in ranges:
1286
yield data.read(end - start + 1)
842
1288
def _file_tail(self, relpath, tail_amount):
843
code, data = self.transport._get(relpath, [], tail_amount)
844
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
845
data.seek(-tail_amount + 1, 2)
846
return data.read(tail_amount)
1289
t = self.get_readonly_transport()
1290
code, data = t._get(relpath, [], tail_amount)
1291
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1292
data.seek(-tail_amount, 2)
1293
return data.read(tail_amount)
848
1295
def test_range_header(self):
850
1297
map(self.assertEqual,['0', '234'],
851
1298
list(self._file_contents('a', [(0,0), (2,4)])),)
1300
def test_range_header_tail(self):
853
1301
self.assertEqual('789', self._file_tail('a', 3))
854
# Syntactically invalid range
855
self.assertRaises(errors.InvalidRange,
856
self.transport._get, 'a', [(4, 3)])
857
# Semantically invalid range
858
self.assertRaises(errors.InvalidRange,
859
self.transport._get, 'a', [(42, 128)])
862
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
863
"""Test the Range header in GET methods for urllib implementation"""
865
_transport = HttpTransport_urllib
868
class TestRanges_pycurl(TestWithTransport_pycurl,
870
TestCaseWithWebserver):
871
"""Test the Range header in GET methods for pycurl implementation"""
1303
def test_syntactically_invalid_range_header(self):
1304
self.assertListRaises(errors.InvalidHttpRange,
1305
self._file_contents, 'a', [(4, 3)])
1307
def test_semantically_invalid_range_header(self):
1308
self.assertListRaises(errors.InvalidHttpRange,
1309
self._file_contents, 'a', [(42, 128)])
1312
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1313
"""Test redirection between http servers."""
1315
scenarios = multiply_scenarios(
1316
vary_by_http_client_implementation(),
1317
vary_by_http_protocol_version(),
1321
super(TestHTTPRedirections, self).setUp()
1322
self.build_tree_contents([('a', '0123456789'),
1324
'# Bazaar revision bundle v0.9\n#\n')
1327
def test_redirected(self):
1328
self.assertRaises(errors.RedirectRequested,
1329
self.get_old_transport().get, 'a')
1330
self.assertEqual('0123456789', self.get_new_transport().get('a').read())
1333
class RedirectedRequest(_urllib2_wrappers.Request):
1334
"""Request following redirections. """
1336
init_orig = _urllib2_wrappers.Request.__init__
1338
def __init__(self, method, url, *args, **kwargs):
1342
# Since the tests using this class will replace
1343
# _urllib2_wrappers.Request, we can't just call the base class __init__
1345
RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1346
self.follow_redirections = True
1349
def install_redirected_request(test):
1350
test.overrideAttr(_urllib2_wrappers, 'Request', RedirectedRequest)
1353
def cleanup_http_redirection_connections(test):
1354
# Some sockets are opened but never seen by _urllib, so we trap them at
1355
# the _urllib2_wrappers level to be able to clean them up.
1356
def socket_disconnect(sock):
1358
sock.shutdown(socket.SHUT_RDWR)
1360
except socket.error:
1362
def connect(connection):
1363
test.http_connect_orig(connection)
1364
test.addCleanup(socket_disconnect, connection.sock)
1365
test.http_connect_orig = test.overrideAttr(
1366
_urllib2_wrappers.HTTPConnection, 'connect', connect)
1367
def connect(connection):
1368
test.https_connect_orig(connection)
1369
test.addCleanup(socket_disconnect, connection.sock)
1370
test.https_connect_orig = test.overrideAttr(
1371
_urllib2_wrappers.HTTPSConnection, 'connect', connect)
1374
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1375
"""Test redirections.
1377
http implementations do not redirect silently anymore (they
1378
do not redirect at all in fact). The mechanism is still in
1379
place at the _urllib2_wrappers.Request level and these tests
1382
For the pycurl implementation
1383
the redirection have been deleted as we may deprecate pycurl
1384
and I have no place to keep a working implementation.
1388
scenarios = multiply_scenarios(
1389
vary_by_http_client_implementation(),
1390
vary_by_http_protocol_version(),
1394
if (features.pycurl.available()
1395
and self._transport == PyCurlTransport):
1396
raise tests.TestNotApplicable(
1397
"pycurl doesn't redirect silently anymore")
1398
super(TestHTTPSilentRedirections, self).setUp()
1399
install_redirected_request(self)
1400
cleanup_http_redirection_connections(self)
1401
self.build_tree_contents([('a','a'),
1403
('1/a', 'redirected once'),
1405
('2/a', 'redirected twice'),
1407
('3/a', 'redirected thrice'),
1409
('4/a', 'redirected 4 times'),
1411
('5/a', 'redirected 5 times'),
1414
def test_one_redirection(self):
1415
t = self.get_old_transport()
1416
req = RedirectedRequest('GET', t._remote_path('a'))
1417
new_prefix = 'http://%s:%s' % (self.new_server.host,
1418
self.new_server.port)
1419
self.old_server.redirections = \
1420
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1421
self.assertEqual('redirected once', t._perform(req).read())
1423
def test_five_redirections(self):
1424
t = self.get_old_transport()
1425
req = RedirectedRequest('GET', t._remote_path('a'))
1426
old_prefix = 'http://%s:%s' % (self.old_server.host,
1427
self.old_server.port)
1428
new_prefix = 'http://%s:%s' % (self.new_server.host,
1429
self.new_server.port)
1430
self.old_server.redirections = [
1431
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1432
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1433
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1434
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1435
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1437
self.assertEqual('redirected 5 times', t._perform(req).read())
1440
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1441
"""Test transport.do_catching_redirections."""
1443
scenarios = multiply_scenarios(
1444
vary_by_http_client_implementation(),
1445
vary_by_http_protocol_version(),
1449
super(TestDoCatchRedirections, self).setUp()
1450
self.build_tree_contents([('a', '0123456789'),],)
1451
cleanup_http_redirection_connections(self)
1453
self.old_transport = self.get_old_transport()
1458
def test_no_redirection(self):
1459
t = self.get_new_transport()
1461
# We use None for redirected so that we fail if redirected
1462
self.assertEqual('0123456789',
1463
transport.do_catching_redirections(
1464
self.get_a, t, None).read())
1466
def test_one_redirection(self):
1467
self.redirections = 0
1469
def redirected(t, exception, redirection_notice):
1470
self.redirections += 1
1471
redirected_t = t._redirected_to(exception.source, exception.target)
1474
self.assertEqual('0123456789',
1475
transport.do_catching_redirections(
1476
self.get_a, self.old_transport, redirected).read())
1477
self.assertEqual(1, self.redirections)
1479
def test_redirection_loop(self):
1481
def redirected(transport, exception, redirection_notice):
1482
# By using the redirected url as a base dir for the
1483
# *old* transport, we create a loop: a => a/a =>
1485
return self.old_transport.clone(exception.target)
1487
self.assertRaises(errors.TooManyRedirections,
1488
transport.do_catching_redirections,
1489
self.get_a, self.old_transport, redirected)
1492
class TestAuth(http_utils.TestCaseWithWebserver):
1493
"""Test authentication scheme"""
1495
scenarios = multiply_scenarios(
1496
vary_by_http_client_implementation(),
1497
vary_by_http_protocol_version(),
1498
vary_by_http_auth_scheme(),
1501
_auth_header = 'Authorization'
1502
_password_prompt_prefix = ''
1503
_username_prompt_prefix = ''
1508
super(TestAuth, self).setUp()
1509
self.server = self.get_readonly_server()
1510
self.build_tree_contents([('a', 'contents of a\n'),
1511
('b', 'contents of b\n'),])
1513
def create_transport_readonly_server(self):
1514
server = self._auth_server(protocol_version=self._protocol_version)
1515
server._url_protocol = self._url_protocol
1518
def _testing_pycurl(self):
1519
# TODO: This is duplicated for lots of the classes in this file
1520
return (features.pycurl.available()
1521
and self._transport == PyCurlTransport)
1523
def get_user_url(self, user, password):
1524
"""Build an url embedding user and password"""
1525
url = '%s://' % self.server._url_protocol
1526
if user is not None:
1528
if password is not None:
1529
url += ':' + password
1531
url += '%s:%s/' % (self.server.host, self.server.port)
1534
def get_user_transport(self, user, password):
1535
t = transport.get_transport(self.get_user_url(user, password))
1538
def test_no_user(self):
1539
self.server.add_user('joe', 'foo')
1540
t = self.get_user_transport(None, None)
1541
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1542
# Only one 'Authentication Required' error should occur
1543
self.assertEqual(1, self.server.auth_required_errors)
1545
def test_empty_pass(self):
1546
self.server.add_user('joe', '')
1547
t = self.get_user_transport('joe', '')
1548
self.assertEqual('contents of a\n', t.get('a').read())
1549
# Only one 'Authentication Required' error should occur
1550
self.assertEqual(1, self.server.auth_required_errors)
1552
def test_user_pass(self):
1553
self.server.add_user('joe', 'foo')
1554
t = self.get_user_transport('joe', 'foo')
1555
self.assertEqual('contents of a\n', t.get('a').read())
1556
# Only one 'Authentication Required' error should occur
1557
self.assertEqual(1, self.server.auth_required_errors)
1559
def test_unknown_user(self):
1560
self.server.add_user('joe', 'foo')
1561
t = self.get_user_transport('bill', 'foo')
1562
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1563
# Two 'Authentication Required' errors should occur (the
1564
# initial 'who are you' and 'I don't know you, who are
1566
self.assertEqual(2, self.server.auth_required_errors)
1568
def test_wrong_pass(self):
1569
self.server.add_user('joe', 'foo')
1570
t = self.get_user_transport('joe', 'bar')
1571
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1572
# Two 'Authentication Required' errors should occur (the
1573
# initial 'who are you' and 'this is not you, who are you')
1574
self.assertEqual(2, self.server.auth_required_errors)
1576
def test_prompt_for_username(self):
1577
if self._testing_pycurl():
1578
raise tests.TestNotApplicable(
1579
'pycurl cannot prompt, it handles auth by embedding'
1580
' user:pass in urls only')
1582
self.server.add_user('joe', 'foo')
1583
t = self.get_user_transport(None, None)
1584
stdout = tests.StringIOWrapper()
1585
stderr = tests.StringIOWrapper()
1586
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1587
stdout=stdout, stderr=stderr)
1588
self.assertEqual('contents of a\n',t.get('a').read())
1589
# stdin should be empty
1590
self.assertEqual('', ui.ui_factory.stdin.readline())
1592
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1593
self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1594
self.assertEqual('', stdout.getvalue())
1595
self._check_password_prompt(t._unqualified_scheme, 'joe',
1598
def test_prompt_for_password(self):
1599
if self._testing_pycurl():
1600
raise tests.TestNotApplicable(
1601
'pycurl cannot prompt, it handles auth by embedding'
1602
' user:pass in urls only')
1604
self.server.add_user('joe', 'foo')
1605
t = self.get_user_transport('joe', None)
1606
stdout = tests.StringIOWrapper()
1607
stderr = tests.StringIOWrapper()
1608
ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1609
stdout=stdout, stderr=stderr)
1610
self.assertEqual('contents of a\n', t.get('a').read())
1611
# stdin should be empty
1612
self.assertEqual('', ui.ui_factory.stdin.readline())
1613
self._check_password_prompt(t._unqualified_scheme, 'joe',
1615
self.assertEqual('', stdout.getvalue())
1616
# And we shouldn't prompt again for a different request
1617
# against the same transport.
1618
self.assertEqual('contents of b\n',t.get('b').read())
1620
# And neither against a clone
1621
self.assertEqual('contents of b\n',t2.get('b').read())
1622
# Only one 'Authentication Required' error should occur
1623
self.assertEqual(1, self.server.auth_required_errors)
1625
def _check_password_prompt(self, scheme, user, actual_prompt):
1626
expected_prompt = (self._password_prompt_prefix
1627
+ ("%s %s@%s:%d, Realm: '%s' password: "
1629
user, self.server.host, self.server.port,
1630
self.server.auth_realm)))
1631
self.assertEqual(expected_prompt, actual_prompt)
1633
def _expected_username_prompt(self, scheme):
1634
return (self._username_prompt_prefix
1635
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1636
self.server.host, self.server.port,
1637
self.server.auth_realm))
1639
def test_no_prompt_for_password_when_using_auth_config(self):
1640
if self._testing_pycurl():
1641
raise tests.TestNotApplicable(
1642
'pycurl does not support authentication.conf'
1643
' since it cannot prompt')
1647
stdin_content = 'bar\n' # Not the right password
1648
self.server.add_user(user, password)
1649
t = self.get_user_transport(user, None)
1650
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1651
stderr=tests.StringIOWrapper())
1652
# Create a minimal config file with the right password
1653
_setup_authentication_config(
1655
port=self.server.port,
1658
# Issue a request to the server to connect
1659
self.assertEqual('contents of a\n',t.get('a').read())
1660
# stdin should have been left untouched
1661
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1662
# Only one 'Authentication Required' error should occur
1663
self.assertEqual(1, self.server.auth_required_errors)
1665
def test_changing_nonce(self):
1666
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1667
http_utils.ProxyDigestAuthServer):
1668
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1669
if self._testing_pycurl():
1670
raise tests.KnownFailure(
1671
'pycurl does not handle a nonce change')
1672
self.server.add_user('joe', 'foo')
1673
t = self.get_user_transport('joe', 'foo')
1674
self.assertEqual('contents of a\n', t.get('a').read())
1675
self.assertEqual('contents of b\n', t.get('b').read())
1676
# Only one 'Authentication Required' error should have
1678
self.assertEqual(1, self.server.auth_required_errors)
1679
# The server invalidates the current nonce
1680
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1681
self.assertEqual('contents of a\n', t.get('a').read())
1682
# Two 'Authentication Required' errors should occur (the
1683
# initial 'who are you' and a second 'who are you' with the new nonce)
1684
self.assertEqual(2, self.server.auth_required_errors)
1686
def test_user_from_auth_conf(self):
1687
if self._testing_pycurl():
1688
raise tests.TestNotApplicable(
1689
'pycurl does not support authentication.conf')
1692
self.server.add_user(user, password)
1693
_setup_authentication_config(
1695
port=self.server.port,
1698
t = self.get_user_transport(None, None)
1699
# Issue a request to the server to connect
1700
self.assertEqual('contents of a\n', t.get('a').read())
1701
# Only one 'Authentication Required' error should occur
1702
self.assertEqual(1, self.server.auth_required_errors)
1705
def _setup_authentication_config(**kwargs):
1706
conf = config.AuthenticationConfig()
1707
conf._get_config().update({'httptest': kwargs})
1712
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1713
"""Unit tests for glue by which urllib2 asks us for authentication"""
1715
def test_get_user_password_without_port(self):
1716
"""We cope if urllib2 doesn't tell us the port.
1718
See https://bugs.launchpad.net/bzr/+bug/654684
1722
_setup_authentication_config(
1727
handler = _urllib2_wrappers.HTTPAuthHandler()
1728
got_pass = handler.get_user_password(dict(
1735
self.assertEquals((user, password), got_pass)
1738
class TestProxyAuth(TestAuth):
1739
"""Test proxy authentication schemes."""
1741
scenarios = multiply_scenarios(
1742
vary_by_http_client_implementation(),
1743
vary_by_http_protocol_version(),
1744
vary_by_http_proxy_auth_scheme(),
1747
_auth_header = 'Proxy-authorization'
1748
_password_prompt_prefix = 'Proxy '
1749
_username_prompt_prefix = 'Proxy '
1752
super(TestProxyAuth, self).setUp()
1753
# Override the contents to avoid false positives
1754
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1755
('b', 'not proxied contents of b\n'),
1756
('a-proxied', 'contents of a\n'),
1757
('b-proxied', 'contents of b\n'),
1760
def get_user_transport(self, user, password):
1761
self.overrideEnv('all_proxy', self.get_user_url(user, password))
1762
return TestAuth.get_user_transport(self, user, password)
1764
def test_empty_pass(self):
1765
if self._testing_pycurl():
1767
if pycurl.version_info()[1] < '7.16.0':
1768
raise tests.KnownFailure(
1769
'pycurl < 7.16.0 does not handle empty proxy passwords')
1770
super(TestProxyAuth, self).test_empty_pass()
1773
class SampleSocket(object):
1774
"""A socket-like object for use in testing the HTTP request handler."""
1776
def __init__(self, socket_read_content):
1777
"""Constructs a sample socket.
1779
:param socket_read_content: a byte sequence
1781
# Use plain python StringIO so we can monkey-patch the close method to
1782
# not discard the contents.
1783
from StringIO import StringIO
1784
self.readfile = StringIO(socket_read_content)
1785
self.writefile = StringIO()
1786
self.writefile.close = lambda: None
1787
self.close = lambda: None
1789
def makefile(self, mode='r', bufsize=None):
1791
return self.readfile
1793
return self.writefile
1796
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1798
scenarios = multiply_scenarios(
1799
vary_by_http_client_implementation(),
1800
vary_by_http_protocol_version(),
1804
super(SmartHTTPTunnellingTest, self).setUp()
1805
# We use the VFS layer as part of HTTP tunnelling tests.
1806
self.overrideEnv('BZR_NO_SMART_VFS', None)
1807
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1808
self.http_server = self.get_readonly_server()
1810
def create_transport_readonly_server(self):
1811
server = http_utils.HTTPServerWithSmarts(
1812
protocol_version=self._protocol_version)
1813
server._url_protocol = self._url_protocol
1816
def test_open_bzrdir(self):
1817
branch = self.make_branch('relpath')
1818
url = self.http_server.get_url() + 'relpath'
1819
bd = bzrdir.BzrDir.open(url)
1820
self.addCleanup(bd.transport.disconnect)
1821
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1823
def test_bulk_data(self):
1824
# We should be able to send and receive bulk data in a single message.
1825
# The 'readv' command in the smart protocol both sends and receives
1826
# bulk data, so we use that.
1827
self.build_tree(['data-file'])
1828
http_transport = transport.get_transport(self.http_server.get_url())
1829
medium = http_transport.get_smart_medium()
1830
# Since we provide the medium, the url below will be mostly ignored
1831
# during the test, as long as the path is '/'.
1832
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1835
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1837
def test_http_send_smart_request(self):
1839
post_body = 'hello\n'
1840
expected_reply_body = 'ok\x012\n'
1842
http_transport = transport.get_transport(self.http_server.get_url())
1843
medium = http_transport.get_smart_medium()
1844
response = medium.send_http_smart_request(post_body)
1845
reply_body = response.read()
1846
self.assertEqual(expected_reply_body, reply_body)
1848
def test_smart_http_server_post_request_handler(self):
1849
httpd = self.http_server.server
1851
socket = SampleSocket(
1852
'POST /.bzr/smart %s \r\n' % self._protocol_version
1853
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1855
+ 'Content-Length: 6\r\n'
1858
# Beware: the ('localhost', 80) below is the
1859
# client_address parameter, but we don't have one because
1860
# we have defined a socket which is not bound to an
1861
# address. The test framework never uses this client
1862
# address, so far...
1863
request_handler = http_utils.SmartRequestHandler(socket,
1866
response = socket.writefile.getvalue()
1867
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1868
# This includes the end of the HTTP headers, and all the body.
1869
expected_end_of_response = '\r\n\r\nok\x012\n'
1870
self.assertEndsWith(response, expected_end_of_response)
1873
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1874
"""No smart server here request handler."""
1877
self.send_error(403, "Forbidden")
1880
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1881
"""Test smart client behaviour against an http server without smarts."""
1883
_req_handler_class = ForbiddenRequestHandler
1885
def test_probe_smart_server(self):
1886
"""Test error handling against server refusing smart requests."""
1887
t = self.get_readonly_transport()
1888
# No need to build a valid smart request here, the server will not even
1889
# try to interpret it.
1890
self.assertRaises(errors.SmartProtocolError,
1891
t.get_smart_medium().send_http_smart_request,
1895
class Test_redirected_to(tests.TestCase):
1897
scenarios = vary_by_http_client_implementation()
1899
def test_redirected_to_subdir(self):
1900
t = self._transport('http://www.example.com/foo')
1901
r = t._redirected_to('http://www.example.com/foo',
1902
'http://www.example.com/foo/subdir')
1903
self.assertIsInstance(r, type(t))
1904
# Both transports share the some connection
1905
self.assertEqual(t._get_connection(), r._get_connection())
1907
def test_redirected_to_self_with_slash(self):
1908
t = self._transport('http://www.example.com/foo')
1909
r = t._redirected_to('http://www.example.com/foo',
1910
'http://www.example.com/foo/')
1911
self.assertIsInstance(r, type(t))
1912
# Both transports share the some connection (one can argue that we
1913
# should return the exact same transport here, but that seems
1915
self.assertEqual(t._get_connection(), r._get_connection())
1917
def test_redirected_to_host(self):
1918
t = self._transport('http://www.example.com/foo')
1919
r = t._redirected_to('http://www.example.com/foo',
1920
'http://foo.example.com/foo/subdir')
1921
self.assertIsInstance(r, type(t))
1923
def test_redirected_to_same_host_sibling_protocol(self):
1924
t = self._transport('http://www.example.com/foo')
1925
r = t._redirected_to('http://www.example.com/foo',
1926
'https://www.example.com/foo')
1927
self.assertIsInstance(r, type(t))
1929
def test_redirected_to_same_host_different_protocol(self):
1930
t = self._transport('http://www.example.com/foo')
1931
r = t._redirected_to('http://www.example.com/foo',
1932
'ftp://www.example.com/foo')
1933
self.assertNotEquals(type(r), type(t))
1935
def test_redirected_to_different_host_same_user(self):
1936
t = self._transport('http://joe@www.example.com/foo')
1937
r = t._redirected_to('http://www.example.com/foo',
1938
'https://foo.example.com/foo')
1939
self.assertIsInstance(r, type(t))
1940
self.assertEqual(t._user, r._user)
1943
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1944
"""Request handler for a unique and pre-defined request.
1946
The only thing we care about here is how many bytes travel on the wire. But
1947
since we want to measure it for a real http client, we have to send it
1950
We expect to receive a *single* request nothing more (and we won't even
1951
check what request it is, we just measure the bytes read until an empty
1955
def _handle_one_request(self):
1956
tcs = self.server.test_case_server
1957
requestline = self.rfile.readline()
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('\r\n')
1965
if requestline.startswith('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 tests.HTTPSServerFeature.available():
2000
from bzrlib.tests import https_server
2001
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
2005
class TestActivityMixin(object):
2006
"""Test socket activity reporting.
2008
We use a special purpose server to control the bytes sent and received and
2009
be able to predict the activity on the client socket.
2013
tests.TestCase.setUp(self)
2014
self.server = self._activity_server(self._protocol_version)
2015
self.server.start_server()
2016
self.activities = {}
2017
def report_activity(t, bytes, direction):
2018
count = self.activities.get(direction, 0)
2020
self.activities[direction] = count
2022
# We override at class level because constructors may propagate the
2023
# bound method and render instance overriding ineffective (an
2024
# alternative would be to define a specific ui factory instead...)
2025
self.overrideAttr(self._transport, '_report_activity', report_activity)
2026
self.addCleanup(self.server.stop_server)
2028
def get_transport(self):
2029
t = self._transport(self.server.get_url())
2030
# FIXME: Needs cleanup -- vila 20100611
2033
def assertActivitiesMatch(self):
2034
self.assertEqual(self.server.bytes_read,
2035
self.activities.get('write', 0), 'written bytes')
2036
self.assertEqual(self.server.bytes_written,
2037
self.activities.get('read', 0), 'read bytes')
2040
self.server.canned_response = '''HTTP/1.1 200 OK\r
2041
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2042
Server: Apache/2.0.54 (Fedora)\r
2043
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2044
ETag: "56691-23-38e9ae00"\r
2045
Accept-Ranges: bytes\r
2046
Content-Length: 35\r
2048
Content-Type: text/plain; charset=UTF-8\r
2050
Bazaar-NG meta directory, format 1
2052
t = self.get_transport()
2053
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2054
t.get('foo/bar').read())
2055
self.assertActivitiesMatch()
2058
self.server.canned_response = '''HTTP/1.1 200 OK\r
2059
Server: SimpleHTTP/0.6 Python/2.5.2\r
2060
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2061
Content-type: application/octet-stream\r
2062
Content-Length: 20\r
2063
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2066
t = self.get_transport()
2067
self.assertTrue(t.has('foo/bar'))
2068
self.assertActivitiesMatch()
2070
def test_readv(self):
2071
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2072
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2073
Server: Apache/2.0.54 (Fedora)\r
2074
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2075
ETag: "238a3c-16ec2-805c5540"\r
2076
Accept-Ranges: bytes\r
2077
Content-Length: 1534\r
2079
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2082
--418470f848b63279b\r
2083
Content-type: text/plain; charset=UTF-8\r
2084
Content-range: bytes 0-254/93890\r
2086
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2087
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2088
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2089
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2090
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2092
--418470f848b63279b\r
2093
Content-type: text/plain; charset=UTF-8\r
2094
Content-range: bytes 1000-2049/93890\r
2097
mbp@sourcefrog.net-20050311063625-07858525021f270b
2098
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2099
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2100
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2101
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2102
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2103
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2104
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2105
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2106
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2107
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2108
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2109
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2110
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2111
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2112
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2113
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2114
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2115
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2116
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2118
--418470f848b63279b--\r
2120
t = self.get_transport()
2121
# Remember that the request is ignored and that the ranges below
2122
# doesn't have to match the canned response.
2123
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2124
self.assertEqual(2, len(l))
2125
self.assertActivitiesMatch()
2127
def test_post(self):
2128
self.server.canned_response = '''HTTP/1.1 200 OK\r
2129
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2130
Server: Apache/2.0.54 (Fedora)\r
2131
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2132
ETag: "56691-23-38e9ae00"\r
2133
Accept-Ranges: bytes\r
2134
Content-Length: 35\r
2136
Content-Type: text/plain; charset=UTF-8\r
2138
lalala whatever as long as itsssss
2140
t = self.get_transport()
2141
# We must send a single line of body bytes, see
2142
# PredefinedRequestHandler._handle_one_request
2143
code, f = t._post('abc def end-of-body\n')
2144
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2145
self.assertActivitiesMatch()
2148
class TestActivity(tests.TestCase, TestActivityMixin):
2150
scenarios = multiply_scenarios(
2151
vary_by_http_activity(),
2152
vary_by_http_protocol_version(),
2156
TestActivityMixin.setUp(self)
2159
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2161
# Unlike TestActivity, we are really testing ReportingFileSocket and
2162
# ReportingSocket, so we don't need all the parametrization. Since
2163
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2164
# test them through their use by the transport than directly (that's a
2165
# bit less clean but far more simpler and effective).
2166
_activity_server = ActivityHTTPServer
2167
_protocol_version = 'HTTP/1.1'
2170
self._transport =_urllib.HttpTransport_urllib
2171
TestActivityMixin.setUp(self)
2173
def assertActivitiesMatch(self):
2174
# Nothing to check here
2178
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2179
"""Test authentication on the redirected http server."""
2181
scenarios = vary_by_http_protocol_version()
2183
_auth_header = 'Authorization'
2184
_password_prompt_prefix = ''
2185
_username_prompt_prefix = ''
2186
_auth_server = http_utils.HTTPBasicAuthServer
2187
_transport = _urllib.HttpTransport_urllib
2190
super(TestAuthOnRedirected, self).setUp()
2191
self.build_tree_contents([('a','a'),
2193
('1/a', 'redirected once'),
2195
new_prefix = 'http://%s:%s' % (self.new_server.host,
2196
self.new_server.port)
2197
self.old_server.redirections = [
2198
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2199
self.old_transport = self.get_old_transport()
2200
self.new_server.add_user('joe', 'foo')
2201
cleanup_http_redirection_connections(self)
2203
def create_transport_readonly_server(self):
2204
server = self._auth_server(protocol_version=self._protocol_version)
2205
server._url_protocol = self._url_protocol
2211
def test_auth_on_redirected_via_do_catching_redirections(self):
2212
self.redirections = 0
2214
def redirected(t, exception, redirection_notice):
2215
self.redirections += 1
2216
redirected_t = t._redirected_to(exception.source, exception.target)
2217
self.addCleanup(redirected_t.disconnect)
2220
stdout = tests.StringIOWrapper()
2221
stderr = tests.StringIOWrapper()
2222
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2223
stdout=stdout, stderr=stderr)
2224
self.assertEqual('redirected once',
2225
transport.do_catching_redirections(
2226
self.get_a, self.old_transport, redirected).read())
2227
self.assertEqual(1, self.redirections)
2228
# stdin should be empty
2229
self.assertEqual('', ui.ui_factory.stdin.readline())
2230
# stdout should be empty, stderr will contains the prompts
2231
self.assertEqual('', stdout.getvalue())
2233
def test_auth_on_redirected_via_following_redirections(self):
2234
self.new_server.add_user('joe', 'foo')
2235
stdout = tests.StringIOWrapper()
2236
stderr = tests.StringIOWrapper()
2237
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2238
stdout=stdout, stderr=stderr)
2239
t = self.old_transport
2240
req = RedirectedRequest('GET', t.abspath('a'))
2241
new_prefix = 'http://%s:%s' % (self.new_server.host,
2242
self.new_server.port)
2243
self.old_server.redirections = [
2244
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2245
self.assertEqual('redirected once', t._perform(req).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('', stdout.getvalue())
added
removed
bzrlib/tests/test_http.py
