13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
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.
20
23
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
24
# TODO: What about renaming to bzrlib.tests.transport.http ?
23
26
from cStringIO import StringIO
30
import SimpleHTTPServer
31
36
from bzrlib import (
41
remote as _mod_remote,
47
from bzrlib.symbol_versioning import (
37
50
from bzrlib.tests import (
43
from bzrlib.tests.HttpServer import (
48
from bzrlib.tests.HTTPTestUtil import (
49
BadProtocolRequestHandler,
50
BadStatusRequestHandler,
51
ForbiddenRequestHandler,
54
HTTPServerRedirecting,
55
InvalidStatusRequestHandler,
56
LimitedRangeHTTPServer,
57
NoRangeRequestHandler,
59
ProxyDigestAuthServer,
61
SingleRangeRequestHandler,
62
SingleOnlyRangeRequestHandler,
63
TestCaseWithRedirectedWebserver,
64
TestCaseWithTwoWebservers,
65
TestCaseWithWebserver,
68
55
from bzrlib.transport import (
70
do_catching_redirections,
74
59
from bzrlib.transport.http import (
79
from bzrlib.transport.http._urllib import HttpTransport_urllib
80
from bzrlib.transport.http._urllib2_wrappers import (
65
if features.pycurl.available():
66
from bzrlib.transport.http._pycurl import PyCurlTransport
69
def load_tests(standard_tests, module, loader):
70
"""Multiply tests for http clients and protocol versions."""
71
result = loader.suiteClass()
73
# one for each transport implementation
74
t_tests, remaining_tests = tests.split_suite_by_condition(
75
standard_tests, tests.condition_isinstance((
76
TestHttpTransportRegistration,
77
TestHttpTransportUrls,
80
transport_scenarios = [
81
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
82
_server=http_server.HttpServer_urllib,
83
_qualified_prefix='http+urllib',)),
85
if features.pycurl.available():
86
transport_scenarios.append(
87
('pycurl', dict(_transport=PyCurlTransport,
88
_server=http_server.HttpServer_PyCurl,
89
_qualified_prefix='http+pycurl',)))
90
tests.multiply_tests(t_tests, transport_scenarios, result)
92
# each implementation tested with each HTTP version
93
tp_tests, remaining_tests = tests.split_suite_by_condition(
94
remaining_tests, tests.condition_isinstance((
95
SmartHTTPTunnellingTest,
96
TestDoCatchRedirections,
99
TestHTTPSilentRedirections,
100
TestLimitedRangeRequestServer,
104
TestSpecificRequestHandler,
106
protocol_scenarios = [
107
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
108
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
110
tp_scenarios = tests.multiply_scenarios(transport_scenarios,
112
tests.multiply_tests(tp_tests, tp_scenarios, result)
114
# proxy auth: each auth scheme on all http versions on all implementations.
115
tppa_tests, remaining_tests = tests.split_suite_by_condition(
116
remaining_tests, tests.condition_isinstance((
119
proxy_auth_scheme_scenarios = [
120
('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
121
('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
123
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
125
tppa_scenarios = tests.multiply_scenarios(tp_scenarios,
126
proxy_auth_scheme_scenarios)
127
tests.multiply_tests(tppa_tests, tppa_scenarios, result)
129
# auth: each auth scheme on all http versions on all implementations.
130
tpa_tests, remaining_tests = tests.split_suite_by_condition(
131
remaining_tests, tests.condition_isinstance((
134
auth_scheme_scenarios = [
135
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
136
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
138
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
140
tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
141
auth_scheme_scenarios)
142
tests.multiply_tests(tpa_tests, tpa_scenarios, result)
144
# activity: on all http[s] versions on all implementations
145
tpact_tests, remaining_tests = tests.split_suite_by_condition(
146
remaining_tests, tests.condition_isinstance((
149
activity_scenarios = [
150
('urllib,http', dict(_activity_server=ActivityHTTPServer,
151
_transport=_urllib.HttpTransport_urllib,)),
153
if tests.HTTPSServerFeature.available():
154
activity_scenarios.append(
155
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
156
_transport=_urllib.HttpTransport_urllib,)),)
157
if features.pycurl.available():
158
activity_scenarios.append(
159
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
160
_transport=PyCurlTransport,)),)
161
if tests.HTTPSServerFeature.available():
162
from bzrlib.tests import (
165
# FIXME: Until we have a better way to handle self-signed
166
# certificates (like allowing them in a test specific
167
# authentication.conf for example), we need some specialized pycurl
168
# transport for tests.
169
class HTTPS_pycurl_transport(PyCurlTransport):
171
def __init__(self, base, _from_transport=None):
172
super(HTTPS_pycurl_transport, self).__init__(
173
base, _from_transport)
174
self.cabundle = str(ssl_certs.build_path('ca.crt'))
176
activity_scenarios.append(
177
('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
178
_transport=HTTPS_pycurl_transport,)),)
180
tpact_scenarios = tests.multiply_scenarios(activity_scenarios,
182
tests.multiply_tests(tpact_tests, tpact_scenarios, result)
184
# No parametrization for the remaining tests
185
result.addTests(remaining_tests)
87
190
class FakeManager(object):
260
class TestAuthHeader(tests.TestCase):
262
def parse_header(self, header, auth_handler_class=None):
263
if auth_handler_class is None:
264
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
265
self.auth_handler = auth_handler_class()
266
return self.auth_handler._parse_auth_header(header)
268
def test_empty_header(self):
269
scheme, remainder = self.parse_header('')
270
self.assertEqual('', scheme)
271
self.assertIs(None, remainder)
273
def test_negotiate_header(self):
274
scheme, remainder = self.parse_header('Negotiate')
275
self.assertEqual('negotiate', scheme)
276
self.assertIs(None, remainder)
278
def test_basic_header(self):
279
scheme, remainder = self.parse_header(
280
'Basic realm="Thou should not pass"')
281
self.assertEqual('basic', scheme)
282
self.assertEqual('realm="Thou should not pass"', remainder)
284
def test_basic_extract_realm(self):
285
scheme, remainder = self.parse_header(
286
'Basic realm="Thou should not pass"',
287
_urllib2_wrappers.BasicAuthHandler)
288
match, realm = self.auth_handler.extract_realm(remainder)
289
self.assertTrue(match is not None)
290
self.assertEqual('Thou should not pass', realm)
292
def test_digest_header(self):
293
scheme, remainder = self.parse_header(
294
'Digest realm="Thou should not pass"')
295
self.assertEqual('digest', scheme)
296
self.assertEqual('realm="Thou should not pass"', remainder)
299
class TestHTTPServer(tests.TestCase):
300
"""Test the HTTP servers implementations."""
302
def test_invalid_protocol(self):
303
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
305
protocol_version = 'HTTP/0.1'
307
server = http_server.HttpServer(BogusRequestHandler)
309
self.assertRaises(httplib.UnknownProtocol, server.start_server)
312
self.fail('HTTP Server creation did not raise UnknownProtocol')
314
def test_force_invalid_protocol(self):
315
server = http_server.HttpServer(protocol_version='HTTP/0.1')
317
self.assertRaises(httplib.UnknownProtocol, server.start_server)
320
self.fail('HTTP Server creation did not raise UnknownProtocol')
322
def test_server_start_and_stop(self):
323
server = http_server.HttpServer()
324
server.start_server()
326
self.assertTrue(server._http_running)
329
self.assertFalse(server._http_running)
331
def test_create_http_server_one_zero(self):
332
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
334
protocol_version = 'HTTP/1.0'
336
server = http_server.HttpServer(RequestHandlerOneZero)
337
self.start_server(server)
338
self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
340
def test_create_http_server_one_one(self):
341
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
343
protocol_version = 'HTTP/1.1'
345
server = http_server.HttpServer(RequestHandlerOneOne)
346
self.start_server(server)
347
self.assertIsInstance(server._httpd,
348
http_server.TestingThreadingHTTPServer)
350
def test_create_http_server_force_one_one(self):
351
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
353
protocol_version = 'HTTP/1.0'
355
server = http_server.HttpServer(RequestHandlerOneZero,
356
protocol_version='HTTP/1.1')
357
self.start_server(server)
358
self.assertIsInstance(server._httpd,
359
http_server.TestingThreadingHTTPServer)
361
def test_create_http_server_force_one_zero(self):
362
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
364
protocol_version = 'HTTP/1.1'
366
server = http_server.HttpServer(RequestHandlerOneOne,
367
protocol_version='HTTP/1.0')
368
self.start_server(server)
369
self.assertIsInstance(server._httpd,
370
http_server.TestingHTTPServer)
153
373
class TestWithTransport_pycurl(object):
154
374
"""Test case to inherit from if pycurl is present"""
156
376
def _get_pycurl_maybe(self):
158
from bzrlib.transport.http._pycurl import PyCurlTransport
159
return PyCurlTransport
160
except errors.DependencyNotPresent:
161
raise TestSkipped('pycurl not present')
377
self.requireFeature(features.pycurl)
378
return PyCurlTransport
163
380
_transport = property(_get_pycurl_maybe)
166
class TestHttpUrls(TestCase):
383
class TestHttpUrls(tests.TestCase):
168
385
# TODO: This should be moved to authorization tests once they
171
388
def test_url_parsing(self):
172
389
f = FakeManager()
173
url = extract_auth('http://example.com', f)
174
self.assertEquals('http://example.com', url)
175
self.assertEquals(0, len(f.credentials))
176
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
177
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
178
self.assertEquals(1, len(f.credentials))
179
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
183
class TestHttpTransportUrls(object):
184
"""Test the http urls.
186
This MUST be used by daughter classes that also inherit from
189
We can't inherit directly from TestCase or the
190
test framework will try to create an instance which cannot
191
run, its implementation being incomplete.
390
url = http.extract_auth('http://example.com', f)
391
self.assertEqual('http://example.com', url)
392
self.assertEqual(0, len(f.credentials))
393
url = http.extract_auth(
394
'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
395
self.assertEqual('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
396
self.assertEqual(1, len(f.credentials))
397
self.assertEqual([None, 'www.bazaar-vcs.org', 'user', 'pass'],
401
class TestHttpTransportUrls(tests.TestCase):
402
"""Test the http urls."""
194
404
def test_abs_url(self):
195
405
"""Construction of absolute http URLs"""
653
827
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
654
828
t.readv, 'a', [(12,2)])
830
def test_readv_multiple_get_requests(self):
831
server = self.get_readonly_server()
832
t = self._transport(server.get_url())
833
# force transport to issue multiple requests
834
t._max_readv_combine = 1
835
t._max_get_ranges = 1
836
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
837
self.assertEqual(l[0], (0, '0'))
838
self.assertEqual(l[1], (1, '1'))
839
self.assertEqual(l[2], (3, '34'))
840
self.assertEqual(l[3], (9, '9'))
841
# The server should have issued 4 requests
842
self.assertEqual(4, server.GET_request_nb)
844
def test_readv_get_max_size(self):
845
server = self.get_readonly_server()
846
t = self._transport(server.get_url())
847
# force transport to issue multiple requests by limiting the number of
848
# bytes by request. Note that this apply to coalesced offsets only, a
849
# single range will keep its size even if bigger than the limit.
851
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
852
self.assertEqual(l[0], (0, '0'))
853
self.assertEqual(l[1], (1, '1'))
854
self.assertEqual(l[2], (2, '2345'))
855
self.assertEqual(l[3], (6, '6789'))
856
# The server should have issued 3 requests
857
self.assertEqual(3, server.GET_request_nb)
859
def test_complete_readv_leave_pipe_clean(self):
860
server = self.get_readonly_server()
861
t = self._transport(server.get_url())
862
# force transport to issue multiple requests
864
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
865
# The server should have issued 3 requests
866
self.assertEqual(3, server.GET_request_nb)
867
self.assertEqual('0123456789', t.get_bytes('a'))
868
self.assertEqual(4, server.GET_request_nb)
870
def test_incomplete_readv_leave_pipe_clean(self):
871
server = self.get_readonly_server()
872
t = self._transport(server.get_url())
873
# force transport to issue multiple requests
875
# Don't collapse readv results into a list so that we leave unread
876
# bytes on the socket
877
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
878
self.assertEqual((0, '0'), ireadv.next())
879
# The server should have issued one request so far
880
self.assertEqual(1, server.GET_request_nb)
881
self.assertEqual('0123456789', t.get_bytes('a'))
882
# get_bytes issued an additional request, the readv pending ones are
884
self.assertEqual(2, server.GET_request_nb)
887
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
888
"""Always reply to range request as if they were single.
890
Don't be explicit about it, just to annoy the clients.
893
def get_multiple_ranges(self, file, file_size, ranges):
894
"""Answer as if it was a single range request and ignores the rest"""
895
(start, end) = ranges[0]
896
return self.get_single_range(file, file_size, start, end)
657
899
class TestSingleRangeRequestServer(TestRangeRequestServer):
658
900
"""Test readv against a server which accept only single range requests"""
660
def create_transport_readonly_server(self):
661
return HttpServer(SingleRangeRequestHandler)
664
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
665
TestCaseWithWebserver):
666
"""Tests single range requests accepting server for urllib implementation"""
668
_transport = HttpTransport_urllib
671
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
672
TestSingleRangeRequestServer,
673
TestCaseWithWebserver):
674
"""Tests single range requests accepting server for pycurl implementation"""
902
_req_handler_class = SingleRangeRequestHandler
905
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
906
"""Only reply to simple range requests, errors out on multiple"""
908
def get_multiple_ranges(self, file, file_size, ranges):
909
"""Refuses the multiple ranges request"""
912
self.send_error(416, "Requested range not satisfiable")
914
(start, end) = ranges[0]
915
return self.get_single_range(file, file_size, start, end)
677
918
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
678
919
"""Test readv against a server which only accept single range requests"""
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleOnlyRangeRequestHandler)
684
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
688
_transport = HttpTransport_urllib
691
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleOnlyRangeRequestServer,
693
TestCaseWithWebserver):
694
"""Tests single range requests accepting server for pycurl implementation"""
921
_req_handler_class = SingleOnlyRangeRequestHandler
924
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
925
"""Ignore range requests without notice"""
928
# Update the statistics
929
self.server.test_case_server.GET_request_nb += 1
930
# Just bypass the range handling done by TestingHTTPRequestHandler
931
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
697
934
class TestNoRangeRequestServer(TestRangeRequestServer):
698
935
"""Test readv against a server which do not accept range requests"""
700
def create_transport_readonly_server(self):
701
return HttpServer(NoRangeRequestHandler)
704
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests range requests refusing server for urllib implementation"""
708
_transport = HttpTransport_urllib
711
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestNoRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests range requests refusing server for pycurl implementation"""
717
class TestLimitedRangeRequestServer(object):
718
"""Tests readv requests against server that errors out on too much ranges.
720
This MUST be used by daughter classes that also inherit from
721
TestCaseWithWebserver.
723
We can't inherit directly from TestCaseWithWebserver or the
724
test framework will try to create an instance which cannot
725
run, its implementation being incomplete.
937
_req_handler_class = NoRangeRequestHandler
940
class MultipleRangeWithoutContentLengthRequestHandler(
941
http_server.TestingHTTPRequestHandler):
942
"""Reply to multiple range requests without content length header."""
944
def get_multiple_ranges(self, file, file_size, ranges):
945
self.send_response(206)
946
self.send_header('Accept-Ranges', 'bytes')
947
boundary = "%d" % random.randint(0,0x7FFFFFFF)
948
self.send_header("Content-Type",
949
"multipart/byteranges; boundary=%s" % boundary)
951
for (start, end) in ranges:
952
self.wfile.write("--%s\r\n" % boundary)
953
self.send_header("Content-type", 'application/octet-stream')
954
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
958
self.send_range_content(file, start, end - start + 1)
960
self.wfile.write("--%s\r\n" % boundary)
963
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
965
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
968
class TruncatedMultipleRangeRequestHandler(
969
http_server.TestingHTTPRequestHandler):
970
"""Reply to multiple range requests truncating the last ones.
972
This server generates responses whose Content-Length describes all the
973
ranges, but fail to include the last ones leading to client short reads.
974
This has been observed randomly with lighttpd (bug #179368).
977
_truncated_ranges = 2
979
def get_multiple_ranges(self, file, file_size, ranges):
980
self.send_response(206)
981
self.send_header('Accept-Ranges', 'bytes')
983
self.send_header('Content-Type',
984
'multipart/byteranges; boundary=%s' % boundary)
985
boundary_line = '--%s\r\n' % boundary
986
# Calculate the Content-Length
988
for (start, end) in ranges:
989
content_length += len(boundary_line)
990
content_length += self._header_line_length(
991
'Content-type', 'application/octet-stream')
992
content_length += self._header_line_length(
993
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
994
content_length += len('\r\n') # end headers
995
content_length += end - start # + 1
996
content_length += len(boundary_line)
997
self.send_header('Content-length', content_length)
1000
# Send the multipart body
1002
for (start, end) in ranges:
1003
self.wfile.write(boundary_line)
1004
self.send_header('Content-type', 'application/octet-stream')
1005
self.send_header('Content-Range', 'bytes %d-%d/%d'
1006
% (start, end, file_size))
1008
if cur + self._truncated_ranges >= len(ranges):
1009
# Abruptly ends the response and close the connection
1010
self.close_connection = 1
1012
self.send_range_content(file, start, end - start + 1)
1015
self.wfile.write(boundary_line)
1018
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
1020
_req_handler_class = TruncatedMultipleRangeRequestHandler
1023
super(TestTruncatedMultipleRangeServer, self).setUp()
1024
self.build_tree_contents([('a', '0123456789')],)
1026
def test_readv_with_short_reads(self):
1027
server = self.get_readonly_server()
1028
t = self._transport(server.get_url())
1029
# Force separate ranges for each offset
1030
t._bytes_to_read_before_seek = 0
1031
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1032
self.assertEqual((0, '0'), ireadv.next())
1033
self.assertEqual((2, '2'), ireadv.next())
1034
if not self._testing_pycurl():
1035
# Only one request have been issued so far (except for pycurl that
1036
# try to read the whole response at once)
1037
self.assertEqual(1, server.GET_request_nb)
1038
self.assertEqual((4, '45'), ireadv.next())
1039
self.assertEqual((9, '9'), ireadv.next())
1040
# Both implementations issue 3 requests but:
1041
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1043
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
1044
self.assertEqual(3, server.GET_request_nb)
1045
# Finally the client have tried a single range request and stays in
1047
self.assertEqual('single', t._range_hint)
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
# Requests with more range specifiers will error out
730
1081
def create_transport_readonly_server(self):
731
# Requests with more range specifiers will error out
732
return LimitedRangeHTTPServer(range_limit=self.range_limit)
1082
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1083
protocol_version=self._protocol_version)
734
1085
def get_transport(self):
735
1086
return self._transport(self.get_readonly_server().get_url())
737
1088
def setUp(self):
738
TestCaseWithWebserver.setUp(self)
1089
http_utils.TestCaseWithWebserver.setUp(self)
739
1090
# We need to manipulate ranges that correspond to real chunks in the
740
1091
# response, so we build a content appropriately.
741
filler = ''.join(['abcdefghij' for _ in range(102)])
1092
filler = ''.join(['abcdefghij' for x in range(102)])
742
1093
content = ''.join(['%04d' % v + filler for v in range(16)])
743
1094
self.build_tree_contents([('a', content)],)
1297
1610
# Only one 'Authentication Required' error should occur
1298
1611
self.assertEqual(1, self.server.auth_required_errors)
1301
class TestHTTPAuth(TestAuth):
1302
"""Test HTTP authentication schemes.
1304
Daughter classes MUST inherit from TestCaseWithWebserver too.
1307
_auth_header = 'Authorization'
1310
TestCaseWithWebserver.setUp(self)
1311
self.server = self.get_readonly_server()
1312
TestAuth.setUp(self)
1314
def get_user_transport(self, user=None, password=None):
1315
return self._transport(self.get_user_url(user, password))
1613
def _check_password_prompt(self, scheme, user, actual_prompt):
1614
expected_prompt = (self._password_prompt_prefix
1615
+ ("%s %s@%s:%d, Realm: '%s' password: "
1617
user, self.server.host, self.server.port,
1618
self.server.auth_realm)))
1619
self.assertEqual(expected_prompt, actual_prompt)
1621
def _expected_username_prompt(self, scheme):
1622
return (self._username_prompt_prefix
1623
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1624
self.server.host, self.server.port,
1625
self.server.auth_realm))
1627
def test_no_prompt_for_password_when_using_auth_config(self):
1628
if self._testing_pycurl():
1629
raise tests.TestNotApplicable(
1630
'pycurl does not support authentication.conf'
1631
' since it cannot prompt')
1635
stdin_content = 'bar\n' # Not the right password
1636
self.server.add_user(user, password)
1637
t = self.get_user_transport(user, None)
1638
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1639
stderr=tests.StringIOWrapper())
1640
# Create a minimal config file with the right password
1641
conf = config.AuthenticationConfig()
1642
conf._get_config().update(
1643
{'httptest': {'scheme': 'http', 'port': self.server.port,
1644
'user': user, 'password': password}})
1646
# Issue a request to the server to connect
1647
self.assertEqual('contents of a\n',t.get('a').read())
1648
# stdin should have been left untouched
1649
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1650
# Only one 'Authentication Required' error should occur
1651
self.assertEqual(1, self.server.auth_required_errors)
1653
def test_user_from_auth_conf(self):
1654
if self._testing_pycurl():
1655
raise tests.TestNotApplicable(
1656
'pycurl does not support authentication.conf')
1659
self.server.add_user(user, password)
1660
# Create a minimal config file with the right password
1661
conf = config.AuthenticationConfig()
1662
conf._get_config().update(
1663
{'httptest': {'scheme': 'http', 'port': self.server.port,
1664
'user': user, 'password': password}})
1666
t = self.get_user_transport(None, None)
1667
# Issue a request to the server to connect
1668
self.assertEqual('contents of a\n', t.get('a').read())
1669
# Only one 'Authentication Required' error should occur
1670
self.assertEqual(1, self.server.auth_required_errors)
1672
def test_changing_nonce(self):
1673
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1674
http_utils.ProxyDigestAuthServer):
1675
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1676
if self._testing_pycurl():
1677
raise tests.KnownFailure(
1678
'pycurl does not handle a nonce change')
1679
self.server.add_user('joe', 'foo')
1680
t = self.get_user_transport('joe', 'foo')
1681
self.assertEqual('contents of a\n', t.get('a').read())
1682
self.assertEqual('contents of b\n', t.get('b').read())
1683
# Only one 'Authentication Required' error should have
1685
self.assertEqual(1, self.server.auth_required_errors)
1686
# The server invalidates the current nonce
1687
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1688
self.assertEqual('contents of a\n', t.get('a').read())
1689
# Two 'Authentication Required' errors should occur (the
1690
# initial 'who are you' and a second 'who are you' with the new nonce)
1691
self.assertEqual(2, self.server.auth_required_errors)
1318
1695
class TestProxyAuth(TestAuth):
1319
"""Test proxy authentication schemes.
1696
"""Test proxy authentication schemes."""
1321
Daughter classes MUST also inherit from TestCaseWithWebserver.
1323
1698
_auth_header = 'Proxy-authorization'
1699
_password_prompt_prefix = 'Proxy '
1700
_username_prompt_prefix = 'Proxy '
1325
1702
def setUp(self):
1326
TestCaseWithWebserver.setUp(self)
1327
self.server = self.get_readonly_server()
1703
super(TestProxyAuth, self).setUp()
1328
1704
self._old_env = {}
1329
1705
self.addCleanup(self._restore_env)
1330
TestAuth.setUp(self)
1331
1706
# Override the contents to avoid false positives
1332
1707
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1333
1708
('b', 'not proxied contents of b\n'),
1347
1722
for name, value in self._old_env.iteritems():
1348
1723
osutils.set_or_unset_env(name, value)
1351
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1352
"""Test http basic authentication scheme"""
1354
_transport = HttpTransport_urllib
1356
def create_transport_readonly_server(self):
1357
return HTTPBasicAuthServer()
1360
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1361
"""Test proxy basic authentication scheme"""
1363
_transport = HttpTransport_urllib
1365
def create_transport_readonly_server(self):
1366
return ProxyBasicAuthServer()
1369
class TestDigestAuth(object):
1370
"""Digest Authentication specific tests"""
1372
def test_changing_nonce(self):
1373
self.server.add_user('joe', 'foo')
1374
t = self.get_user_transport('joe', 'foo')
1375
self.assertEqual('contents of a\n', t.get('a').read())
1376
self.assertEqual('contents of b\n', t.get('b').read())
1377
# Only one 'Authentication Required' error should have
1379
self.assertEqual(1, self.server.auth_required_errors)
1380
# The server invalidates the current nonce
1381
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1382
self.assertEqual('contents of a\n', t.get('a').read())
1383
# Two 'Authentication Required' errors should occur (the
1384
# initial 'who are you' and a second 'who are you' with the new nonce)
1385
self.assertEqual(2, self.server.auth_required_errors)
1388
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1389
"""Test http digest authentication scheme"""
1391
_transport = HttpTransport_urllib
1393
def create_transport_readonly_server(self):
1394
return HTTPDigestAuthServer()
1397
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1398
TestCaseWithWebserver):
1399
"""Test proxy digest authentication scheme"""
1401
_transport = HttpTransport_urllib
1403
def create_transport_readonly_server(self):
1404
return ProxyDigestAuthServer()
1725
def test_empty_pass(self):
1726
if self._testing_pycurl():
1728
if pycurl.version_info()[1] < '7.16.0':
1729
raise tests.KnownFailure(
1730
'pycurl < 7.16.0 does not handle empty proxy passwords')
1731
super(TestProxyAuth, self).test_empty_pass()
1734
class SampleSocket(object):
1735
"""A socket-like object for use in testing the HTTP request handler."""
1737
def __init__(self, socket_read_content):
1738
"""Constructs a sample socket.
1740
:param socket_read_content: a byte sequence
1742
# Use plain python StringIO so we can monkey-patch the close method to
1743
# not discard the contents.
1744
from StringIO import StringIO
1745
self.readfile = StringIO(socket_read_content)
1746
self.writefile = StringIO()
1747
self.writefile.close = lambda: None
1749
def makefile(self, mode='r', bufsize=None):
1751
return self.readfile
1753
return self.writefile
1756
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1759
super(SmartHTTPTunnellingTest, self).setUp()
1760
# We use the VFS layer as part of HTTP tunnelling tests.
1761
self._captureVar('BZR_NO_SMART_VFS', None)
1762
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1764
def create_transport_readonly_server(self):
1765
return http_utils.HTTPServerWithSmarts(
1766
protocol_version=self._protocol_version)
1768
def test_open_bzrdir(self):
1769
branch = self.make_branch('relpath')
1770
http_server = self.get_readonly_server()
1771
url = http_server.get_url() + 'relpath'
1772
bd = bzrdir.BzrDir.open(url)
1773
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1775
def test_bulk_data(self):
1776
# We should be able to send and receive bulk data in a single message.
1777
# The 'readv' command in the smart protocol both sends and receives
1778
# bulk data, so we use that.
1779
self.build_tree(['data-file'])
1780
http_server = self.get_readonly_server()
1781
http_transport = self._transport(http_server.get_url())
1782
medium = http_transport.get_smart_medium()
1783
# Since we provide the medium, the url below will be mostly ignored
1784
# during the test, as long as the path is '/'.
1785
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1788
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1790
def test_http_send_smart_request(self):
1792
post_body = 'hello\n'
1793
expected_reply_body = 'ok\x012\n'
1795
http_server = self.get_readonly_server()
1796
http_transport = self._transport(http_server.get_url())
1797
medium = http_transport.get_smart_medium()
1798
response = medium.send_http_smart_request(post_body)
1799
reply_body = response.read()
1800
self.assertEqual(expected_reply_body, reply_body)
1802
def test_smart_http_server_post_request_handler(self):
1803
httpd = self.get_readonly_server()._get_httpd()
1805
socket = SampleSocket(
1806
'POST /.bzr/smart %s \r\n' % self._protocol_version
1807
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1809
+ 'Content-Length: 6\r\n'
1812
# Beware: the ('localhost', 80) below is the
1813
# client_address parameter, but we don't have one because
1814
# we have defined a socket which is not bound to an
1815
# address. The test framework never uses this client
1816
# address, so far...
1817
request_handler = http_utils.SmartRequestHandler(socket,
1820
response = socket.writefile.getvalue()
1821
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1822
# This includes the end of the HTTP headers, and all the body.
1823
expected_end_of_response = '\r\n\r\nok\x012\n'
1824
self.assertEndsWith(response, expected_end_of_response)
1827
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1828
"""No smart server here request handler."""
1831
self.send_error(403, "Forbidden")
1834
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1835
"""Test smart client behaviour against an http server without smarts."""
1837
_req_handler_class = ForbiddenRequestHandler
1839
def test_probe_smart_server(self):
1840
"""Test error handling against server refusing smart requests."""
1841
server = self.get_readonly_server()
1842
t = self._transport(server.get_url())
1843
# No need to build a valid smart request here, the server will not even
1844
# try to interpret it.
1845
self.assertRaises(errors.SmartProtocolError,
1846
t.get_smart_medium().send_http_smart_request,
1849
class Test_redirected_to(tests.TestCase):
1851
def test_redirected_to_subdir(self):
1852
t = self._transport('http://www.example.com/foo')
1853
r = t._redirected_to('http://www.example.com/foo',
1854
'http://www.example.com/foo/subdir')
1855
self.assertIsInstance(r, type(t))
1856
# Both transports share the some connection
1857
self.assertEqual(t._get_connection(), r._get_connection())
1859
def test_redirected_to_self_with_slash(self):
1860
t = self._transport('http://www.example.com/foo')
1861
r = t._redirected_to('http://www.example.com/foo',
1862
'http://www.example.com/foo/')
1863
self.assertIsInstance(r, type(t))
1864
# Both transports share the some connection (one can argue that we
1865
# should return the exact same transport here, but that seems
1867
self.assertEqual(t._get_connection(), r._get_connection())
1869
def test_redirected_to_host(self):
1870
t = self._transport('http://www.example.com/foo')
1871
r = t._redirected_to('http://www.example.com/foo',
1872
'http://foo.example.com/foo/subdir')
1873
self.assertIsInstance(r, type(t))
1875
def test_redirected_to_same_host_sibling_protocol(self):
1876
t = self._transport('http://www.example.com/foo')
1877
r = t._redirected_to('http://www.example.com/foo',
1878
'https://www.example.com/foo')
1879
self.assertIsInstance(r, type(t))
1881
def test_redirected_to_same_host_different_protocol(self):
1882
t = self._transport('http://www.example.com/foo')
1883
r = t._redirected_to('http://www.example.com/foo',
1884
'ftp://www.example.com/foo')
1885
self.assertNotEquals(type(r), type(t))
1887
def test_redirected_to_different_host_same_user(self):
1888
t = self._transport('http://joe@www.example.com/foo')
1889
r = t._redirected_to('http://www.example.com/foo',
1890
'https://foo.example.com/foo')
1891
self.assertIsInstance(r, type(t))
1892
self.assertEqual(t._user, r._user)
1895
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1896
"""Request handler for a unique and pre-defined request.
1898
The only thing we care about here is how many bytes travel on the wire. But
1899
since we want to measure it for a real http client, we have to send it
1902
We expect to receive a *single* request nothing more (and we won't even
1903
check what request it is, we just measure the bytes read until an empty
1907
def handle_one_request(self):
1908
tcs = self.server.test_case_server
1909
requestline = self.rfile.readline()
1910
headers = self.MessageClass(self.rfile, 0)
1911
# We just read: the request, the headers, an empty line indicating the
1912
# end of the headers.
1913
bytes_read = len(requestline)
1914
for line in headers.headers:
1915
bytes_read += len(line)
1916
bytes_read += len('\r\n')
1917
if requestline.startswith('POST'):
1918
# The body should be a single line (or we don't know where it ends
1919
# and we don't want to issue a blocking read)
1920
body = self.rfile.readline()
1921
bytes_read += len(body)
1922
tcs.bytes_read = bytes_read
1924
# We set the bytes written *before* issuing the write, the client is
1925
# supposed to consume every produced byte *before* checking that value.
1927
# Doing the oppposite may lead to test failure: we may be interrupted
1928
# after the write but before updating the value. The client can then
1929
# continue and read the value *before* we can update it. And yes,
1930
# this has been observed -- vila 20090129
1931
tcs.bytes_written = len(tcs.canned_response)
1932
self.wfile.write(tcs.canned_response)
1935
class ActivityServerMixin(object):
1937
def __init__(self, protocol_version):
1938
super(ActivityServerMixin, self).__init__(
1939
request_handler=PredefinedRequestHandler,
1940
protocol_version=protocol_version)
1941
# Bytes read and written by the server
1943
self.bytes_written = 0
1944
self.canned_response = None
1947
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1951
if tests.HTTPSServerFeature.available():
1952
from bzrlib.tests import https_server
1953
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1957
class TestActivityMixin(object):
1958
"""Test socket activity reporting.
1960
We use a special purpose server to control the bytes sent and received and
1961
be able to predict the activity on the client socket.
1965
tests.TestCase.setUp(self)
1966
self.server = self._activity_server(self._protocol_version)
1967
self.server.start_server()
1968
self.activities = {}
1969
def report_activity(t, bytes, direction):
1970
count = self.activities.get(direction, 0)
1972
self.activities[direction] = count
1974
# We override at class level because constructors may propagate the
1975
# bound method and render instance overriding ineffective (an
1976
# alternative would be to define a specific ui factory instead...)
1977
self.orig_report_activity = self._transport._report_activity
1978
self._transport._report_activity = report_activity
1981
self._transport._report_activity = self.orig_report_activity
1982
self.server.stop_server()
1983
tests.TestCase.tearDown(self)
1985
def get_transport(self):
1986
return self._transport(self.server.get_url())
1988
def assertActivitiesMatch(self):
1989
self.assertEqual(self.server.bytes_read,
1990
self.activities.get('write', 0), 'written bytes')
1991
self.assertEqual(self.server.bytes_written,
1992
self.activities.get('read', 0), 'read bytes')
1995
self.server.canned_response = '''HTTP/1.1 200 OK\r
1996
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
1997
Server: Apache/2.0.54 (Fedora)\r
1998
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
1999
ETag: "56691-23-38e9ae00"\r
2000
Accept-Ranges: bytes\r
2001
Content-Length: 35\r
2003
Content-Type: text/plain; charset=UTF-8\r
2005
Bazaar-NG meta directory, format 1
2007
t = self.get_transport()
2008
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2009
t.get('foo/bar').read())
2010
self.assertActivitiesMatch()
2013
self.server.canned_response = '''HTTP/1.1 200 OK\r
2014
Server: SimpleHTTP/0.6 Python/2.5.2\r
2015
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2016
Content-type: application/octet-stream\r
2017
Content-Length: 20\r
2018
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2021
t = self.get_transport()
2022
self.assertTrue(t.has('foo/bar'))
2023
self.assertActivitiesMatch()
2025
def test_readv(self):
2026
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2027
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2028
Server: Apache/2.0.54 (Fedora)\r
2029
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2030
ETag: "238a3c-16ec2-805c5540"\r
2031
Accept-Ranges: bytes\r
2032
Content-Length: 1534\r
2034
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2037
--418470f848b63279b\r
2038
Content-type: text/plain; charset=UTF-8\r
2039
Content-range: bytes 0-254/93890\r
2041
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2042
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2043
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2044
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2045
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2047
--418470f848b63279b\r
2048
Content-type: text/plain; charset=UTF-8\r
2049
Content-range: bytes 1000-2049/93890\r
2052
mbp@sourcefrog.net-20050311063625-07858525021f270b
2053
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2054
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2055
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2056
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2057
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2058
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2059
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2060
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2061
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2062
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2063
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2064
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2065
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2066
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2067
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2068
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2069
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2070
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2071
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2073
--418470f848b63279b--\r
2075
t = self.get_transport()
2076
# Remember that the request is ignored and that the ranges below
2077
# doesn't have to match the canned response.
2078
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2079
self.assertEqual(2, len(l))
2080
self.assertActivitiesMatch()
2082
def test_post(self):
2083
self.server.canned_response = '''HTTP/1.1 200 OK\r
2084
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2085
Server: Apache/2.0.54 (Fedora)\r
2086
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2087
ETag: "56691-23-38e9ae00"\r
2088
Accept-Ranges: bytes\r
2089
Content-Length: 35\r
2091
Content-Type: text/plain; charset=UTF-8\r
2093
lalala whatever as long as itsssss
2095
t = self.get_transport()
2096
# We must send a single line of body bytes, see
2097
# PredefinedRequestHandler.handle_one_request
2098
code, f = t._post('abc def end-of-body\n')
2099
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2100
self.assertActivitiesMatch()
2103
class TestActivity(tests.TestCase, TestActivityMixin):
2106
tests.TestCase.setUp(self)
2107
self.server = self._activity_server(self._protocol_version)
2108
self.server.start_server()
2109
self.activities = {}
2110
def report_activity(t, bytes, direction):
2111
count = self.activities.get(direction, 0)
2113
self.activities[direction] = count
2115
# We override at class level because constructors may propagate the
2116
# bound method and render instance overriding ineffective (an
2117
# alternative would be to define a specific ui factory instead...)
2118
self.orig_report_activity = self._transport._report_activity
2119
self._transport._report_activity = report_activity
2122
self._transport._report_activity = self.orig_report_activity
2123
self.server.stop_server()
2124
tests.TestCase.tearDown(self)
2127
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2130
tests.TestCase.setUp(self)
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
self.server = ActivityHTTPServer('HTTP/1.1')
2137
self._transport=_urllib.HttpTransport_urllib
2139
self.server.start_server()
2141
# We override at class level because constructors may propagate the
2142
# bound method and render instance overriding ineffective (an
2143
# alternative would be to define a specific ui factory instead...)
2144
self.orig_report_activity = self._transport._report_activity
2145
self._transport._report_activity = None
2148
self._transport._report_activity = self.orig_report_activity
2149
self.server.stop_server()
2150
tests.TestCase.tearDown(self)
2152
def assertActivitiesMatch(self):
2153
# Nothing to check here
2157
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2158
"""Test authentication on the redirected http server."""
2160
_auth_header = 'Authorization'
2161
_password_prompt_prefix = ''
2162
_username_prompt_prefix = ''
2163
_auth_server = http_utils.HTTPBasicAuthServer
2164
_transport = _urllib.HttpTransport_urllib
2166
def create_transport_readonly_server(self):
2167
return self._auth_server()
2169
def create_transport_secondary_server(self):
2170
"""Create the secondary server redirecting to the primary server"""
2171
new = self.get_readonly_server()
2173
redirecting = http_utils.HTTPServerRedirecting()
2174
redirecting.redirect_to(new.host, new.port)
2178
super(TestAuthOnRedirected, self).setUp()
2179
self.build_tree_contents([('a','a'),
2181
('1/a', 'redirected once'),
2183
new_prefix = 'http://%s:%s' % (self.new_server.host,
2184
self.new_server.port)
2185
self.old_server.redirections = [
2186
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2187
self.old_transport = self._transport(self.old_server.get_url())
2188
self.new_server.add_user('joe', 'foo')
2190
def get_a(self, transport):
2191
return transport.get('a')
2193
def test_auth_on_redirected_via_do_catching_redirections(self):
2194
self.redirections = 0
2196
def redirected(transport, exception, redirection_notice):
2197
self.redirections += 1
2198
dir, file = urlutils.split(exception.target)
2199
return self._transport(dir)
2201
stdout = tests.StringIOWrapper()
2202
stderr = tests.StringIOWrapper()
2203
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2204
stdout=stdout, stderr=stderr)
2205
self.assertEqual('redirected once',
2206
transport.do_catching_redirections(
2207
self.get_a, self.old_transport, redirected).read())
2208
self.assertEqual(1, self.redirections)
2209
# stdin should be empty
2210
self.assertEqual('', ui.ui_factory.stdin.readline())
2211
# stdout should be empty, stderr will contains the prompts
2212
self.assertEqual('', stdout.getvalue())
2214
def test_auth_on_redirected_via_following_redirections(self):
2215
self.new_server.add_user('joe', 'foo')
2216
stdout = tests.StringIOWrapper()
2217
stderr = tests.StringIOWrapper()
2218
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2219
stdout=stdout, stderr=stderr)
2220
t = self.old_transport
2221
req = RedirectedRequest('GET', t.abspath('a'))
2222
new_prefix = 'http://%s:%s' % (self.new_server.host,
2223
self.new_server.port)
2224
self.old_server.redirections = [
2225
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2226
self.assertEqual('redirected once',t._perform(req).read())
2227
# stdin should be empty
2228
self.assertEqual('', ui.ui_factory.stdin.readline())
2229
# stdout should be empty, stderr will contains the prompts
2230
self.assertEqual('', stdout.getvalue())
added
removed
bzrlib/tests/test_http.py
