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., 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.
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.
23
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
24
21
# TODO: What about renaming to bzrlib.tests.transport.http ?
26
from cStringIO import StringIO
30
import SimpleHTTPServer
36
29
from bzrlib import (
41
remote as _mod_remote,
47
34
from bzrlib.tests import (
38
from bzrlib.tests.HttpServer import (
43
from bzrlib.tests.HTTPTestUtil import (
44
BadProtocolRequestHandler,
45
BadStatusRequestHandler,
46
FakeProxyRequestHandler,
47
ForbiddenRequestHandler,
48
HTTPServerRedirecting,
49
InvalidStatusRequestHandler,
50
NoRangeRequestHandler,
51
SingleRangeRequestHandler,
52
TestCaseWithRedirectedWebserver,
53
TestCaseWithTwoWebservers,
54
TestCaseWithWebserver,
53
57
from bzrlib.transport import (
58
do_catching_redirections,
57
62
from bzrlib.transport.http import (
63
if features.pycurl.available():
64
from bzrlib.transport.http._pycurl import PyCurlTransport
67
def load_tests(standard_tests, module, loader):
68
"""Multiply tests for http clients and protocol versions."""
69
result = loader.suiteClass()
71
# one for each transport implementation
72
t_tests, remaining_tests = tests.split_suite_by_condition(
73
standard_tests, tests.condition_isinstance((
74
TestHttpTransportRegistration,
75
TestHttpTransportUrls,
78
transport_scenarios = [
79
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
80
_server=http_server.HttpServer_urllib,
81
_url_protocol='http+urllib',)),
83
if features.pycurl.available():
84
transport_scenarios.append(
85
('pycurl', dict(_transport=PyCurlTransport,
86
_server=http_server.HttpServer_PyCurl,
87
_url_protocol='http+pycurl',)))
88
tests.multiply_tests(t_tests, transport_scenarios, result)
90
protocol_scenarios = [
91
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
92
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
95
# some tests are parametrized by the protocol version only
96
p_tests, remaining_tests = tests.split_suite_by_condition(
97
remaining_tests, tests.condition_isinstance((
100
tests.multiply_tests(p_tests, protocol_scenarios, result)
102
# each implementation tested with each HTTP version
103
tp_tests, remaining_tests = tests.split_suite_by_condition(
104
remaining_tests, tests.condition_isinstance((
105
SmartHTTPTunnellingTest,
106
TestDoCatchRedirections,
108
TestHTTPRedirections,
109
TestHTTPSilentRedirections,
110
TestLimitedRangeRequestServer,
114
TestSpecificRequestHandler,
116
tp_scenarios = tests.multiply_scenarios(transport_scenarios,
118
tests.multiply_tests(tp_tests, tp_scenarios, result)
120
# proxy auth: each auth scheme on all http versions on all implementations.
121
tppa_tests, remaining_tests = tests.split_suite_by_condition(
122
remaining_tests, tests.condition_isinstance((
125
proxy_auth_scheme_scenarios = [
126
('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
127
('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
129
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
131
tppa_scenarios = tests.multiply_scenarios(tp_scenarios,
132
proxy_auth_scheme_scenarios)
133
tests.multiply_tests(tppa_tests, tppa_scenarios, result)
135
# auth: each auth scheme on all http versions on all implementations.
136
tpa_tests, remaining_tests = tests.split_suite_by_condition(
137
remaining_tests, tests.condition_isinstance((
140
auth_scheme_scenarios = [
141
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
142
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
144
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
146
tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
147
auth_scheme_scenarios)
148
tests.multiply_tests(tpa_tests, tpa_scenarios, result)
150
# activity: on all http[s] versions on all implementations
151
tpact_tests, remaining_tests = tests.split_suite_by_condition(
152
remaining_tests, tests.condition_isinstance((
155
activity_scenarios = [
156
('urllib,http', dict(_activity_server=ActivityHTTPServer,
157
_transport=_urllib.HttpTransport_urllib,)),
159
if tests.HTTPSServerFeature.available():
160
activity_scenarios.append(
161
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
162
_transport=_urllib.HttpTransport_urllib,)),)
163
if features.pycurl.available():
164
activity_scenarios.append(
165
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
166
_transport=PyCurlTransport,)),)
167
if tests.HTTPSServerFeature.available():
168
from bzrlib.tests import (
171
# FIXME: Until we have a better way to handle self-signed
172
# certificates (like allowing them in a test specific
173
# authentication.conf for example), we need some specialized pycurl
174
# transport for tests.
175
class HTTPS_pycurl_transport(PyCurlTransport):
177
def __init__(self, base, _from_transport=None):
178
super(HTTPS_pycurl_transport, self).__init__(
179
base, _from_transport)
180
self.cabundle = str(ssl_certs.build_path('ca.crt'))
182
activity_scenarios.append(
183
('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
184
_transport=HTTPS_pycurl_transport,)),)
186
tpact_scenarios = tests.multiply_scenarios(activity_scenarios,
188
tests.multiply_tests(tpact_tests, tpact_scenarios, result)
190
# No parametrization for the remaining tests
191
result.addTests(remaining_tests)
67
from bzrlib.transport.http._urllib import HttpTransport_urllib
196
70
class FakeManager(object):
221
95
self.received_bytes = ''
225
return '%s://%s:%s/' % (self.scheme, self.host, self.port)
227
def start_server(self):
228
98
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
229
99
self._sock.bind(('127.0.0.1', 0))
230
100
self.host, self.port = self._sock.getsockname()
231
101
self._ready = threading.Event()
232
self._thread = test_server.ThreadWithException(
233
event=self._ready, target=self._accept_read_and_reply)
102
self._thread = threading.Thread(target=self._accept_read_and_reply)
103
self._thread.setDaemon(True)
234
104
self._thread.start()
235
if 'threads' in tests.selftest_debug_flags:
236
sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
239
107
def _accept_read_and_reply(self):
240
108
self._sock.listen(1)
241
109
self._ready.set()
242
conn, address = self._sock.accept()
243
if self._expect_body_tail is not None:
110
self._sock.settimeout(5)
112
conn, address = self._sock.accept()
113
# On win32, the accepted connection will be non-blocking to start
114
# with because we're using settimeout.
115
conn.setblocking(True)
244
116
while not self.received_bytes.endswith(self._expect_body_tail):
245
117
self.received_bytes += conn.recv(4096)
246
118
conn.sendall('HTTP/1.1 200 OK\r\n')
119
except socket.timeout:
120
# Make sure the client isn't stuck waiting for us to e.g. accept.
248
121
self._sock.close()
249
122
except socket.error:
250
123
# The client may have already closed the socket.
253
def stop_server(self):
255
# Issue a fake connection to wake up the server and allow it to
257
fake_conn = osutils.connect_socket((self.host, self.port))
259
129
except socket.error:
260
130
# We might have already closed it. We don't care.
265
if 'threads' in tests.selftest_debug_flags:
266
sys.stderr.write('Thread joined: %s\n' % (self._thread.ident,))
269
class TestAuthHeader(tests.TestCase):
271
def parse_header(self, header, auth_handler_class=None):
272
if auth_handler_class is None:
273
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
274
self.auth_handler = auth_handler_class()
275
return self.auth_handler._parse_auth_header(header)
277
def test_empty_header(self):
278
scheme, remainder = self.parse_header('')
279
self.assertEqual('', scheme)
280
self.assertIs(None, remainder)
282
def test_negotiate_header(self):
283
scheme, remainder = self.parse_header('Negotiate')
284
self.assertEqual('negotiate', scheme)
285
self.assertIs(None, remainder)
287
def test_basic_header(self):
288
scheme, remainder = self.parse_header(
289
'Basic realm="Thou should not pass"')
290
self.assertEqual('basic', scheme)
291
self.assertEqual('realm="Thou should not pass"', remainder)
293
def test_basic_extract_realm(self):
294
scheme, remainder = self.parse_header(
295
'Basic realm="Thou should not pass"',
296
_urllib2_wrappers.BasicAuthHandler)
297
match, realm = self.auth_handler.extract_realm(remainder)
298
self.assertTrue(match is not None)
299
self.assertEqual('Thou should not pass', realm)
301
def test_digest_header(self):
302
scheme, remainder = self.parse_header(
303
'Digest realm="Thou should not pass"')
304
self.assertEqual('digest', scheme)
305
self.assertEqual('realm="Thou should not pass"', remainder)
308
class TestHTTPServer(tests.TestCase):
309
"""Test the HTTP servers implementations."""
311
def test_invalid_protocol(self):
312
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
314
protocol_version = 'HTTP/0.1'
316
self.assertRaises(httplib.UnknownProtocol,
317
http_server.HttpServer, BogusRequestHandler)
319
def test_force_invalid_protocol(self):
320
self.assertRaises(httplib.UnknownProtocol,
321
http_server.HttpServer, protocol_version='HTTP/0.1')
323
def test_server_start_and_stop(self):
324
server = http_server.HttpServer()
325
self.addCleanup(server.stop_server)
326
server.start_server()
327
self.assertTrue(server.server is not None)
328
self.assertTrue(server.server.serving is not None)
329
self.assertTrue(server.server.serving)
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.server, 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.server,
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.server,
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.server,
370
http_server.TestingHTTPServer)
373
136
class TestWithTransport_pycurl(object):
374
137
"""Test case to inherit from if pycurl is present"""
376
139
def _get_pycurl_maybe(self):
377
self.requireFeature(features.pycurl)
378
return PyCurlTransport
141
from bzrlib.transport.http._pycurl import PyCurlTransport
142
return PyCurlTransport
143
except errors.DependencyNotPresent:
144
raise TestSkipped('pycurl not present')
380
146
_transport = property(_get_pycurl_maybe)
383
class TestHttpUrls(tests.TestCase):
149
class TestHttpUrls(TestCase):
385
151
# TODO: This should be moved to authorization tests once they
388
154
def test_url_parsing(self):
389
155
f = FakeManager()
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@example.com/bzr/bzr.dev', f)
395
self.assertEqual('http://example.com/bzr/bzr.dev', url)
396
self.assertEqual(1, len(f.credentials))
397
self.assertEqual([None, 'example.com', 'user', 'pass'],
401
class TestHttpTransportUrls(tests.TestCase):
402
"""Test the http urls."""
156
url = extract_auth('http://example.com', f)
157
self.assertEquals('http://example.com', url)
158
self.assertEquals(0, len(f.credentials))
159
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
160
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
161
self.assertEquals(1, len(f.credentials))
162
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
166
class TestHttpTransportUrls(object):
167
"""Test the http urls.
169
This MUST be used by daughter classes that also inherit from
172
We can't inherit directly from TestCase or the
173
test framework will try to create an instance which cannot
174
run, its implementation being incomplete.
404
177
def test_abs_url(self):
405
178
"""Construction of absolute http URLs"""
807
658
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
808
659
t.readv, 'a', [(12,2)])
810
def test_readv_multiple_get_requests(self):
811
server = self.get_readonly_server()
812
t = self.get_readonly_transport()
813
# force transport to issue multiple requests
814
t._max_readv_combine = 1
815
t._max_get_ranges = 1
816
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
817
self.assertEqual(l[0], (0, '0'))
818
self.assertEqual(l[1], (1, '1'))
819
self.assertEqual(l[2], (3, '34'))
820
self.assertEqual(l[3], (9, '9'))
821
# The server should have issued 4 requests
822
self.assertEqual(4, server.GET_request_nb)
824
def test_readv_get_max_size(self):
825
server = self.get_readonly_server()
826
t = self.get_readonly_transport()
827
# force transport to issue multiple requests by limiting the number of
828
# bytes by request. Note that this apply to coalesced offsets only, a
829
# single range will keep its size even if bigger than the limit.
831
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
832
self.assertEqual(l[0], (0, '0'))
833
self.assertEqual(l[1], (1, '1'))
834
self.assertEqual(l[2], (2, '2345'))
835
self.assertEqual(l[3], (6, '6789'))
836
# The server should have issued 3 requests
837
self.assertEqual(3, server.GET_request_nb)
839
def test_complete_readv_leave_pipe_clean(self):
840
server = self.get_readonly_server()
841
t = self.get_readonly_transport()
842
# force transport to issue multiple requests
844
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
845
# The server should have issued 3 requests
846
self.assertEqual(3, server.GET_request_nb)
847
self.assertEqual('0123456789', t.get_bytes('a'))
848
self.assertEqual(4, server.GET_request_nb)
850
def test_incomplete_readv_leave_pipe_clean(self):
851
server = self.get_readonly_server()
852
t = self.get_readonly_transport()
853
# force transport to issue multiple requests
855
# Don't collapse readv results into a list so that we leave unread
856
# bytes on the socket
857
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
858
self.assertEqual((0, '0'), ireadv.next())
859
# The server should have issued one request so far
860
self.assertEqual(1, server.GET_request_nb)
861
self.assertEqual('0123456789', t.get_bytes('a'))
862
# get_bytes issued an additional request, the readv pending ones are
864
self.assertEqual(2, server.GET_request_nb)
867
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
868
"""Always reply to range request as if they were single.
870
Don't be explicit about it, just to annoy the clients.
873
def get_multiple_ranges(self, file, file_size, ranges):
874
"""Answer as if it was a single range request and ignores the rest"""
875
(start, end) = ranges[0]
876
return self.get_single_range(file, file_size, start, end)
879
662
class TestSingleRangeRequestServer(TestRangeRequestServer):
880
663
"""Test readv against a server which accept only single range requests"""
882
_req_handler_class = SingleRangeRequestHandler
885
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
886
"""Only reply to simple range requests, errors out on multiple"""
888
def get_multiple_ranges(self, file, file_size, ranges):
889
"""Refuses the multiple ranges request"""
892
self.send_error(416, "Requested range not satisfiable")
894
(start, end) = ranges[0]
895
return self.get_single_range(file, file_size, start, end)
898
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
899
"""Test readv against a server which only accept single range requests"""
901
_req_handler_class = SingleOnlyRangeRequestHandler
904
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
905
"""Ignore range requests without notice"""
908
# Update the statistics
909
self.server.test_case_server.GET_request_nb += 1
910
# Just bypass the range handling done by TestingHTTPRequestHandler
911
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
665
def create_transport_readonly_server(self):
666
return HttpServer(SingleRangeRequestHandler)
669
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
670
TestCaseWithWebserver):
671
"""Tests single range requests accepting server for urllib implementation"""
673
_transport = HttpTransport_urllib
676
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
677
TestSingleRangeRequestServer,
678
TestCaseWithWebserver):
679
"""Tests single range requests accepting server for pycurl implementation"""
914
682
class TestNoRangeRequestServer(TestRangeRequestServer):
915
683
"""Test readv against a server which do not accept range requests"""
917
_req_handler_class = NoRangeRequestHandler
920
class MultipleRangeWithoutContentLengthRequestHandler(
921
http_server.TestingHTTPRequestHandler):
922
"""Reply to multiple range requests without content length header."""
924
def get_multiple_ranges(self, file, file_size, ranges):
925
self.send_response(206)
926
self.send_header('Accept-Ranges', 'bytes')
927
boundary = "%d" % random.randint(0,0x7FFFFFFF)
928
self.send_header("Content-Type",
929
"multipart/byteranges; boundary=%s" % boundary)
931
for (start, end) in ranges:
932
self.wfile.write("--%s\r\n" % boundary)
933
self.send_header("Content-type", 'application/octet-stream')
934
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
938
self.send_range_content(file, start, end - start + 1)
940
self.wfile.write("--%s\r\n" % boundary)
943
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
945
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
948
class TruncatedMultipleRangeRequestHandler(
949
http_server.TestingHTTPRequestHandler):
950
"""Reply to multiple range requests truncating the last ones.
952
This server generates responses whose Content-Length describes all the
953
ranges, but fail to include the last ones leading to client short reads.
954
This has been observed randomly with lighttpd (bug #179368).
957
_truncated_ranges = 2
959
def get_multiple_ranges(self, file, file_size, ranges):
960
self.send_response(206)
961
self.send_header('Accept-Ranges', 'bytes')
963
self.send_header('Content-Type',
964
'multipart/byteranges; boundary=%s' % boundary)
965
boundary_line = '--%s\r\n' % boundary
966
# Calculate the Content-Length
968
for (start, end) in ranges:
969
content_length += len(boundary_line)
970
content_length += self._header_line_length(
971
'Content-type', 'application/octet-stream')
972
content_length += self._header_line_length(
973
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
974
content_length += len('\r\n') # end headers
975
content_length += end - start # + 1
976
content_length += len(boundary_line)
977
self.send_header('Content-length', content_length)
980
# Send the multipart body
982
for (start, end) in ranges:
983
self.wfile.write(boundary_line)
984
self.send_header('Content-type', 'application/octet-stream')
985
self.send_header('Content-Range', 'bytes %d-%d/%d'
986
% (start, end, file_size))
988
if cur + self._truncated_ranges >= len(ranges):
989
# Abruptly ends the response and close the connection
990
self.close_connection = 1
992
self.send_range_content(file, start, end - start + 1)
995
self.wfile.write(boundary_line)
998
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
1000
_req_handler_class = TruncatedMultipleRangeRequestHandler
1003
super(TestTruncatedMultipleRangeServer, self).setUp()
1004
self.build_tree_contents([('a', '0123456789')],)
1006
def test_readv_with_short_reads(self):
1007
server = self.get_readonly_server()
1008
t = self.get_readonly_transport()
1009
# Force separate ranges for each offset
1010
t._bytes_to_read_before_seek = 0
1011
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1012
self.assertEqual((0, '0'), ireadv.next())
1013
self.assertEqual((2, '2'), ireadv.next())
1014
if not self._testing_pycurl():
1015
# Only one request have been issued so far (except for pycurl that
1016
# try to read the whole response at once)
1017
self.assertEqual(1, server.GET_request_nb)
1018
self.assertEqual((4, '45'), ireadv.next())
1019
self.assertEqual((9, '9'), ireadv.next())
1020
# Both implementations issue 3 requests but:
1021
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1023
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
1024
self.assertEqual(3, server.GET_request_nb)
1025
# Finally the client have tried a single range request and stays in
1027
self.assertEqual('single', t._range_hint)
1029
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1030
"""Errors out when range specifiers exceed the limit"""
1032
def get_multiple_ranges(self, file, file_size, ranges):
1033
"""Refuses the multiple ranges request"""
1034
tcs = self.server.test_case_server
1035
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1037
# Emulate apache behavior
1038
self.send_error(400, "Bad Request")
1040
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1041
self, file, file_size, ranges)
1044
class LimitedRangeHTTPServer(http_server.HttpServer):
1045
"""An HttpServer erroring out on requests with too much range specifiers"""
1047
def __init__(self, request_handler=LimitedRangeRequestHandler,
1048
protocol_version=None,
1050
http_server.HttpServer.__init__(self, request_handler,
1051
protocol_version=protocol_version)
1052
self.range_limit = range_limit
1055
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1056
"""Tests readv requests against a server erroring out on too much ranges."""
1058
# Requests with more range specifiers will error out
1061
685
def create_transport_readonly_server(self):
1062
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1063
protocol_version=self._protocol_version)
1066
http_utils.TestCaseWithWebserver.setUp(self)
1067
# We need to manipulate ranges that correspond to real chunks in the
1068
# response, so we build a content appropriately.
1069
filler = ''.join(['abcdefghij' for x in range(102)])
1070
content = ''.join(['%04d' % v + filler for v in range(16)])
1071
self.build_tree_contents([('a', content)],)
1073
def test_few_ranges(self):
1074
t = self.get_readonly_transport()
1075
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1076
self.assertEqual(l[0], (0, '0000'))
1077
self.assertEqual(l[1], (1024, '0001'))
1078
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1080
def test_more_ranges(self):
1081
t = self.get_readonly_transport()
1082
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1083
self.assertEqual(l[0], (0, '0000'))
1084
self.assertEqual(l[1], (1024, '0001'))
1085
self.assertEqual(l[2], (4096, '0004'))
1086
self.assertEqual(l[3], (8192, '0008'))
1087
# The server will refuse to serve the first request (too much ranges),
1088
# a second request will succeed.
1089
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1092
class TestHttpProxyWhiteBox(tests.TestCase):
686
return HttpServer(NoRangeRequestHandler)
689
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
690
TestCaseWithWebserver):
691
"""Tests range requests refusing server for urllib implementation"""
693
_transport = HttpTransport_urllib
696
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
697
TestNoRangeRequestServer,
698
TestCaseWithWebserver):
699
"""Tests range requests refusing server for pycurl implementation"""
702
class TestHttpProxyWhiteBox(TestCase):
1093
703
"""Whitebox test proxy http authorization.
1095
Only the urllib implementation is tested here.
705
These tests concern urllib implementation only.
1098
708
def setUp(self):
1099
tests.TestCase.setUp(self)
1100
710
self._old_env = {}
1101
self.addCleanup(self._restore_env)
715
def _set_and_capture_env_var(self, name, new_value):
716
"""Set an environment variable, and reset it when finished."""
717
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
1103
719
def _install_env(self, env):
1104
720
for name, value in env.iteritems():
1105
self._old_env[name] = osutils.set_or_unset_env(name, value)
721
self._set_and_capture_env_var(name, value)
1107
723
def _restore_env(self):
1108
724
for name, value in self._old_env.iteritems():
1109
725
osutils.set_or_unset_env(name, value)
1111
727
def _proxied_request(self):
1112
handler = _urllib2_wrappers.ProxyHandler()
1113
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
728
from bzrlib.transport.http._urllib2_wrappers import (
733
handler = ProxyHandler()
734
request = Request('GET','http://baz/buzzle')
1114
735
handler.set_proxy(request, 'http')
1222
859
'NO_PROXY': self.no_proxy_host})
1224
861
def test_http_proxy_without_scheme(self):
1225
if self._testing_pycurl():
1226
# pycurl *ignores* invalid proxy env variables. If that ever change
1227
# in the future, this test will fail indicating that pycurl do not
1228
# ignore anymore such variables.
1229
self.not_proxied_in_env({'http_proxy': self.server_host_port})
1231
self.assertRaises(errors.InvalidURL,
1232
self.proxied_in_env,
1233
{'http_proxy': self.server_host_port})
1236
class TestRanges(http_utils.TestCaseWithWebserver):
1237
"""Test the Range header in GET methods."""
1240
http_utils.TestCaseWithWebserver.setUp(self)
862
self.assertRaises(errors.InvalidURL,
864
{'http_proxy': self.proxy_address})
867
class TestProxyHttpServer_urllib(TestProxyHttpServer,
868
TestCaseWithTwoWebservers):
869
"""Tests proxy server for urllib implementation"""
871
_transport = HttpTransport_urllib
874
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
876
TestCaseWithTwoWebservers):
877
"""Tests proxy server for pycurl implementation"""
880
TestProxyHttpServer.setUp(self)
881
# Oh my ! pycurl does not check for the port as part of
882
# no_proxy :-( So we just test the host part
883
self.no_proxy_host = 'localhost'
885
def test_HTTP_PROXY(self):
886
# pycurl do not check HTTP_PROXY for security reasons
887
# (for use in a CGI context that we do not care
888
# about. Should we ?)
891
def test_HTTP_PROXY_with_NO_PROXY(self):
894
def test_http_proxy_without_scheme(self):
895
# pycurl *ignores* invalid proxy env variables. If that
896
# ever change in the future, this test will fail
897
# indicating that pycurl do not ignore anymore such
899
self.not_proxied_in_env({'http_proxy': self.proxy_address})
902
class TestRanges(object):
903
"""Test the Range header in GET methods..
905
This MUST be used by daughter classes that also inherit from
906
TestCaseWithWebserver.
908
We can't inherit directly from TestCaseWithWebserver or the
909
test framework will try to create an instance which cannot
910
run, its implementation being incomplete.
914
TestCaseWithWebserver.setUp(self)
1241
915
self.build_tree_contents([('a', '0123456789')],)
1243
def create_transport_readonly_server(self):
1244
return http_server.HttpServer(protocol_version=self._protocol_version)
1246
def _file_contents(self, relpath, ranges):
1247
t = self.get_readonly_transport()
1248
offsets = [ (start, end - start + 1) for start, end in ranges]
1249
coalesce = t._coalesce_offsets
1250
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1251
code, data = t._get(relpath, coalesced)
1252
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1253
for start, end in ranges:
1255
yield data.read(end - start + 1)
916
server = self.get_readonly_server()
917
self.transport = self._transport(server.get_url())
919
def _file_contents(self, relpath, ranges, tail_amount=0):
920
code, data = self.transport._get(relpath, ranges)
921
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
922
for start, end in ranges:
924
yield data.read(end - start + 1)
1257
926
def _file_tail(self, relpath, tail_amount):
1258
t = self.get_readonly_transport()
1259
code, data = t._get(relpath, [], tail_amount)
1260
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1261
data.seek(-tail_amount, 2)
1262
return data.read(tail_amount)
927
code, data = self.transport._get(relpath, [], tail_amount)
928
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
929
data.seek(-tail_amount + 1, 2)
930
return data.read(tail_amount)
1264
932
def test_range_header(self):
1266
934
map(self.assertEqual,['0', '234'],
1267
935
list(self._file_contents('a', [(0,0), (2,4)])),)
1269
def test_range_header_tail(self):
1270
937
self.assertEqual('789', self._file_tail('a', 3))
1272
def test_syntactically_invalid_range_header(self):
1273
self.assertListRaises(errors.InvalidHttpRange,
1274
self._file_contents, 'a', [(4, 3)])
1276
def test_semantically_invalid_range_header(self):
1277
self.assertListRaises(errors.InvalidHttpRange,
1278
self._file_contents, 'a', [(42, 128)])
1281
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1282
"""Test redirection between http servers."""
938
# Syntactically invalid range
939
self.assertRaises(errors.InvalidRange,
940
self.transport._get, 'a', [(4, 3)])
941
# Semantically invalid range
942
self.assertRaises(errors.InvalidRange,
943
self.transport._get, 'a', [(42, 128)])
946
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
947
"""Test the Range header in GET methods for urllib implementation"""
949
_transport = HttpTransport_urllib
952
class TestRanges_pycurl(TestWithTransport_pycurl,
954
TestCaseWithWebserver):
955
"""Test the Range header in GET methods for pycurl implementation"""
958
class TestHTTPRedirections(object):
959
"""Test redirection between http servers.
961
This MUST be used by daughter classes that also inherit from
962
TestCaseWithRedirectedWebserver.
964
We can't inherit directly from TestCaseWithTwoWebservers or the
965
test framework will try to create an instance which cannot
966
run, its implementation being incomplete.
969
def create_transport_secondary_server(self):
970
"""Create the secondary server redirecting to the primary server"""
971
new = self.get_readonly_server()
973
redirecting = HTTPServerRedirecting()
974
redirecting.redirect_to(new.host, new.port)
1284
977
def setUp(self):
1285
978
super(TestHTTPRedirections, self).setUp()
1439
1144
return self.old_transport.clone(exception.target)
1441
self.assertRaises(errors.TooManyRedirections,
1442
transport.do_catching_redirections,
1146
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1443
1147
self.get_a, self.old_transport, redirected)
1446
class TestAuth(http_utils.TestCaseWithWebserver):
1447
"""Test authentication scheme"""
1449
_auth_header = 'Authorization'
1450
_password_prompt_prefix = ''
1451
_username_prompt_prefix = ''
1456
super(TestAuth, self).setUp()
1457
self.server = self.get_readonly_server()
1458
self.build_tree_contents([('a', 'contents of a\n'),
1459
('b', 'contents of b\n'),])
1461
def create_transport_readonly_server(self):
1462
server = self._auth_server(protocol_version=self._protocol_version)
1463
server._url_protocol = self._url_protocol
1466
def _testing_pycurl(self):
1467
# TODO: This is duplicated for lots of the classes in this file
1468
return (features.pycurl.available()
1469
and self._transport == PyCurlTransport)
1471
def get_user_url(self, user, password):
1472
"""Build an url embedding user and password"""
1473
url = '%s://' % self.server._url_protocol
1474
if user is not None:
1476
if password is not None:
1477
url += ':' + password
1479
url += '%s:%s/' % (self.server.host, self.server.port)
1482
def get_user_transport(self, user, password):
1483
t = transport.get_transport(self.get_user_url(user, password))
1486
def test_no_user(self):
1487
self.server.add_user('joe', 'foo')
1488
t = self.get_user_transport(None, None)
1489
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1490
# Only one 'Authentication Required' error should occur
1491
self.assertEqual(1, self.server.auth_required_errors)
1493
def test_empty_pass(self):
1494
self.server.add_user('joe', '')
1495
t = self.get_user_transport('joe', '')
1496
self.assertEqual('contents of a\n', t.get('a').read())
1497
# Only one 'Authentication Required' error should occur
1498
self.assertEqual(1, self.server.auth_required_errors)
1500
def test_user_pass(self):
1501
self.server.add_user('joe', 'foo')
1502
t = self.get_user_transport('joe', 'foo')
1503
self.assertEqual('contents of a\n', t.get('a').read())
1504
# Only one 'Authentication Required' error should occur
1505
self.assertEqual(1, self.server.auth_required_errors)
1507
def test_unknown_user(self):
1508
self.server.add_user('joe', 'foo')
1509
t = self.get_user_transport('bill', 'foo')
1510
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1511
# Two 'Authentication Required' errors should occur (the
1512
# initial 'who are you' and 'I don't know you, who are
1514
self.assertEqual(2, self.server.auth_required_errors)
1516
def test_wrong_pass(self):
1517
self.server.add_user('joe', 'foo')
1518
t = self.get_user_transport('joe', 'bar')
1519
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1520
# Two 'Authentication Required' errors should occur (the
1521
# initial 'who are you' and 'this is not you, who are you')
1522
self.assertEqual(2, self.server.auth_required_errors)
1524
def test_prompt_for_username(self):
1525
if self._testing_pycurl():
1526
raise tests.TestNotApplicable(
1527
'pycurl cannot prompt, it handles auth by embedding'
1528
' user:pass in urls only')
1530
self.server.add_user('joe', 'foo')
1531
t = self.get_user_transport(None, None)
1532
stdout = tests.StringIOWrapper()
1533
stderr = tests.StringIOWrapper()
1534
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1535
stdout=stdout, stderr=stderr)
1536
self.assertEqual('contents of a\n',t.get('a').read())
1537
# stdin should be empty
1538
self.assertEqual('', ui.ui_factory.stdin.readline())
1540
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1541
self.assertEqual(expected_prompt, stderr.read(len(expected_prompt)))
1542
self.assertEqual('', stdout.getvalue())
1543
self._check_password_prompt(t._unqualified_scheme, 'joe',
1546
def test_prompt_for_password(self):
1547
if self._testing_pycurl():
1548
raise tests.TestNotApplicable(
1549
'pycurl cannot prompt, it handles auth by embedding'
1550
' user:pass in urls only')
1552
self.server.add_user('joe', 'foo')
1553
t = self.get_user_transport('joe', None)
1554
stdout = tests.StringIOWrapper()
1555
stderr = tests.StringIOWrapper()
1556
ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1557
stdout=stdout, stderr=stderr)
1558
self.assertEqual('contents of a\n', t.get('a').read())
1559
# stdin should be empty
1560
self.assertEqual('', ui.ui_factory.stdin.readline())
1561
self._check_password_prompt(t._unqualified_scheme, 'joe',
1563
self.assertEqual('', stdout.getvalue())
1564
# And we shouldn't prompt again for a different request
1565
# against the same transport.
1566
self.assertEqual('contents of b\n',t.get('b').read())
1568
# And neither against a clone
1569
self.assertEqual('contents of b\n',t2.get('b').read())
1570
# Only one 'Authentication Required' error should occur
1571
self.assertEqual(1, self.server.auth_required_errors)
1573
def _check_password_prompt(self, scheme, user, actual_prompt):
1574
expected_prompt = (self._password_prompt_prefix
1575
+ ("%s %s@%s:%d, Realm: '%s' password: "
1577
user, self.server.host, self.server.port,
1578
self.server.auth_realm)))
1579
self.assertEqual(expected_prompt, actual_prompt)
1581
def _expected_username_prompt(self, scheme):
1582
return (self._username_prompt_prefix
1583
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1584
self.server.host, self.server.port,
1585
self.server.auth_realm))
1587
def test_no_prompt_for_password_when_using_auth_config(self):
1588
if self._testing_pycurl():
1589
raise tests.TestNotApplicable(
1590
'pycurl does not support authentication.conf'
1591
' since it cannot prompt')
1595
stdin_content = 'bar\n' # Not the right password
1596
self.server.add_user(user, password)
1597
t = self.get_user_transport(user, None)
1598
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1599
stderr=tests.StringIOWrapper())
1600
# Create a minimal config file with the right password
1601
conf = config.AuthenticationConfig()
1602
conf._get_config().update(
1603
{'httptest': {'scheme': 'http', 'port': self.server.port,
1604
'user': user, 'password': password}})
1606
# Issue a request to the server to connect
1607
self.assertEqual('contents of a\n',t.get('a').read())
1608
# stdin should have been left untouched
1609
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1610
# Only one 'Authentication Required' error should occur
1611
self.assertEqual(1, self.server.auth_required_errors)
1613
def test_user_from_auth_conf(self):
1614
if self._testing_pycurl():
1615
raise tests.TestNotApplicable(
1616
'pycurl does not support authentication.conf')
1619
self.server.add_user(user, password)
1620
# Create a minimal config file with the right password
1621
conf = config.AuthenticationConfig()
1622
conf._get_config().update(
1623
{'httptest': {'scheme': 'http', 'port': self.server.port,
1624
'user': user, 'password': password}})
1626
t = self.get_user_transport(None, None)
1627
# Issue a request to the server to connect
1628
self.assertEqual('contents of a\n', t.get('a').read())
1629
# Only one 'Authentication Required' error should occur
1630
self.assertEqual(1, self.server.auth_required_errors)
1632
def test_changing_nonce(self):
1633
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1634
http_utils.ProxyDigestAuthServer):
1635
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1636
if self._testing_pycurl():
1637
raise tests.KnownFailure(
1638
'pycurl does not handle a nonce change')
1639
self.server.add_user('joe', 'foo')
1640
t = self.get_user_transport('joe', 'foo')
1641
self.assertEqual('contents of a\n', t.get('a').read())
1642
self.assertEqual('contents of b\n', t.get('b').read())
1643
# Only one 'Authentication Required' error should have
1645
self.assertEqual(1, self.server.auth_required_errors)
1646
# The server invalidates the current nonce
1647
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1648
self.assertEqual('contents of a\n', t.get('a').read())
1649
# Two 'Authentication Required' errors should occur (the
1650
# initial 'who are you' and a second 'who are you' with the new nonce)
1651
self.assertEqual(2, self.server.auth_required_errors)
1655
class TestProxyAuth(TestAuth):
1656
"""Test proxy authentication schemes."""
1658
_auth_header = 'Proxy-authorization'
1659
_password_prompt_prefix = 'Proxy '
1660
_username_prompt_prefix = 'Proxy '
1663
super(TestProxyAuth, self).setUp()
1665
self.addCleanup(self._restore_env)
1666
# Override the contents to avoid false positives
1667
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1668
('b', 'not proxied contents of b\n'),
1669
('a-proxied', 'contents of a\n'),
1670
('b-proxied', 'contents of b\n'),
1673
def get_user_transport(self, user, password):
1674
self._install_env({'all_proxy': self.get_user_url(user, password)})
1675
return TestAuth.get_user_transport(self, user, password)
1677
def _install_env(self, env):
1678
for name, value in env.iteritems():
1679
self._old_env[name] = osutils.set_or_unset_env(name, value)
1681
def _restore_env(self):
1682
for name, value in self._old_env.iteritems():
1683
osutils.set_or_unset_env(name, value)
1685
def test_empty_pass(self):
1686
if self._testing_pycurl():
1688
if pycurl.version_info()[1] < '7.16.0':
1689
raise tests.KnownFailure(
1690
'pycurl < 7.16.0 does not handle empty proxy passwords')
1691
super(TestProxyAuth, self).test_empty_pass()
1694
class SampleSocket(object):
1695
"""A socket-like object for use in testing the HTTP request handler."""
1697
def __init__(self, socket_read_content):
1698
"""Constructs a sample socket.
1700
:param socket_read_content: a byte sequence
1702
# Use plain python StringIO so we can monkey-patch the close method to
1703
# not discard the contents.
1704
from StringIO import StringIO
1705
self.readfile = StringIO(socket_read_content)
1706
self.writefile = StringIO()
1707
self.writefile.close = lambda: None
1708
self.close = lambda: None
1710
def makefile(self, mode='r', bufsize=None):
1712
return self.readfile
1714
return self.writefile
1717
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1720
super(SmartHTTPTunnellingTest, self).setUp()
1721
# We use the VFS layer as part of HTTP tunnelling tests.
1722
self._captureVar('BZR_NO_SMART_VFS', None)
1723
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1724
self.http_server = self.get_readonly_server()
1726
def create_transport_readonly_server(self):
1727
server = http_utils.HTTPServerWithSmarts(
1728
protocol_version=self._protocol_version)
1729
server._url_protocol = self._url_protocol
1732
def test_open_bzrdir(self):
1733
branch = self.make_branch('relpath')
1734
url = self.http_server.get_url() + 'relpath'
1735
bd = bzrdir.BzrDir.open(url)
1736
self.addCleanup(bd.transport.disconnect)
1737
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1739
def test_bulk_data(self):
1740
# We should be able to send and receive bulk data in a single message.
1741
# The 'readv' command in the smart protocol both sends and receives
1742
# bulk data, so we use that.
1743
self.build_tree(['data-file'])
1744
http_transport = transport.get_transport(self.http_server.get_url())
1745
medium = http_transport.get_smart_medium()
1746
# Since we provide the medium, the url below will be mostly ignored
1747
# during the test, as long as the path is '/'.
1748
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1751
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1753
def test_http_send_smart_request(self):
1755
post_body = 'hello\n'
1756
expected_reply_body = 'ok\x012\n'
1758
http_transport = transport.get_transport(self.http_server.get_url())
1759
medium = http_transport.get_smart_medium()
1760
response = medium.send_http_smart_request(post_body)
1761
reply_body = response.read()
1762
self.assertEqual(expected_reply_body, reply_body)
1764
def test_smart_http_server_post_request_handler(self):
1765
httpd = self.http_server.server
1767
socket = SampleSocket(
1768
'POST /.bzr/smart %s \r\n' % self._protocol_version
1769
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1771
+ 'Content-Length: 6\r\n'
1774
# Beware: the ('localhost', 80) below is the
1775
# client_address parameter, but we don't have one because
1776
# we have defined a socket which is not bound to an
1777
# address. The test framework never uses this client
1778
# address, so far...
1779
request_handler = http_utils.SmartRequestHandler(socket,
1782
response = socket.writefile.getvalue()
1783
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1784
# This includes the end of the HTTP headers, and all the body.
1785
expected_end_of_response = '\r\n\r\nok\x012\n'
1786
self.assertEndsWith(response, expected_end_of_response)
1789
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1790
"""No smart server here request handler."""
1793
self.send_error(403, "Forbidden")
1796
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1797
"""Test smart client behaviour against an http server without smarts."""
1799
_req_handler_class = ForbiddenRequestHandler
1801
def test_probe_smart_server(self):
1802
"""Test error handling against server refusing smart requests."""
1803
t = self.get_readonly_transport()
1804
# No need to build a valid smart request here, the server will not even
1805
# try to interpret it.
1806
self.assertRaises(errors.SmartProtocolError,
1807
t.get_smart_medium().send_http_smart_request,
1811
class Test_redirected_to(tests.TestCase):
1813
def test_redirected_to_subdir(self):
1814
t = self._transport('http://www.example.com/foo')
1815
r = t._redirected_to('http://www.example.com/foo',
1816
'http://www.example.com/foo/subdir')
1817
self.assertIsInstance(r, type(t))
1818
# Both transports share the some connection
1819
self.assertEqual(t._get_connection(), r._get_connection())
1821
def test_redirected_to_self_with_slash(self):
1822
t = self._transport('http://www.example.com/foo')
1823
r = t._redirected_to('http://www.example.com/foo',
1824
'http://www.example.com/foo/')
1825
self.assertIsInstance(r, type(t))
1826
# Both transports share the some connection (one can argue that we
1827
# should return the exact same transport here, but that seems
1829
self.assertEqual(t._get_connection(), r._get_connection())
1831
def test_redirected_to_host(self):
1832
t = self._transport('http://www.example.com/foo')
1833
r = t._redirected_to('http://www.example.com/foo',
1834
'http://foo.example.com/foo/subdir')
1835
self.assertIsInstance(r, type(t))
1837
def test_redirected_to_same_host_sibling_protocol(self):
1838
t = self._transport('http://www.example.com/foo')
1839
r = t._redirected_to('http://www.example.com/foo',
1840
'https://www.example.com/foo')
1841
self.assertIsInstance(r, type(t))
1843
def test_redirected_to_same_host_different_protocol(self):
1844
t = self._transport('http://www.example.com/foo')
1845
r = t._redirected_to('http://www.example.com/foo',
1846
'ftp://www.example.com/foo')
1847
self.assertNotEquals(type(r), type(t))
1849
def test_redirected_to_different_host_same_user(self):
1850
t = self._transport('http://joe@www.example.com/foo')
1851
r = t._redirected_to('http://www.example.com/foo',
1852
'https://foo.example.com/foo')
1853
self.assertIsInstance(r, type(t))
1854
self.assertEqual(t._user, r._user)
1857
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1858
"""Request handler for a unique and pre-defined request.
1860
The only thing we care about here is how many bytes travel on the wire. But
1861
since we want to measure it for a real http client, we have to send it
1864
We expect to receive a *single* request nothing more (and we won't even
1865
check what request it is, we just measure the bytes read until an empty
1869
def _handle_one_request(self):
1870
tcs = self.server.test_case_server
1871
requestline = self.rfile.readline()
1872
headers = self.MessageClass(self.rfile, 0)
1873
# We just read: the request, the headers, an empty line indicating the
1874
# end of the headers.
1875
bytes_read = len(requestline)
1876
for line in headers.headers:
1877
bytes_read += len(line)
1878
bytes_read += len('\r\n')
1879
if requestline.startswith('POST'):
1880
# The body should be a single line (or we don't know where it ends
1881
# and we don't want to issue a blocking read)
1882
body = self.rfile.readline()
1883
bytes_read += len(body)
1884
tcs.bytes_read = bytes_read
1886
# We set the bytes written *before* issuing the write, the client is
1887
# supposed to consume every produced byte *before* checking that value.
1889
# Doing the oppposite may lead to test failure: we may be interrupted
1890
# after the write but before updating the value. The client can then
1891
# continue and read the value *before* we can update it. And yes,
1892
# this has been observed -- vila 20090129
1893
tcs.bytes_written = len(tcs.canned_response)
1894
self.wfile.write(tcs.canned_response)
1897
class ActivityServerMixin(object):
1899
def __init__(self, protocol_version):
1900
super(ActivityServerMixin, self).__init__(
1901
request_handler=PredefinedRequestHandler,
1902
protocol_version=protocol_version)
1903
# Bytes read and written by the server
1905
self.bytes_written = 0
1906
self.canned_response = None
1909
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1913
if tests.HTTPSServerFeature.available():
1914
from bzrlib.tests import https_server
1915
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1919
class TestActivityMixin(object):
1920
"""Test socket activity reporting.
1922
We use a special purpose server to control the bytes sent and received and
1923
be able to predict the activity on the client socket.
1927
tests.TestCase.setUp(self)
1928
self.server = self._activity_server(self._protocol_version)
1929
self.server.start_server()
1930
self.activities = {}
1931
def report_activity(t, bytes, direction):
1932
count = self.activities.get(direction, 0)
1934
self.activities[direction] = count
1936
# We override at class level because constructors may propagate the
1937
# bound method and render instance overriding ineffective (an
1938
# alternative would be to define a specific ui factory instead...)
1939
self.overrideAttr(self._transport, '_report_activity', report_activity)
1940
self.addCleanup(self.server.stop_server)
1942
def get_transport(self):
1943
t = self._transport(self.server.get_url())
1944
# FIXME: Needs cleanup -- vila 20100611
1947
def assertActivitiesMatch(self):
1948
self.assertEqual(self.server.bytes_read,
1949
self.activities.get('write', 0), 'written bytes')
1950
self.assertEqual(self.server.bytes_written,
1951
self.activities.get('read', 0), 'read bytes')
1954
self.server.canned_response = '''HTTP/1.1 200 OK\r
1955
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
1956
Server: Apache/2.0.54 (Fedora)\r
1957
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
1958
ETag: "56691-23-38e9ae00"\r
1959
Accept-Ranges: bytes\r
1960
Content-Length: 35\r
1962
Content-Type: text/plain; charset=UTF-8\r
1964
Bazaar-NG meta directory, format 1
1966
t = self.get_transport()
1967
self.assertEqual('Bazaar-NG meta directory, format 1\n',
1968
t.get('foo/bar').read())
1969
self.assertActivitiesMatch()
1972
self.server.canned_response = '''HTTP/1.1 200 OK\r
1973
Server: SimpleHTTP/0.6 Python/2.5.2\r
1974
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
1975
Content-type: application/octet-stream\r
1976
Content-Length: 20\r
1977
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
1980
t = self.get_transport()
1981
self.assertTrue(t.has('foo/bar'))
1982
self.assertActivitiesMatch()
1984
def test_readv(self):
1985
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
1986
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
1987
Server: Apache/2.0.54 (Fedora)\r
1988
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
1989
ETag: "238a3c-16ec2-805c5540"\r
1990
Accept-Ranges: bytes\r
1991
Content-Length: 1534\r
1993
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
1996
--418470f848b63279b\r
1997
Content-type: text/plain; charset=UTF-8\r
1998
Content-range: bytes 0-254/93890\r
2000
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2001
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2002
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2003
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2004
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2006
--418470f848b63279b\r
2007
Content-type: text/plain; charset=UTF-8\r
2008
Content-range: bytes 1000-2049/93890\r
2011
mbp@sourcefrog.net-20050311063625-07858525021f270b
2012
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2013
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2014
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2015
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2016
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2017
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2018
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2019
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2020
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2021
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2022
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2023
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2024
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2025
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2026
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2027
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2028
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2029
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2030
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2032
--418470f848b63279b--\r
2034
t = self.get_transport()
2035
# Remember that the request is ignored and that the ranges below
2036
# doesn't have to match the canned response.
2037
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2038
self.assertEqual(2, len(l))
2039
self.assertActivitiesMatch()
2041
def test_post(self):
2042
self.server.canned_response = '''HTTP/1.1 200 OK\r
2043
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2044
Server: Apache/2.0.54 (Fedora)\r
2045
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2046
ETag: "56691-23-38e9ae00"\r
2047
Accept-Ranges: bytes\r
2048
Content-Length: 35\r
2050
Content-Type: text/plain; charset=UTF-8\r
2052
lalala whatever as long as itsssss
2054
t = self.get_transport()
2055
# We must send a single line of body bytes, see
2056
# PredefinedRequestHandler._handle_one_request
2057
code, f = t._post('abc def end-of-body\n')
2058
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2059
self.assertActivitiesMatch()
2062
class TestActivity(tests.TestCase, TestActivityMixin):
2065
TestActivityMixin.setUp(self)
2068
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2070
# Unlike TestActivity, we are really testing ReportingFileSocket and
2071
# ReportingSocket, so we don't need all the parametrization. Since
2072
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2073
# test them through their use by the transport than directly (that's a
2074
# bit less clean but far more simpler and effective).
2075
_activity_server = ActivityHTTPServer
2076
_protocol_version = 'HTTP/1.1'
2079
self._transport =_urllib.HttpTransport_urllib
2080
TestActivityMixin.setUp(self)
2082
def assertActivitiesMatch(self):
2083
# Nothing to check here
2087
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2088
"""Test authentication on the redirected http server."""
2090
_auth_header = 'Authorization'
2091
_password_prompt_prefix = ''
2092
_username_prompt_prefix = ''
2093
_auth_server = http_utils.HTTPBasicAuthServer
2094
_transport = _urllib.HttpTransport_urllib
2097
super(TestAuthOnRedirected, self).setUp()
2098
self.build_tree_contents([('a','a'),
2100
('1/a', 'redirected once'),
2102
new_prefix = 'http://%s:%s' % (self.new_server.host,
2103
self.new_server.port)
2104
self.old_server.redirections = [
2105
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2106
self.old_transport = self.get_old_transport()
2107
self.new_server.add_user('joe', 'foo')
2108
cleanup_http_redirection_connections(self)
2110
def create_transport_readonly_server(self):
2111
server = self._auth_server(protocol_version=self._protocol_version)
2112
server._url_protocol = self._url_protocol
2118
def test_auth_on_redirected_via_do_catching_redirections(self):
2119
self.redirections = 0
2121
def redirected(t, exception, redirection_notice):
2122
self.redirections += 1
2123
redirected_t = t._redirected_to(exception.source, exception.target)
2124
self.addCleanup(redirected_t.disconnect)
2127
stdout = tests.StringIOWrapper()
2128
stderr = tests.StringIOWrapper()
2129
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2130
stdout=stdout, stderr=stderr)
2131
self.assertEqual('redirected once',
2132
transport.do_catching_redirections(
2133
self.get_a, self.old_transport, redirected).read())
2134
self.assertEqual(1, self.redirections)
2135
# stdin should be empty
2136
self.assertEqual('', ui.ui_factory.stdin.readline())
2137
# stdout should be empty, stderr will contains the prompts
2138
self.assertEqual('', stdout.getvalue())
2140
def test_auth_on_redirected_via_following_redirections(self):
2141
self.new_server.add_user('joe', 'foo')
2142
stdout = tests.StringIOWrapper()
2143
stderr = tests.StringIOWrapper()
2144
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2145
stdout=stdout, stderr=stderr)
2146
t = self.old_transport
2147
req = RedirectedRequest('GET', t.abspath('a'))
2148
new_prefix = 'http://%s:%s' % (self.new_server.host,
2149
self.new_server.port)
2150
self.old_server.redirections = [
2151
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2152
self.assertEqual('redirected once', t._perform(req).read())
2153
# stdin should be empty
2154
self.assertEqual('', ui.ui_factory.stdin.readline())
2155
# stdout should be empty, stderr will contains the prompts
2156
self.assertEqual('', stdout.getvalue())
added
removed
bzrlib/tests/test_http.py
