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
from bzrlib.symbol_versioning import (
50
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,
54
57
from bzrlib.transport import (
58
do_catching_redirections,
58
62
from bzrlib.transport.http import (
65
from bzrlib.transport.http._pycurl import PyCurlTransport
67
except errors.DependencyNotPresent:
68
pycurl_present = False
71
def load_tests(standard_tests, module, loader):
72
"""Multiply tests for http clients and protocol versions."""
73
result = loader.suiteClass()
75
# one for each transport implementation
76
t_tests, remaining_tests = tests.split_suite_by_condition(
77
standard_tests, tests.condition_isinstance((
78
TestHttpTransportRegistration,
79
TestHttpTransportUrls,
82
transport_scenarios = [
83
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
84
_server=http_server.HttpServer_urllib,
85
_qualified_prefix='http+urllib',)),
88
transport_scenarios.append(
89
('pycurl', dict(_transport=PyCurlTransport,
90
_server=http_server.HttpServer_PyCurl,
91
_qualified_prefix='http+pycurl',)))
92
tests.multiply_tests(t_tests, transport_scenarios, result)
94
# each implementation tested with each HTTP version
95
tp_tests, remaining_tests = tests.split_suite_by_condition(
96
remaining_tests, tests.condition_isinstance((
97
SmartHTTPTunnellingTest,
98
TestDoCatchRedirections,
100
TestHTTPRedirections,
101
TestHTTPSilentRedirections,
102
TestLimitedRangeRequestServer,
106
TestSpecificRequestHandler,
108
protocol_scenarios = [
109
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
110
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
112
tp_scenarios = tests.multiply_scenarios(transport_scenarios,
114
tests.multiply_tests(tp_tests, tp_scenarios, result)
116
# proxy auth: each auth scheme on all http versions on all implementations.
117
tppa_tests, remaining_tests = tests.split_suite_by_condition(
118
remaining_tests, tests.condition_isinstance((
121
proxy_auth_scheme_scenarios = [
122
('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
123
('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
125
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
127
tppa_scenarios = tests.multiply_scenarios(tp_scenarios,
128
proxy_auth_scheme_scenarios)
129
tests.multiply_tests(tppa_tests, tppa_scenarios, result)
131
# auth: each auth scheme on all http versions on all implementations.
132
tpa_tests, remaining_tests = tests.split_suite_by_condition(
133
remaining_tests, tests.condition_isinstance((
136
auth_scheme_scenarios = [
137
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
138
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
140
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
142
tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
143
auth_scheme_scenarios)
144
tests.multiply_tests(tpa_tests, tpa_scenarios, result)
146
# activity: on all http[s] versions on all implementations
147
tpact_tests, remaining_tests = tests.split_suite_by_condition(
148
remaining_tests, tests.condition_isinstance((
151
activity_scenarios = [
152
('urllib,http', dict(_activity_server=ActivityHTTPServer,
153
_transport=_urllib.HttpTransport_urllib,)),
155
if tests.HTTPSServerFeature.available():
156
activity_scenarios.append(
157
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
158
_transport=_urllib.HttpTransport_urllib,)),)
160
activity_scenarios.append(
161
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
162
_transport=PyCurlTransport,)),)
163
if tests.HTTPSServerFeature.available():
164
from bzrlib.tests import (
167
# FIXME: Until we have a better way to handle self-signed
168
# certificates (like allowing them in a test specific
169
# authentication.conf for example), we need some specialized pycurl
170
# transport for tests.
171
class HTTPS_pycurl_transport(PyCurlTransport):
173
def __init__(self, base, _from_transport=None):
174
super(HTTPS_pycurl_transport, self).__init__(
175
base, _from_transport)
176
self.cabundle = str(ssl_certs.build_path('ca.crt'))
178
activity_scenarios.append(
179
('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
180
_transport=HTTPS_pycurl_transport,)),)
182
tpact_scenarios = tests.multiply_scenarios(activity_scenarios,
184
tests.multiply_tests(tpact_tests, tpact_scenarios, result)
186
# No parametrization for the remaining tests
187
result.addTests(remaining_tests)
67
from bzrlib.transport.http._urllib import HttpTransport_urllib
192
70
class FakeManager(object):
829
658
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
830
659
t.readv, 'a', [(12,2)])
832
def test_readv_multiple_get_requests(self):
833
server = self.get_readonly_server()
834
t = self._transport(server.get_url())
835
# force transport to issue multiple requests
836
t._max_readv_combine = 1
837
t._max_get_ranges = 1
838
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
839
self.assertEqual(l[0], (0, '0'))
840
self.assertEqual(l[1], (1, '1'))
841
self.assertEqual(l[2], (3, '34'))
842
self.assertEqual(l[3], (9, '9'))
843
# The server should have issued 4 requests
844
self.assertEqual(4, server.GET_request_nb)
846
def test_readv_get_max_size(self):
847
server = self.get_readonly_server()
848
t = self._transport(server.get_url())
849
# force transport to issue multiple requests by limiting the number of
850
# bytes by request. Note that this apply to coalesced offsets only, a
851
# single range will keep its size even if bigger than the limit.
853
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
854
self.assertEqual(l[0], (0, '0'))
855
self.assertEqual(l[1], (1, '1'))
856
self.assertEqual(l[2], (2, '2345'))
857
self.assertEqual(l[3], (6, '6789'))
858
# The server should have issued 3 requests
859
self.assertEqual(3, server.GET_request_nb)
861
def test_complete_readv_leave_pipe_clean(self):
862
server = self.get_readonly_server()
863
t = self._transport(server.get_url())
864
# force transport to issue multiple requests
866
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
867
# The server should have issued 3 requests
868
self.assertEqual(3, server.GET_request_nb)
869
self.assertEqual('0123456789', t.get_bytes('a'))
870
self.assertEqual(4, server.GET_request_nb)
872
def test_incomplete_readv_leave_pipe_clean(self):
873
server = self.get_readonly_server()
874
t = self._transport(server.get_url())
875
# force transport to issue multiple requests
877
# Don't collapse readv results into a list so that we leave unread
878
# bytes on the socket
879
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
880
self.assertEqual((0, '0'), ireadv.next())
881
# The server should have issued one request so far
882
self.assertEqual(1, server.GET_request_nb)
883
self.assertEqual('0123456789', t.get_bytes('a'))
884
# get_bytes issued an additional request, the readv pending ones are
886
self.assertEqual(2, server.GET_request_nb)
889
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
890
"""Always reply to range request as if they were single.
892
Don't be explicit about it, just to annoy the clients.
895
def get_multiple_ranges(self, file, file_size, ranges):
896
"""Answer as if it was a single range request and ignores the rest"""
897
(start, end) = ranges[0]
898
return self.get_single_range(file, file_size, start, end)
901
662
class TestSingleRangeRequestServer(TestRangeRequestServer):
902
663
"""Test readv against a server which accept only single range requests"""
904
_req_handler_class = SingleRangeRequestHandler
907
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
908
"""Only reply to simple range requests, errors out on multiple"""
910
def get_multiple_ranges(self, file, file_size, ranges):
911
"""Refuses the multiple ranges request"""
914
self.send_error(416, "Requested range not satisfiable")
916
(start, end) = ranges[0]
917
return self.get_single_range(file, file_size, start, end)
920
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
921
"""Test readv against a server which only accept single range requests"""
923
_req_handler_class = SingleOnlyRangeRequestHandler
926
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
927
"""Ignore range requests without notice"""
930
# Update the statistics
931
self.server.test_case_server.GET_request_nb += 1
932
# Just bypass the range handling done by TestingHTTPRequestHandler
933
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"""
936
682
class TestNoRangeRequestServer(TestRangeRequestServer):
937
683
"""Test readv against a server which do not accept range requests"""
939
_req_handler_class = NoRangeRequestHandler
942
class MultipleRangeWithoutContentLengthRequestHandler(
943
http_server.TestingHTTPRequestHandler):
944
"""Reply to multiple range requests without content length header."""
946
def get_multiple_ranges(self, file, file_size, ranges):
947
self.send_response(206)
948
self.send_header('Accept-Ranges', 'bytes')
949
boundary = "%d" % random.randint(0,0x7FFFFFFF)
950
self.send_header("Content-Type",
951
"multipart/byteranges; boundary=%s" % boundary)
953
for (start, end) in ranges:
954
self.wfile.write("--%s\r\n" % boundary)
955
self.send_header("Content-type", 'application/octet-stream')
956
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
960
self.send_range_content(file, start, end - start + 1)
962
self.wfile.write("--%s\r\n" % boundary)
965
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
967
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
970
class TruncatedMultipleRangeRequestHandler(
971
http_server.TestingHTTPRequestHandler):
972
"""Reply to multiple range requests truncating the last ones.
974
This server generates responses whose Content-Length describes all the
975
ranges, but fail to include the last ones leading to client short reads.
976
This has been observed randomly with lighttpd (bug #179368).
979
_truncated_ranges = 2
981
def get_multiple_ranges(self, file, file_size, ranges):
982
self.send_response(206)
983
self.send_header('Accept-Ranges', 'bytes')
985
self.send_header('Content-Type',
986
'multipart/byteranges; boundary=%s' % boundary)
987
boundary_line = '--%s\r\n' % boundary
988
# Calculate the Content-Length
990
for (start, end) in ranges:
991
content_length += len(boundary_line)
992
content_length += self._header_line_length(
993
'Content-type', 'application/octet-stream')
994
content_length += self._header_line_length(
995
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
996
content_length += len('\r\n') # end headers
997
content_length += end - start # + 1
998
content_length += len(boundary_line)
999
self.send_header('Content-length', content_length)
1002
# Send the multipart body
1004
for (start, end) in ranges:
1005
self.wfile.write(boundary_line)
1006
self.send_header('Content-type', 'application/octet-stream')
1007
self.send_header('Content-Range', 'bytes %d-%d/%d'
1008
% (start, end, file_size))
1010
if cur + self._truncated_ranges >= len(ranges):
1011
# Abruptly ends the response and close the connection
1012
self.close_connection = 1
1014
self.send_range_content(file, start, end - start + 1)
1017
self.wfile.write(boundary_line)
1020
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
1022
_req_handler_class = TruncatedMultipleRangeRequestHandler
1025
super(TestTruncatedMultipleRangeServer, self).setUp()
1026
self.build_tree_contents([('a', '0123456789')],)
1028
def test_readv_with_short_reads(self):
1029
server = self.get_readonly_server()
1030
t = self._transport(server.get_url())
1031
# Force separate ranges for each offset
1032
t._bytes_to_read_before_seek = 0
1033
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1034
self.assertEqual((0, '0'), ireadv.next())
1035
self.assertEqual((2, '2'), ireadv.next())
1036
if not self._testing_pycurl():
1037
# Only one request have been issued so far (except for pycurl that
1038
# try to read the whole response at once)
1039
self.assertEqual(1, server.GET_request_nb)
1040
self.assertEqual((4, '45'), ireadv.next())
1041
self.assertEqual((9, '9'), ireadv.next())
1042
# Both implementations issue 3 requests but:
1043
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1045
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
1046
self.assertEqual(3, server.GET_request_nb)
1047
# Finally the client have tried a single range request and stays in
1049
self.assertEqual('single', t._range_hint)
1051
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1052
"""Errors out when range specifiers exceed the limit"""
1054
def get_multiple_ranges(self, file, file_size, ranges):
1055
"""Refuses the multiple ranges request"""
1056
tcs = self.server.test_case_server
1057
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1059
# Emulate apache behavior
1060
self.send_error(400, "Bad Request")
1062
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1063
self, file, file_size, ranges)
1066
class LimitedRangeHTTPServer(http_server.HttpServer):
1067
"""An HttpServer erroring out on requests with too much range specifiers"""
1069
def __init__(self, request_handler=LimitedRangeRequestHandler,
1070
protocol_version=None,
1072
http_server.HttpServer.__init__(self, request_handler,
1073
protocol_version=protocol_version)
1074
self.range_limit = range_limit
1077
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1078
"""Tests readv requests against a server erroring out on too much ranges."""
1080
# Requests with more range specifiers will error out
1083
685
def create_transport_readonly_server(self):
1084
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1085
protocol_version=self._protocol_version)
1087
def get_transport(self):
1088
return self._transport(self.get_readonly_server().get_url())
1091
http_utils.TestCaseWithWebserver.setUp(self)
1092
# We need to manipulate ranges that correspond to real chunks in the
1093
# response, so we build a content appropriately.
1094
filler = ''.join(['abcdefghij' for x in range(102)])
1095
content = ''.join(['%04d' % v + filler for v in range(16)])
1096
self.build_tree_contents([('a', content)],)
1098
def test_few_ranges(self):
1099
t = self.get_transport()
1100
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1101
self.assertEqual(l[0], (0, '0000'))
1102
self.assertEqual(l[1], (1024, '0001'))
1103
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1105
def test_more_ranges(self):
1106
t = self.get_transport()
1107
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1108
self.assertEqual(l[0], (0, '0000'))
1109
self.assertEqual(l[1], (1024, '0001'))
1110
self.assertEqual(l[2], (4096, '0004'))
1111
self.assertEqual(l[3], (8192, '0008'))
1112
# The server will refuse to serve the first request (too much ranges),
1113
# a second request will succeed.
1114
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
1117
class TestHttpProxyWhiteBox(tests.TestCase):
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):
1118
703
"""Whitebox test proxy http authorization.
1120
Only the urllib implementation is tested here.
705
These tests concern urllib implementation only.
1123
708
def setUp(self):
1124
tests.TestCase.setUp(self)
1125
710
self._old_env = {}
1127
712
def tearDown(self):
1128
713
self._restore_env()
1129
tests.TestCase.tearDown(self)
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)
1131
719
def _install_env(self, env):
1132
720
for name, value in env.iteritems():
1133
self._old_env[name] = osutils.set_or_unset_env(name, value)
721
self._set_and_capture_env_var(name, value)
1135
723
def _restore_env(self):
1136
724
for name, value in self._old_env.iteritems():
1137
725
osutils.set_or_unset_env(name, value)
1139
727
def _proxied_request(self):
1140
handler = _urllib2_wrappers.ProxyHandler()
1141
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')
1142
735
handler.set_proxy(request, 'http')
1256
859
'NO_PROXY': self.no_proxy_host})
1258
861
def test_http_proxy_without_scheme(self):
1259
if self._testing_pycurl():
1260
# pycurl *ignores* invalid proxy env variables. If that ever change
1261
# in the future, this test will fail indicating that pycurl do not
1262
# ignore anymore such variables.
1263
self.not_proxied_in_env({'http_proxy': self.proxy_address})
1265
self.assertRaises(errors.InvalidURL,
1266
self.proxied_in_env,
1267
{'http_proxy': self.proxy_address})
1270
class TestRanges(http_utils.TestCaseWithWebserver):
1271
"""Test the Range header in GET methods."""
1274
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)
1275
915
self.build_tree_contents([('a', '0123456789')],)
1276
916
server = self.get_readonly_server()
1277
917
self.transport = self._transport(server.get_url())
1279
def create_transport_readonly_server(self):
1280
return http_server.HttpServer(protocol_version=self._protocol_version)
1282
def _file_contents(self, relpath, ranges):
1283
offsets = [ (start, end - start + 1) for start, end in ranges]
1284
coalesce = self.transport._coalesce_offsets
1285
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1286
code, data = self.transport._get(relpath, coalesced)
1287
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1288
for start, end in ranges:
1290
yield data.read(end - start + 1)
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)
1292
926
def _file_tail(self, relpath, tail_amount):
1293
code, data = self.transport._get(relpath, [], tail_amount)
1294
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1295
data.seek(-tail_amount, 2)
1296
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)
1298
932
def test_range_header(self):
1300
934
map(self.assertEqual,['0', '234'],
1301
935
list(self._file_contents('a', [(0,0), (2,4)])),)
1303
def test_range_header_tail(self):
1304
937
self.assertEqual('789', self._file_tail('a', 3))
1306
def test_syntactically_invalid_range_header(self):
1307
self.assertListRaises(errors.InvalidHttpRange,
1308
self._file_contents, 'a', [(4, 3)])
1310
def test_semantically_invalid_range_header(self):
1311
self.assertListRaises(errors.InvalidHttpRange,
1312
self._file_contents, 'a', [(42, 128)])
1315
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1316
"""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.
1318
969
def create_transport_secondary_server(self):
1319
970
"""Create the secondary server redirecting to the primary server"""
1320
971
new = self.get_readonly_server()
1322
redirecting = http_utils.HTTPServerRedirecting(
1323
protocol_version=self._protocol_version)
973
redirecting = HTTPServerRedirecting()
1324
974
redirecting.redirect_to(new.host, new.port)
1325
975
return redirecting
1483
1144
return self.old_transport.clone(exception.target)
1485
self.assertRaises(errors.TooManyRedirections,
1486
transport.do_catching_redirections,
1146
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1487
1147
self.get_a, self.old_transport, redirected)
1490
class TestAuth(http_utils.TestCaseWithWebserver):
1491
"""Test authentication scheme"""
1493
_auth_header = 'Authorization'
1494
_password_prompt_prefix = ''
1495
_username_prompt_prefix = ''
1500
super(TestAuth, self).setUp()
1501
self.server = self.get_readonly_server()
1502
self.build_tree_contents([('a', 'contents of a\n'),
1503
('b', 'contents of b\n'),])
1505
def create_transport_readonly_server(self):
1506
return self._auth_server(protocol_version=self._protocol_version)
1508
def _testing_pycurl(self):
1509
return pycurl_present and self._transport == PyCurlTransport
1511
def get_user_url(self, user, password):
1512
"""Build an url embedding user and password"""
1513
url = '%s://' % self.server._url_protocol
1514
if user is not None:
1516
if password is not None:
1517
url += ':' + password
1519
url += '%s:%s/' % (self.server.host, self.server.port)
1522
def get_user_transport(self, user, password):
1523
return self._transport(self.get_user_url(user, password))
1525
def test_no_user(self):
1526
self.server.add_user('joe', 'foo')
1527
t = self.get_user_transport(None, None)
1528
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1529
# Only one 'Authentication Required' error should occur
1530
self.assertEqual(1, self.server.auth_required_errors)
1532
def test_empty_pass(self):
1533
self.server.add_user('joe', '')
1534
t = self.get_user_transport('joe', '')
1535
self.assertEqual('contents of a\n', t.get('a').read())
1536
# Only one 'Authentication Required' error should occur
1537
self.assertEqual(1, self.server.auth_required_errors)
1539
def test_user_pass(self):
1540
self.server.add_user('joe', 'foo')
1541
t = self.get_user_transport('joe', 'foo')
1542
self.assertEqual('contents of a\n', t.get('a').read())
1543
# Only one 'Authentication Required' error should occur
1544
self.assertEqual(1, self.server.auth_required_errors)
1546
def test_unknown_user(self):
1547
self.server.add_user('joe', 'foo')
1548
t = self.get_user_transport('bill', 'foo')
1549
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1550
# Two 'Authentication Required' errors should occur (the
1551
# initial 'who are you' and 'I don't know you, who are
1553
self.assertEqual(2, self.server.auth_required_errors)
1555
def test_wrong_pass(self):
1556
self.server.add_user('joe', 'foo')
1557
t = self.get_user_transport('joe', 'bar')
1558
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1559
# Two 'Authentication Required' errors should occur (the
1560
# initial 'who are you' and 'this is not you, who are you')
1561
self.assertEqual(2, self.server.auth_required_errors)
1563
def test_prompt_for_username(self):
1564
if self._testing_pycurl():
1565
raise tests.TestNotApplicable(
1566
'pycurl cannot prompt, it handles auth by embedding'
1567
' user:pass in urls only')
1569
self.server.add_user('joe', 'foo')
1570
t = self.get_user_transport(None, None)
1571
stdout = tests.StringIOWrapper()
1572
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n', stdout=stdout)
1573
self.assertEqual('contents of a\n',t.get('a').read())
1574
# stdin should be empty
1575
self.assertEqual('', ui.ui_factory.stdin.readline())
1577
expected_prompt = self._expected_username_prompt(t._unqualified_scheme)
1578
self.assertEquals(expected_prompt, stdout.read(len(expected_prompt)))
1579
self._check_password_prompt(t._unqualified_scheme, 'joe',
1582
def test_prompt_for_password(self):
1583
if self._testing_pycurl():
1584
raise tests.TestNotApplicable(
1585
'pycurl cannot prompt, it handles auth by embedding'
1586
' user:pass in urls only')
1588
self.server.add_user('joe', 'foo')
1589
t = self.get_user_transport('joe', None)
1590
stdout = tests.StringIOWrapper()
1591
ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1592
self.assertEqual('contents of a\n',t.get('a').read())
1593
# stdin should be empty
1594
self.assertEqual('', ui.ui_factory.stdin.readline())
1595
self._check_password_prompt(t._unqualified_scheme, 'joe',
1597
# And we shouldn't prompt again for a different request
1598
# against the same transport.
1599
self.assertEqual('contents of b\n',t.get('b').read())
1601
# And neither against a clone
1602
self.assertEqual('contents of b\n',t2.get('b').read())
1603
# Only one 'Authentication Required' error should occur
1604
self.assertEqual(1, self.server.auth_required_errors)
1606
def _check_password_prompt(self, scheme, user, actual_prompt):
1607
expected_prompt = (self._password_prompt_prefix
1608
+ ("%s %s@%s:%d, Realm: '%s' password: "
1610
user, self.server.host, self.server.port,
1611
self.server.auth_realm)))
1612
self.assertEquals(expected_prompt, actual_prompt)
1614
def _expected_username_prompt(self, scheme):
1615
return (self._username_prompt_prefix
1616
+ "%s %s:%d, Realm: '%s' username: " % (scheme.upper(),
1617
self.server.host, self.server.port,
1618
self.server.auth_realm))
1620
def test_no_prompt_for_password_when_using_auth_config(self):
1621
if self._testing_pycurl():
1622
raise tests.TestNotApplicable(
1623
'pycurl does not support authentication.conf'
1624
' since it cannot prompt')
1628
stdin_content = 'bar\n' # Not the right password
1629
self.server.add_user(user, password)
1630
t = self.get_user_transport(user, None)
1631
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1632
stdout=tests.StringIOWrapper())
1633
# Create a minimal config file with the right password
1634
conf = config.AuthenticationConfig()
1635
conf._get_config().update(
1636
{'httptest': {'scheme': 'http', 'port': self.server.port,
1637
'user': user, 'password': password}})
1639
# Issue a request to the server to connect
1640
self.assertEqual('contents of a\n',t.get('a').read())
1641
# stdin should have been left untouched
1642
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1643
# Only one 'Authentication Required' error should occur
1644
self.assertEqual(1, self.server.auth_required_errors)
1646
def test_user_from_auth_conf(self):
1647
if self._testing_pycurl():
1648
raise tests.TestNotApplicable(
1649
'pycurl does not support authentication.conf')
1652
self.server.add_user(user, password)
1653
# Create a minimal config file with the right password
1654
conf = config.AuthenticationConfig()
1655
conf._get_config().update(
1656
{'httptest': {'scheme': 'http', 'port': self.server.port,
1657
'user': user, 'password': password}})
1659
t = self.get_user_transport(None, None)
1660
# Issue a request to the server to connect
1661
self.assertEqual('contents of a\n', t.get('a').read())
1662
# Only one 'Authentication Required' error should occur
1663
self.assertEqual(1, self.server.auth_required_errors)
1665
def test_changing_nonce(self):
1666
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1667
http_utils.ProxyDigestAuthServer):
1668
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1669
if self._testing_pycurl():
1670
raise tests.KnownFailure(
1671
'pycurl does not handle a nonce change')
1672
self.server.add_user('joe', 'foo')
1673
t = self.get_user_transport('joe', 'foo')
1674
self.assertEqual('contents of a\n', t.get('a').read())
1675
self.assertEqual('contents of b\n', t.get('b').read())
1676
# Only one 'Authentication Required' error should have
1678
self.assertEqual(1, self.server.auth_required_errors)
1679
# The server invalidates the current nonce
1680
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1681
self.assertEqual('contents of a\n', t.get('a').read())
1682
# Two 'Authentication Required' errors should occur (the
1683
# initial 'who are you' and a second 'who are you' with the new nonce)
1684
self.assertEqual(2, self.server.auth_required_errors)
1688
class TestProxyAuth(TestAuth):
1689
"""Test proxy authentication schemes."""
1691
_auth_header = 'Proxy-authorization'
1692
_password_prompt_prefix = 'Proxy '
1693
_username_prompt_prefix = 'Proxy '
1696
super(TestProxyAuth, self).setUp()
1698
self.addCleanup(self._restore_env)
1699
# Override the contents to avoid false positives
1700
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1701
('b', 'not proxied contents of b\n'),
1702
('a-proxied', 'contents of a\n'),
1703
('b-proxied', 'contents of b\n'),
1706
def get_user_transport(self, user, password):
1707
self._install_env({'all_proxy': self.get_user_url(user, password)})
1708
return self._transport(self.server.get_url())
1710
def _install_env(self, env):
1711
for name, value in env.iteritems():
1712
self._old_env[name] = osutils.set_or_unset_env(name, value)
1714
def _restore_env(self):
1715
for name, value in self._old_env.iteritems():
1716
osutils.set_or_unset_env(name, value)
1718
def test_empty_pass(self):
1719
if self._testing_pycurl():
1721
if pycurl.version_info()[1] < '7.16.0':
1722
raise tests.KnownFailure(
1723
'pycurl < 7.16.0 does not handle empty proxy passwords')
1724
super(TestProxyAuth, self).test_empty_pass()
1727
class SampleSocket(object):
1728
"""A socket-like object for use in testing the HTTP request handler."""
1730
def __init__(self, socket_read_content):
1731
"""Constructs a sample socket.
1733
:param socket_read_content: a byte sequence
1735
# Use plain python StringIO so we can monkey-patch the close method to
1736
# not discard the contents.
1737
from StringIO import StringIO
1738
self.readfile = StringIO(socket_read_content)
1739
self.writefile = StringIO()
1740
self.writefile.close = lambda: None
1742
def makefile(self, mode='r', bufsize=None):
1744
return self.readfile
1746
return self.writefile
1749
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1752
super(SmartHTTPTunnellingTest, self).setUp()
1753
# We use the VFS layer as part of HTTP tunnelling tests.
1754
self._captureVar('BZR_NO_SMART_VFS', None)
1755
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1757
def create_transport_readonly_server(self):
1758
return http_utils.HTTPServerWithSmarts(
1759
protocol_version=self._protocol_version)
1761
def test_open_bzrdir(self):
1762
branch = self.make_branch('relpath')
1763
http_server = self.get_readonly_server()
1764
url = http_server.get_url() + 'relpath'
1765
bd = bzrdir.BzrDir.open(url)
1766
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1768
def test_bulk_data(self):
1769
# We should be able to send and receive bulk data in a single message.
1770
# The 'readv' command in the smart protocol both sends and receives
1771
# bulk data, so we use that.
1772
self.build_tree(['data-file'])
1773
http_server = self.get_readonly_server()
1774
http_transport = self._transport(http_server.get_url())
1775
medium = http_transport.get_smart_medium()
1776
# Since we provide the medium, the url below will be mostly ignored
1777
# during the test, as long as the path is '/'.
1778
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1781
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1783
def test_http_send_smart_request(self):
1785
post_body = 'hello\n'
1786
expected_reply_body = 'ok\x012\n'
1788
http_server = self.get_readonly_server()
1789
http_transport = self._transport(http_server.get_url())
1790
medium = http_transport.get_smart_medium()
1791
response = medium.send_http_smart_request(post_body)
1792
reply_body = response.read()
1793
self.assertEqual(expected_reply_body, reply_body)
1795
def test_smart_http_server_post_request_handler(self):
1796
httpd = self.get_readonly_server()._get_httpd()
1798
socket = SampleSocket(
1799
'POST /.bzr/smart %s \r\n' % self._protocol_version
1800
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1802
+ 'Content-Length: 6\r\n'
1805
# Beware: the ('localhost', 80) below is the
1806
# client_address parameter, but we don't have one because
1807
# we have defined a socket which is not bound to an
1808
# address. The test framework never uses this client
1809
# address, so far...
1810
request_handler = http_utils.SmartRequestHandler(socket,
1813
response = socket.writefile.getvalue()
1814
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1815
# This includes the end of the HTTP headers, and all the body.
1816
expected_end_of_response = '\r\n\r\nok\x012\n'
1817
self.assertEndsWith(response, expected_end_of_response)
1820
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1821
"""No smart server here request handler."""
1824
self.send_error(403, "Forbidden")
1827
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1828
"""Test smart client behaviour against an http server without smarts."""
1830
_req_handler_class = ForbiddenRequestHandler
1832
def test_probe_smart_server(self):
1833
"""Test error handling against server refusing smart requests."""
1834
server = self.get_readonly_server()
1835
t = self._transport(server.get_url())
1836
# No need to build a valid smart request here, the server will not even
1837
# try to interpret it.
1838
self.assertRaises(errors.SmartProtocolError,
1839
t.get_smart_medium().send_http_smart_request,
1842
class Test_redirected_to(tests.TestCase):
1844
def test_redirected_to_subdir(self):
1845
t = self._transport('http://www.example.com/foo')
1846
r = t._redirected_to('http://www.example.com/foo',
1847
'http://www.example.com/foo/subdir')
1848
self.assertIsInstance(r, type(t))
1849
# Both transports share the some connection
1850
self.assertEquals(t._get_connection(), r._get_connection())
1852
def test_redirected_to_self_with_slash(self):
1853
t = self._transport('http://www.example.com/foo')
1854
r = t._redirected_to('http://www.example.com/foo',
1855
'http://www.example.com/foo/')
1856
self.assertIsInstance(r, type(t))
1857
# Both transports share the some connection (one can argue that we
1858
# should return the exact same transport here, but that seems
1860
self.assertEquals(t._get_connection(), r._get_connection())
1862
def test_redirected_to_host(self):
1863
t = self._transport('http://www.example.com/foo')
1864
r = t._redirected_to('http://www.example.com/foo',
1865
'http://foo.example.com/foo/subdir')
1866
self.assertIsInstance(r, type(t))
1868
def test_redirected_to_same_host_sibling_protocol(self):
1869
t = self._transport('http://www.example.com/foo')
1870
r = t._redirected_to('http://www.example.com/foo',
1871
'https://www.example.com/foo')
1872
self.assertIsInstance(r, type(t))
1874
def test_redirected_to_same_host_different_protocol(self):
1875
t = self._transport('http://www.example.com/foo')
1876
r = t._redirected_to('http://www.example.com/foo',
1877
'ftp://www.example.com/foo')
1878
self.assertNotEquals(type(r), type(t))
1880
def test_redirected_to_different_host_same_user(self):
1881
t = self._transport('http://joe@www.example.com/foo')
1882
r = t._redirected_to('http://www.example.com/foo',
1883
'https://foo.example.com/foo')
1884
self.assertIsInstance(r, type(t))
1885
self.assertEquals(t._user, r._user)
1888
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1889
"""Request handler for a unique and pre-defined request.
1891
The only thing we care about here is how many bytes travel on the wire. But
1892
since we want to measure it for a real http client, we have to send it
1895
We expect to receive a *single* request nothing more (and we won't even
1896
check what request it is, we just measure the bytes read until an empty
1900
def handle_one_request(self):
1901
tcs = self.server.test_case_server
1902
requestline = self.rfile.readline()
1903
headers = self.MessageClass(self.rfile, 0)
1904
# We just read: the request, the headers, an empty line indicating the
1905
# end of the headers.
1906
bytes_read = len(requestline)
1907
for line in headers.headers:
1908
bytes_read += len(line)
1909
bytes_read += len('\r\n')
1910
if requestline.startswith('POST'):
1911
# The body should be a single line (or we don't know where it ends
1912
# and we don't want to issue a blocking read)
1913
body = self.rfile.readline()
1914
bytes_read += len(body)
1915
tcs.bytes_read = bytes_read
1917
# We set the bytes written *before* issuing the write, the client is
1918
# supposed to consume every produced byte *before* checking that value.
1920
# Doing the oppposite may lead to test failure: we may be interrupted
1921
# after the write but before updating the value. The client can then
1922
# continue and read the value *before* we can update it. And yes,
1923
# this has been observed -- vila 20090129
1924
tcs.bytes_written = len(tcs.canned_response)
1925
self.wfile.write(tcs.canned_response)
1928
class ActivityServerMixin(object):
1930
def __init__(self, protocol_version):
1931
super(ActivityServerMixin, self).__init__(
1932
request_handler=PredefinedRequestHandler,
1933
protocol_version=protocol_version)
1934
# Bytes read and written by the server
1936
self.bytes_written = 0
1937
self.canned_response = None
1940
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1944
if tests.HTTPSServerFeature.available():
1945
from bzrlib.tests import https_server
1946
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1950
class TestActivity(tests.TestCase):
1951
"""Test socket activity reporting.
1953
We use a special purpose server to control the bytes sent and received and
1954
be able to predict the activity on the client socket.
1958
tests.TestCase.setUp(self)
1959
self.server = self._activity_server(self._protocol_version)
1961
self.activities = {}
1962
def report_activity(t, bytes, direction):
1963
count = self.activities.get(direction, 0)
1965
self.activities[direction] = count
1967
# We override at class level because constructors may propagate the
1968
# bound method and render instance overriding ineffective (an
1969
# alternative would be to define a specific ui factory instead...)
1970
self.orig_report_activity = self._transport._report_activity
1971
self._transport._report_activity = report_activity
1974
self._transport._report_activity = self.orig_report_activity
1975
self.server.tearDown()
1976
tests.TestCase.tearDown(self)
1978
def get_transport(self):
1979
return self._transport(self.server.get_url())
1981
def assertActivitiesMatch(self):
1982
self.assertEqual(self.server.bytes_read,
1983
self.activities.get('write', 0), 'written bytes')
1984
self.assertEqual(self.server.bytes_written,
1985
self.activities.get('read', 0), 'read bytes')
1988
self.server.canned_response = '''HTTP/1.1 200 OK\r
1989
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
1990
Server: Apache/2.0.54 (Fedora)\r
1991
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
1992
ETag: "56691-23-38e9ae00"\r
1993
Accept-Ranges: bytes\r
1994
Content-Length: 35\r
1996
Content-Type: text/plain; charset=UTF-8\r
1998
Bazaar-NG meta directory, format 1
2000
t = self.get_transport()
2001
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2002
t.get('foo/bar').read())
2003
self.assertActivitiesMatch()
2006
self.server.canned_response = '''HTTP/1.1 200 OK\r
2007
Server: SimpleHTTP/0.6 Python/2.5.2\r
2008
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2009
Content-type: application/octet-stream\r
2010
Content-Length: 20\r
2011
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2014
t = self.get_transport()
2015
self.assertTrue(t.has('foo/bar'))
2016
self.assertActivitiesMatch()
2018
def test_readv(self):
2019
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2020
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2021
Server: Apache/2.0.54 (Fedora)\r
2022
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2023
ETag: "238a3c-16ec2-805c5540"\r
2024
Accept-Ranges: bytes\r
2025
Content-Length: 1534\r
2027
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2030
--418470f848b63279b\r
2031
Content-type: text/plain; charset=UTF-8\r
2032
Content-range: bytes 0-254/93890\r
2034
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2035
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2036
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2037
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2038
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2040
--418470f848b63279b\r
2041
Content-type: text/plain; charset=UTF-8\r
2042
Content-range: bytes 1000-2049/93890\r
2045
mbp@sourcefrog.net-20050311063625-07858525021f270b
2046
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2047
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2048
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2049
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2050
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2051
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2052
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2053
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2054
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2055
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2056
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2057
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2058
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2059
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2060
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2061
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2062
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2063
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2064
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2066
--418470f848b63279b--\r
2068
t = self.get_transport()
2069
# Remember that the request is ignored and that the ranges below
2070
# doesn't have to match the canned response.
2071
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2072
self.assertEqual(2, len(l))
2073
self.assertActivitiesMatch()
2075
def test_post(self):
2076
self.server.canned_response = '''HTTP/1.1 200 OK\r
2077
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2078
Server: Apache/2.0.54 (Fedora)\r
2079
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2080
ETag: "56691-23-38e9ae00"\r
2081
Accept-Ranges: bytes\r
2082
Content-Length: 35\r
2084
Content-Type: text/plain; charset=UTF-8\r
2086
lalala whatever as long as itsssss
2088
t = self.get_transport()
2089
# We must send a single line of body bytes, see
2090
# PredefinedRequestHandler.handle_one_request
2091
code, f = t._post('abc def end-of-body\n')
2092
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2093
self.assertActivitiesMatch()
added
removed
bzrlib/tests/test_http.py
