14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 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.
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
23
from cStringIO import StringIO
30
import SimpleHTTPServer
36
31
from bzrlib import (
41
remote as _mod_remote,
47
38
from bzrlib.tests import (
44
from bzrlib.tests.HttpServer import (
49
from bzrlib.tests.HTTPTestUtil import (
50
BadProtocolRequestHandler,
51
BadStatusRequestHandler,
52
ForbiddenRequestHandler,
55
HTTPServerRedirecting,
56
InvalidStatusRequestHandler,
57
LimitedRangeHTTPServer,
58
NoRangeRequestHandler,
60
ProxyDigestAuthServer,
62
SingleRangeRequestHandler,
63
SingleOnlyRangeRequestHandler,
64
TestCaseWithRedirectedWebserver,
65
TestCaseWithTwoWebservers,
66
TestCaseWithWebserver,
51
69
from bzrlib.transport import (
71
do_catching_redirections,
55
75
from bzrlib.transport.http import (
62
from bzrlib.transport.http._pycurl import PyCurlTransport
64
except errors.DependencyNotPresent:
65
pycurl_present = False
68
class TransportAdapter(tests.TestScenarioApplier):
69
"""Generate the same test for each transport implementation."""
72
transport_scenarios = [
73
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
74
_server=http_server.HttpServer_urllib,
75
_qualified_prefix='http+urllib',)),
78
transport_scenarios.append(
79
('pycurl', dict(_transport=PyCurlTransport,
80
_server=http_server.HttpServer_PyCurl,
81
_qualified_prefix='http+pycurl',)))
82
self.scenarios = transport_scenarios
85
class TransportProtocolAdapter(TransportAdapter):
86
"""Generate the same test for each protocol implementation.
88
In addition to the transport adaptatation that we inherit from.
92
super(TransportProtocolAdapter, self).__init__()
93
protocol_scenarios = [
94
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
95
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
97
self.scenarios = tests.multiply_scenarios(self.scenarios,
101
class TransportProtocolAuthenticationAdapter(TransportProtocolAdapter):
102
"""Generate the same test for each authentication scheme implementation.
104
In addition to the protocol adaptatation that we inherit from.
108
super(TransportProtocolAuthenticationAdapter, self).__init__()
109
auth_scheme_scenarios = [
110
('basic', dict(_auth_scheme='basic')),
111
('digest', dict(_auth_scheme='digest')),
114
self.scenarios = tests.multiply_scenarios(self.scenarios,
115
auth_scheme_scenarios)
117
def load_tests(standard_tests, module, loader):
118
"""Multiply tests for http clients and protocol versions."""
119
# one for each transport
120
t_adapter = TransportAdapter()
121
t_classes= (TestHttpTransportRegistration,
122
TestHttpTransportUrls,
124
is_testing_for_transports = tests.condition_isinstance(t_classes)
126
# multiplied by one for each protocol version
127
tp_adapter = TransportProtocolAdapter()
128
tp_classes= (SmartHTTPTunnellingTest,
129
TestDoCatchRedirections,
131
TestHTTPRedirections,
132
TestHTTPSilentRedirections,
133
TestLimitedRangeRequestServer,
137
TestSpecificRequestHandler,
139
is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)
141
# multiplied by one for each authentication scheme
142
tpa_adapter = TransportProtocolAuthenticationAdapter()
143
tpa_classes = (TestAuth,
145
is_also_testing_for_authentication = tests.condition_isinstance(
148
result = loader.suiteClass()
149
for test_class in tests.iter_suite_tests(standard_tests):
150
# Each test class is either standalone or testing for some combination
151
# of transport, protocol version, authentication scheme. Use the right
152
# adpater (or none) depending on the class.
153
if is_testing_for_transports(test_class):
154
result.addTests(t_adapter.adapt(test_class))
155
elif is_also_testing_for_protocols(test_class):
156
result.addTests(tp_adapter.adapt(test_class))
157
elif is_also_testing_for_authentication(test_class):
158
result.addTests(tpa_adapter.adapt(test_class))
160
result.addTest(test_class)
80
from bzrlib.transport.http._urllib import HttpTransport_urllib
81
from bzrlib.transport.http._urllib2_wrappers import (
164
87
class FakeManager(object):
769
653
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
770
654
t.readv, 'a', [(12,2)])
772
def test_readv_multiple_get_requests(self):
773
server = self.get_readonly_server()
774
t = self._transport(server.get_url())
775
# force transport to issue multiple requests
776
t._max_readv_combine = 1
777
t._max_get_ranges = 1
778
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
779
self.assertEqual(l[0], (0, '0'))
780
self.assertEqual(l[1], (1, '1'))
781
self.assertEqual(l[2], (3, '34'))
782
self.assertEqual(l[3], (9, '9'))
783
# The server should have issued 4 requests
784
self.assertEqual(4, server.GET_request_nb)
786
def test_readv_get_max_size(self):
787
server = self.get_readonly_server()
788
t = self._transport(server.get_url())
789
# force transport to issue multiple requests by limiting the number of
790
# bytes by request. Note that this apply to coalesced offsets only, a
791
# single range will keep its size even if bigger than the limit.
793
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
794
self.assertEqual(l[0], (0, '0'))
795
self.assertEqual(l[1], (1, '1'))
796
self.assertEqual(l[2], (2, '2345'))
797
self.assertEqual(l[3], (6, '6789'))
798
# The server should have issued 3 requests
799
self.assertEqual(3, server.GET_request_nb)
801
def test_complete_readv_leave_pipe_clean(self):
802
server = self.get_readonly_server()
803
t = self._transport(server.get_url())
804
# force transport to issue multiple requests
806
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
807
# The server should have issued 3 requests
808
self.assertEqual(3, server.GET_request_nb)
809
self.assertEqual('0123456789', t.get_bytes('a'))
810
self.assertEqual(4, server.GET_request_nb)
812
def test_incomplete_readv_leave_pipe_clean(self):
813
server = self.get_readonly_server()
814
t = self._transport(server.get_url())
815
# force transport to issue multiple requests
817
# Don't collapse readv results into a list so that we leave unread
818
# bytes on the socket
819
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
820
self.assertEqual((0, '0'), ireadv.next())
821
# The server should have issued one request so far
822
self.assertEqual(1, server.GET_request_nb)
823
self.assertEqual('0123456789', t.get_bytes('a'))
824
# get_bytes issued an additional request, the readv pending ones are
826
self.assertEqual(2, server.GET_request_nb)
829
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
830
"""Always reply to range request as if they were single.
832
Don't be explicit about it, just to annoy the clients.
835
def get_multiple_ranges(self, file, file_size, ranges):
836
"""Answer as if it was a single range request and ignores the rest"""
837
(start, end) = ranges[0]
838
return self.get_single_range(file, file_size, start, end)
841
657
class TestSingleRangeRequestServer(TestRangeRequestServer):
842
658
"""Test readv against a server which accept only single range requests"""
844
_req_handler_class = SingleRangeRequestHandler
847
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
848
"""Only reply to simple range requests, errors out on multiple"""
850
def get_multiple_ranges(self, file, file_size, ranges):
851
"""Refuses the multiple ranges request"""
854
self.send_error(416, "Requested range not satisfiable")
856
(start, end) = ranges[0]
857
return self.get_single_range(file, file_size, start, end)
660
def create_transport_readonly_server(self):
661
return HttpServer(SingleRangeRequestHandler)
664
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
665
TestCaseWithWebserver):
666
"""Tests single range requests accepting server for urllib implementation"""
668
_transport = HttpTransport_urllib
671
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
672
TestSingleRangeRequestServer,
673
TestCaseWithWebserver):
674
"""Tests single range requests accepting server for pycurl implementation"""
860
677
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
861
678
"""Test readv against a server which only accept single range requests"""
863
_req_handler_class = SingleOnlyRangeRequestHandler
866
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
867
"""Ignore range requests without notice"""
870
# Update the statistics
871
self.server.test_case_server.GET_request_nb += 1
872
# Just bypass the range handling done by TestingHTTPRequestHandler
873
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleOnlyRangeRequestHandler)
684
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
688
_transport = HttpTransport_urllib
691
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleOnlyRangeRequestServer,
693
TestCaseWithWebserver):
694
"""Tests single range requests accepting server for pycurl implementation"""
876
697
class TestNoRangeRequestServer(TestRangeRequestServer):
877
698
"""Test readv against a server which do not accept range requests"""
879
_req_handler_class = NoRangeRequestHandler
882
class MultipleRangeWithoutContentLengthRequestHandler(
883
http_server.TestingHTTPRequestHandler):
884
"""Reply to multiple range requests without content length header."""
886
def get_multiple_ranges(self, file, file_size, ranges):
887
self.send_response(206)
888
self.send_header('Accept-Ranges', 'bytes')
889
boundary = "%d" % random.randint(0,0x7FFFFFFF)
890
self.send_header("Content-Type",
891
"multipart/byteranges; boundary=%s" % boundary)
893
for (start, end) in ranges:
894
self.wfile.write("--%s\r\n" % boundary)
895
self.send_header("Content-type", 'application/octet-stream')
896
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
900
self.send_range_content(file, start, end - start + 1)
902
self.wfile.write("--%s\r\n" % boundary)
905
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
907
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
910
class TruncatedMultipleRangeRequestHandler(
911
http_server.TestingHTTPRequestHandler):
912
"""Reply to multiple range requests truncating the last ones.
914
This server generates responses whose Content-Length describes all the
915
ranges, but fail to include the last ones leading to client short reads.
916
This has been observed randomly with lighttpd (bug #179368).
700
def create_transport_readonly_server(self):
701
return HttpServer(NoRangeRequestHandler)
704
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests range requests refusing server for urllib implementation"""
708
_transport = HttpTransport_urllib
711
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestNoRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests range requests refusing server for pycurl implementation"""
717
class TestLimitedRangeRequestServer(object):
718
"""Tests readv requests against server that errors out on too much ranges.
720
This MUST be used by daughter classes that also inherit from
721
TestCaseWithWebserver.
723
We can't inherit directly from TestCaseWithWebserver or the
724
test framework will try to create an instance which cannot
725
run, its implementation being incomplete.
919
_truncated_ranges = 2
921
def get_multiple_ranges(self, file, file_size, ranges):
922
self.send_response(206)
923
self.send_header('Accept-Ranges', 'bytes')
925
self.send_header('Content-Type',
926
'multipart/byteranges; boundary=%s' % boundary)
927
boundary_line = '--%s\r\n' % boundary
928
# Calculate the Content-Length
930
for (start, end) in ranges:
931
content_length += len(boundary_line)
932
content_length += self._header_line_length(
933
'Content-type', 'application/octet-stream')
934
content_length += self._header_line_length(
935
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
936
content_length += len('\r\n') # end headers
937
content_length += end - start # + 1
938
content_length += len(boundary_line)
939
self.send_header('Content-length', content_length)
942
# Send the multipart body
944
for (start, end) in ranges:
945
self.wfile.write(boundary_line)
946
self.send_header('Content-type', 'application/octet-stream')
947
self.send_header('Content-Range', 'bytes %d-%d/%d'
948
% (start, end, file_size))
950
if cur + self._truncated_ranges >= len(ranges):
951
# Abruptly ends the response and close the connection
952
self.close_connection = 1
954
self.send_range_content(file, start, end - start + 1)
957
self.wfile.write(boundary_line)
960
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
962
_req_handler_class = TruncatedMultipleRangeRequestHandler
965
super(TestTruncatedMultipleRangeServer, self).setUp()
966
self.build_tree_contents([('a', '0123456789')],)
968
def test_readv_with_short_reads(self):
969
server = self.get_readonly_server()
970
t = self._transport(server.get_url())
971
# Force separate ranges for each offset
972
t._bytes_to_read_before_seek = 0
973
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
974
self.assertEqual((0, '0'), ireadv.next())
975
self.assertEqual((2, '2'), ireadv.next())
976
if not self._testing_pycurl():
977
# Only one request have been issued so far (except for pycurl that
978
# try to read the whole response at once)
979
self.assertEqual(1, server.GET_request_nb)
980
self.assertEqual((4, '45'), ireadv.next())
981
self.assertEqual((9, '9'), ireadv.next())
982
# Both implementations issue 3 requests but:
983
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
985
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
986
self.assertEqual(3, server.GET_request_nb)
987
# Finally the client have tried a single range request and stays in
989
self.assertEqual('single', t._range_hint)
991
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
992
"""Errors out when range specifiers exceed the limit"""
994
def get_multiple_ranges(self, file, file_size, ranges):
995
"""Refuses the multiple ranges request"""
996
tcs = self.server.test_case_server
997
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
999
# Emulate apache behavior
1000
self.send_error(400, "Bad Request")
1002
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1003
self, file, file_size, ranges)
1006
class LimitedRangeHTTPServer(http_server.HttpServer):
1007
"""An HttpServer erroring out on requests with too much range specifiers"""
1009
def __init__(self, request_handler=LimitedRangeRequestHandler,
1010
protocol_version=None,
1012
http_server.HttpServer.__init__(self, request_handler,
1013
protocol_version=protocol_version)
1014
self.range_limit = range_limit
1017
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1018
"""Tests readv requests against a server erroring out on too much ranges."""
1020
# Requests with more range specifiers will error out
1023
730
def create_transport_readonly_server(self):
1024
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1025
protocol_version=self._protocol_version)
731
# Requests with more range specifiers will error out
732
return LimitedRangeHTTPServer(range_limit=self.range_limit)
1027
734
def get_transport(self):
1028
735
return self._transport(self.get_readonly_server().get_url())
1030
737
def setUp(self):
1031
http_utils.TestCaseWithWebserver.setUp(self)
738
TestCaseWithWebserver.setUp(self)
1032
739
# We need to manipulate ranges that correspond to real chunks in the
1033
740
# response, so we build a content appropriately.
1034
filler = ''.join(['abcdefghij' for x in range(102)])
741
filler = ''.join(['abcdefghij' for _ in range(102)])
1035
742
content = ''.join(['%04d' % v + filler for v in range(16)])
1036
743
self.build_tree_contents([('a', content)],)
1196
906
'NO_PROXY': self.no_proxy_host})
1198
908
def test_http_proxy_without_scheme(self):
1199
if self._testing_pycurl():
1200
# pycurl *ignores* invalid proxy env variables. If that ever change
1201
# in the future, this test will fail indicating that pycurl do not
1202
# ignore anymore such variables.
1203
self.not_proxied_in_env({'http_proxy': self.proxy_address})
1205
self.assertRaises(errors.InvalidURL,
1206
self.proxied_in_env,
1207
{'http_proxy': self.proxy_address})
1210
class TestRanges(http_utils.TestCaseWithWebserver):
1211
"""Test the Range header in GET methods."""
1214
http_utils.TestCaseWithWebserver.setUp(self)
909
self.assertRaises(errors.InvalidURL,
911
{'http_proxy': self.proxy_address})
914
class TestProxyHttpServer_urllib(TestProxyHttpServer,
915
TestCaseWithTwoWebservers):
916
"""Tests proxy server for urllib implementation"""
918
_transport = HttpTransport_urllib
921
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
923
TestCaseWithTwoWebservers):
924
"""Tests proxy server for pycurl implementation"""
927
TestProxyHttpServer.setUp(self)
928
# Oh my ! pycurl does not check for the port as part of
929
# no_proxy :-( So we just test the host part
930
self.no_proxy_host = 'localhost'
932
def test_HTTP_PROXY(self):
933
# pycurl does not check HTTP_PROXY for security reasons
934
# (for use in a CGI context that we do not care
935
# about. Should we ?)
936
raise TestSkipped('pycurl does not check HTTP_PROXY '
937
'for security reasons')
939
def test_HTTP_PROXY_with_NO_PROXY(self):
940
raise TestSkipped('pycurl does not check HTTP_PROXY '
941
'for security reasons')
943
def test_http_proxy_without_scheme(self):
944
# pycurl *ignores* invalid proxy env variables. If that
945
# ever change in the future, this test will fail
946
# indicating that pycurl do not ignore anymore such
948
self.not_proxied_in_env({'http_proxy': self.proxy_address})
951
class TestRanges(object):
952
"""Test the Range header in GET methods..
954
This MUST be used by daughter classes that also inherit from
955
TestCaseWithWebserver.
957
We can't inherit directly from TestCaseWithWebserver or the
958
test framework will try to create an instance which cannot
959
run, its implementation being incomplete.
963
TestCaseWithWebserver.setUp(self)
1215
964
self.build_tree_contents([('a', '0123456789')],)
1216
965
server = self.get_readonly_server()
1217
966
self.transport = self._transport(server.get_url())
1219
def create_transport_readonly_server(self):
1220
return http_server.HttpServer(protocol_version=self._protocol_version)
1222
968
def _file_contents(self, relpath, ranges):
1223
969
offsets = [ (start, end - start + 1) for start, end in ranges]
1224
970
coalesce = self.transport._coalesce_offsets
1232
978
def _file_tail(self, relpath, tail_amount):
1233
979
code, data = self.transport._get(relpath, [], tail_amount)
1234
980
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1235
data.seek(-tail_amount, 2)
981
data.seek(-tail_amount + 1, 2)
1236
982
return data.read(tail_amount)
1238
984
def test_range_header(self):
1240
986
map(self.assertEqual,['0', '234'],
1241
987
list(self._file_contents('a', [(0,0), (2,4)])),)
1243
def test_range_header_tail(self):
1244
989
self.assertEqual('789', self._file_tail('a', 3))
1246
def test_syntactically_invalid_range_header(self):
1247
self.assertListRaises(errors.InvalidHttpRange,
990
# Syntactically invalid range
991
self.assertListRaises(errors.InvalidRange,
1248
992
self._file_contents, 'a', [(4, 3)])
1250
def test_semantically_invalid_range_header(self):
1251
self.assertListRaises(errors.InvalidHttpRange,
993
# Semantically invalid range
994
self.assertListRaises(errors.InvalidRange,
1252
995
self._file_contents, 'a', [(42, 128)])
1255
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1256
"""Test redirection between http servers."""
998
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
999
"""Test the Range header in GET methods for urllib implementation"""
1001
_transport = HttpTransport_urllib
1004
class TestRanges_pycurl(TestWithTransport_pycurl,
1006
TestCaseWithWebserver):
1007
"""Test the Range header in GET methods for pycurl implementation"""
1010
class TestHTTPRedirections(object):
1011
"""Test redirection between http servers.
1013
This MUST be used by daughter classes that also inherit from
1014
TestCaseWithRedirectedWebserver.
1016
We can't inherit directly from TestCaseWithTwoWebservers or the
1017
test framework will try to create an instance which cannot
1018
run, its implementation being incomplete.
1258
1021
def create_transport_secondary_server(self):
1259
1022
"""Create the secondary server redirecting to the primary server"""
1260
1023
new = self.get_readonly_server()
1262
redirecting = http_utils.HTTPServerRedirecting(
1263
protocol_version=self._protocol_version)
1025
redirecting = HTTPServerRedirecting()
1264
1026
redirecting.redirect_to(new.host, new.port)
1265
1027
return redirecting
1626
1370
for name, value in self._old_env.iteritems():
1627
1371
osutils.set_or_unset_env(name, value)
1629
def test_empty_pass(self):
1630
if self._testing_pycurl():
1632
if pycurl.version_info()[1] < '7.16.0':
1633
raise tests.KnownFailure(
1634
'pycurl < 7.16.0 does not handle empty proxy passwords')
1635
super(TestProxyAuth, self).test_empty_pass()
1638
class SampleSocket(object):
1639
"""A socket-like object for use in testing the HTTP request handler."""
1641
def __init__(self, socket_read_content):
1642
"""Constructs a sample socket.
1644
:param socket_read_content: a byte sequence
1646
# Use plain python StringIO so we can monkey-patch the close method to
1647
# not discard the contents.
1648
from StringIO import StringIO
1649
self.readfile = StringIO(socket_read_content)
1650
self.writefile = StringIO()
1651
self.writefile.close = lambda: None
1653
def makefile(self, mode='r', bufsize=None):
1655
return self.readfile
1657
return self.writefile
1660
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1663
super(SmartHTTPTunnellingTest, self).setUp()
1664
# We use the VFS layer as part of HTTP tunnelling tests.
1665
self._captureVar('BZR_NO_SMART_VFS', None)
1666
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1668
def create_transport_readonly_server(self):
1669
return http_utils.HTTPServerWithSmarts(
1670
protocol_version=self._protocol_version)
1672
def test_open_bzrdir(self):
1673
branch = self.make_branch('relpath')
1674
http_server = self.get_readonly_server()
1675
url = http_server.get_url() + 'relpath'
1676
bd = bzrdir.BzrDir.open(url)
1677
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1679
def test_bulk_data(self):
1680
# We should be able to send and receive bulk data in a single message.
1681
# The 'readv' command in the smart protocol both sends and receives
1682
# bulk data, so we use that.
1683
self.build_tree(['data-file'])
1684
http_server = self.get_readonly_server()
1685
http_transport = self._transport(http_server.get_url())
1686
medium = http_transport.get_smart_medium()
1687
# Since we provide the medium, the url below will be mostly ignored
1688
# during the test, as long as the path is '/'.
1689
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1692
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1694
def test_http_send_smart_request(self):
1696
post_body = 'hello\n'
1697
expected_reply_body = 'ok\x012\n'
1699
http_server = self.get_readonly_server()
1700
http_transport = self._transport(http_server.get_url())
1701
medium = http_transport.get_smart_medium()
1702
response = medium.send_http_smart_request(post_body)
1703
reply_body = response.read()
1704
self.assertEqual(expected_reply_body, reply_body)
1706
def test_smart_http_server_post_request_handler(self):
1707
httpd = self.get_readonly_server()._get_httpd()
1709
socket = SampleSocket(
1710
'POST /.bzr/smart %s \r\n' % self._protocol_version
1711
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1713
+ 'Content-Length: 6\r\n'
1716
# Beware: the ('localhost', 80) below is the
1717
# client_address parameter, but we don't have one because
1718
# we have defined a socket which is not bound to an
1719
# address. The test framework never uses this client
1720
# address, so far...
1721
request_handler = http_utils.SmartRequestHandler(socket,
1724
response = socket.writefile.getvalue()
1725
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1726
# This includes the end of the HTTP headers, and all the body.
1727
expected_end_of_response = '\r\n\r\nok\x012\n'
1728
self.assertEndsWith(response, expected_end_of_response)
1731
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1732
"""No smart server here request handler."""
1735
self.send_error(403, "Forbidden")
1738
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1739
"""Test smart client behaviour against an http server without smarts."""
1741
_req_handler_class = ForbiddenRequestHandler
1743
def test_probe_smart_server(self):
1744
"""Test error handling against server refusing smart requests."""
1745
server = self.get_readonly_server()
1746
t = self._transport(server.get_url())
1747
# No need to build a valid smart request here, the server will not even
1748
# try to interpret it.
1749
self.assertRaises(errors.SmartProtocolError,
1750
t.send_http_smart_request, 'whatever')
1374
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1375
"""Test http basic authentication scheme"""
1377
_transport = HttpTransport_urllib
1379
def create_transport_readonly_server(self):
1380
return HTTPBasicAuthServer()
1383
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1384
"""Test proxy basic authentication scheme"""
1386
_transport = HttpTransport_urllib
1388
def create_transport_readonly_server(self):
1389
return ProxyBasicAuthServer()
1392
class TestDigestAuth(object):
1393
"""Digest Authentication specific tests"""
1395
def test_changing_nonce(self):
1396
self.server.add_user('joe', 'foo')
1397
t = self.get_user_transport('joe', 'foo')
1398
self.assertEqual('contents of a\n', t.get('a').read())
1399
self.assertEqual('contents of b\n', t.get('b').read())
1400
# Only one 'Authentication Required' error should have
1402
self.assertEqual(1, self.server.auth_required_errors)
1403
# The server invalidates the current nonce
1404
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1405
self.assertEqual('contents of a\n', t.get('a').read())
1406
# Two 'Authentication Required' errors should occur (the
1407
# initial 'who are you' and a second 'who are you' with the new nonce)
1408
self.assertEqual(2, self.server.auth_required_errors)
1411
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1412
"""Test http digest authentication scheme"""
1414
_transport = HttpTransport_urllib
1416
def create_transport_readonly_server(self):
1417
return HTTPDigestAuthServer()
1420
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1421
TestCaseWithWebserver):
1422
"""Test proxy digest authentication scheme"""
1424
_transport = HttpTransport_urllib
1426
def create_transport_readonly_server(self):
1427
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
