343
623
self.assertEqual('HTTP/1.1 200 OK\r\n',
344
624
osutils.recv_all(sock, 4096))
345
625
self.assertEqual('abc', server.received_bytes)
628
class TestRangeRequestServer(object):
629
"""Tests readv requests against server.
631
This MUST be used by daughter classes that also inherit from
632
TestCaseWithWebserver.
634
We can't inherit directly from TestCaseWithWebserver or the
635
test framework will try to create an instance which cannot
636
run, its implementation being incomplete.
640
TestCaseWithWebserver.setUp(self)
641
self.build_tree_contents([('a', '0123456789')],)
643
def test_readv(self):
644
server = self.get_readonly_server()
645
t = self._transport(server.get_url())
646
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
647
self.assertEqual(l[0], (0, '0'))
648
self.assertEqual(l[1], (1, '1'))
649
self.assertEqual(l[2], (3, '34'))
650
self.assertEqual(l[3], (9, '9'))
652
def test_readv_out_of_order(self):
653
server = self.get_readonly_server()
654
t = self._transport(server.get_url())
655
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
656
self.assertEqual(l[0], (1, '1'))
657
self.assertEqual(l[1], (9, '9'))
658
self.assertEqual(l[2], (0, '0'))
659
self.assertEqual(l[3], (3, '34'))
661
def test_readv_invalid_ranges(self):
662
server = self.get_readonly_server()
663
t = self._transport(server.get_url())
665
# This is intentionally reading off the end of the file
666
# since we are sure that it cannot get there
667
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
668
t.readv, 'a', [(1,1), (8,10)])
670
# This is trying to seek past the end of the file, it should
671
# also raise a special error
672
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
673
t.readv, 'a', [(12,2)])
676
class TestSingleRangeRequestServer(TestRangeRequestServer):
677
"""Test readv against a server which accept only single range requests"""
679
def create_transport_readonly_server(self):
680
return HttpServer(SingleRangeRequestHandler)
683
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
684
TestCaseWithWebserver):
685
"""Tests single range requests accepting server for urllib implementation"""
687
_transport = HttpTransport_urllib
690
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
691
TestSingleRangeRequestServer,
692
TestCaseWithWebserver):
693
"""Tests single range requests accepting server for pycurl implementation"""
696
class TestNoRangeRequestServer(TestRangeRequestServer):
697
"""Test readv against a server which do not accept range requests"""
699
def create_transport_readonly_server(self):
700
return HttpServer(NoRangeRequestHandler)
703
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
704
TestCaseWithWebserver):
705
"""Tests range requests refusing server for urllib implementation"""
707
_transport = HttpTransport_urllib
710
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
711
TestNoRangeRequestServer,
712
TestCaseWithWebserver):
713
"""Tests range requests refusing server for pycurl implementation"""
716
class TestHttpProxyWhiteBox(TestCase):
717
"""Whitebox test proxy http authorization.
719
Only the urllib implementation is tested here.
729
def _install_env(self, env):
730
for name, value in env.iteritems():
731
self._old_env[name] = osutils.set_or_unset_env(name, value)
733
def _restore_env(self):
734
for name, value in self._old_env.iteritems():
735
osutils.set_or_unset_env(name, value)
737
def _proxied_request(self):
738
handler = ProxyHandler(PasswordManager())
739
request = Request('GET','http://baz/buzzle')
740
handler.set_proxy(request, 'http')
743
def test_empty_user(self):
744
self._install_env({'http_proxy': 'http://bar.com'})
745
request = self._proxied_request()
746
self.assertFalse(request.headers.has_key('Proxy-authorization'))
748
def test_invalid_proxy(self):
749
"""A proxy env variable without scheme"""
750
self._install_env({'http_proxy': 'host:1234'})
751
self.assertRaises(errors.InvalidURL, self._proxied_request)
754
class TestProxyHttpServer(object):
755
"""Tests proxy server.
757
This MUST be used by daughter classes that also inherit from
758
TestCaseWithTwoWebservers.
760
We can't inherit directly from TestCaseWithTwoWebservers or
761
the test framework will try to create an instance which
762
cannot run, its implementation being incomplete.
764
Be aware that we do not setup a real proxy here. Instead, we
765
check that the *connection* goes through the proxy by serving
766
different content (the faked proxy server append '-proxied'
770
# FIXME: We don't have an https server available, so we don't
771
# test https connections.
773
# FIXME: Once the test suite is better fitted to test
774
# authorization schemes, test proxy authorizations too (see
778
TestCaseWithTwoWebservers.setUp(self)
779
self.build_tree_contents([('foo', 'contents of foo\n'),
780
('foo-proxied', 'proxied contents of foo\n')])
781
# Let's setup some attributes for tests
782
self.server = self.get_readonly_server()
783
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
784
self.no_proxy_host = self.proxy_address
785
# The secondary server is the proxy
786
self.proxy = self.get_secondary_server()
787
self.proxy_url = self.proxy.get_url()
790
def create_transport_secondary_server(self):
791
"""Creates an http server that will serve files with
792
'-proxied' appended to their names.
796
def _install_env(self, env):
797
for name, value in env.iteritems():
798
self._old_env[name] = osutils.set_or_unset_env(name, value)
800
def _restore_env(self):
801
for name, value in self._old_env.iteritems():
802
osutils.set_or_unset_env(name, value)
804
def proxied_in_env(self, env):
805
self._install_env(env)
806
url = self.server.get_url()
807
t = self._transport(url)
809
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
813
def not_proxied_in_env(self, env):
814
self._install_env(env)
815
url = self.server.get_url()
816
t = self._transport(url)
818
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
822
def test_http_proxy(self):
823
self.proxied_in_env({'http_proxy': self.proxy_url})
825
def test_HTTP_PROXY(self):
826
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
828
def test_all_proxy(self):
829
self.proxied_in_env({'all_proxy': self.proxy_url})
831
def test_ALL_PROXY(self):
832
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
834
def test_http_proxy_with_no_proxy(self):
835
self.not_proxied_in_env({'http_proxy': self.proxy_url,
836
'no_proxy': self.no_proxy_host})
838
def test_HTTP_PROXY_with_NO_PROXY(self):
839
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
840
'NO_PROXY': self.no_proxy_host})
842
def test_all_proxy_with_no_proxy(self):
843
self.not_proxied_in_env({'all_proxy': self.proxy_url,
844
'no_proxy': self.no_proxy_host})
846
def test_ALL_PROXY_with_NO_PROXY(self):
847
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
848
'NO_PROXY': self.no_proxy_host})
850
def test_http_proxy_without_scheme(self):
851
self.assertRaises(errors.InvalidURL,
853
{'http_proxy': self.proxy_address})
856
class TestProxyHttpServer_urllib(TestProxyHttpServer,
857
TestCaseWithTwoWebservers):
858
"""Tests proxy server for urllib implementation"""
860
_transport = HttpTransport_urllib
863
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
865
TestCaseWithTwoWebservers):
866
"""Tests proxy server for pycurl implementation"""
869
TestProxyHttpServer.setUp(self)
870
# Oh my ! pycurl does not check for the port as part of
871
# no_proxy :-( So we just test the host part
872
self.no_proxy_host = 'localhost'
874
def test_HTTP_PROXY(self):
875
# pycurl do not check HTTP_PROXY for security reasons
876
# (for use in a CGI context that we do not care
877
# about. Should we ?)
880
def test_HTTP_PROXY_with_NO_PROXY(self):
883
def test_http_proxy_without_scheme(self):
884
# pycurl *ignores* invalid proxy env variables. If that
885
# ever change in the future, this test will fail
886
# indicating that pycurl do not ignore anymore such
888
self.not_proxied_in_env({'http_proxy': self.proxy_address})
891
class TestRanges(object):
892
"""Test the Range header in GET methods..
894
This MUST be used by daughter classes that also inherit from
895
TestCaseWithWebserver.
897
We can't inherit directly from TestCaseWithWebserver or the
898
test framework will try to create an instance which cannot
899
run, its implementation being incomplete.
903
TestCaseWithWebserver.setUp(self)
904
self.build_tree_contents([('a', '0123456789')],)
905
server = self.get_readonly_server()
906
self.transport = self._transport(server.get_url())
908
def _file_contents(self, relpath, ranges, tail_amount=0):
909
code, data = self.transport._get(relpath, ranges)
910
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
911
for start, end in ranges:
913
yield data.read(end - start + 1)
915
def _file_tail(self, relpath, tail_amount):
916
code, data = self.transport._get(relpath, [], tail_amount)
917
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
918
data.seek(-tail_amount + 1, 2)
919
return data.read(tail_amount)
921
def test_range_header(self):
923
map(self.assertEqual,['0', '234'],
924
list(self._file_contents('a', [(0,0), (2,4)])),)
926
self.assertEqual('789', self._file_tail('a', 3))
927
# Syntactically invalid range
928
self.assertRaises(errors.InvalidRange,
929
self.transport._get, 'a', [(4, 3)])
930
# Semantically invalid range
931
self.assertRaises(errors.InvalidRange,
932
self.transport._get, 'a', [(42, 128)])
935
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
936
"""Test the Range header in GET methods for urllib implementation"""
938
_transport = HttpTransport_urllib
941
class TestRanges_pycurl(TestWithTransport_pycurl,
943
TestCaseWithWebserver):
944
"""Test the Range header in GET methods for pycurl implementation"""
947
class TestHTTPRedirections(object):
948
"""Test redirection between http servers.
950
This MUST be used by daughter classes that also inherit from
951
TestCaseWithRedirectedWebserver.
953
We can't inherit directly from TestCaseWithTwoWebservers or the
954
test framework will try to create an instance which cannot
955
run, its implementation being incomplete.
958
def create_transport_secondary_server(self):
959
"""Create the secondary server redirecting to the primary server"""
960
new = self.get_readonly_server()
962
redirecting = HTTPServerRedirecting()
963
redirecting.redirect_to(new.host, new.port)
967
super(TestHTTPRedirections, self).setUp()
968
self.build_tree_contents([('a', '0123456789'),
970
'# Bazaar revision bundle v0.9\n#\n')
973
self.old_transport = self._transport(self.old_server.get_url())
975
def test_redirected(self):
976
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
977
t = self._transport(self.new_server.get_url())
978
self.assertEqual('0123456789', t.get('a').read())
980
def test_read_redirected_bundle_from_url(self):
981
from bzrlib.bundle import read_bundle_from_url
982
url = self.old_transport.abspath('bundle')
983
bundle = read_bundle_from_url(url)
984
# If read_bundle_from_url was successful we get an empty bundle
985
self.assertEqual([], bundle.revisions)
988
class TestHTTPRedirections_urllib(TestHTTPRedirections,
989
TestCaseWithRedirectedWebserver):
990
"""Tests redirections for urllib implementation"""
992
_transport = HttpTransport_urllib
996
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
997
TestHTTPRedirections,
998
TestCaseWithRedirectedWebserver):
999
"""Tests redirections for pycurl implementation"""
1002
class RedirectedRequest(Request):
1003
"""Request following redirections"""
1005
init_orig = Request.__init__
1007
def __init__(self, method, url, *args, **kwargs):
1008
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1009
self.follow_redirections = True
1012
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1013
"""Test redirections provided by urllib.
1015
http implementations do not redirect silently anymore (they
1016
do not redirect at all in fact). The mechanism is still in
1017
place at the _urllib2_wrappers.Request level and these tests
1020
For the pycurl implementation
1021
the redirection have been deleted as we may deprecate pycurl
1022
and I have no place to keep a working implementation.
1026
_transport = HttpTransport_urllib
1029
super(TestHTTPSilentRedirections_urllib, self).setUp()
1030
self.setup_redirected_request()
1031
self.addCleanup(self.cleanup_redirected_request)
1032
self.build_tree_contents([('a','a'),
1034
('1/a', 'redirected once'),
1036
('2/a', 'redirected twice'),
1038
('3/a', 'redirected thrice'),
1040
('4/a', 'redirected 4 times'),
1042
('5/a', 'redirected 5 times'),
1045
self.old_transport = self._transport(self.old_server.get_url())
1047
def setup_redirected_request(self):
1048
self.original_class = _urllib2_wrappers.Request
1049
_urllib2_wrappers.Request = RedirectedRequest
1051
def cleanup_redirected_request(self):
1052
_urllib2_wrappers.Request = self.original_class
1054
def create_transport_secondary_server(self):
1055
"""Create the secondary server, redirections are defined in the tests"""
1056
return HTTPServerRedirecting()
1058
def test_one_redirection(self):
1059
t = self.old_transport
1061
req = RedirectedRequest('GET', t.abspath('a'))
1062
req.follow_redirections = True
1063
new_prefix = 'http://%s:%s' % (self.new_server.host,
1064
self.new_server.port)
1065
self.old_server.redirections = \
1066
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1067
self.assertEquals('redirected once',t._perform(req).read())
1069
def test_five_redirections(self):
1070
t = self.old_transport
1072
req = RedirectedRequest('GET', t.abspath('a'))
1073
req.follow_redirections = True
1074
old_prefix = 'http://%s:%s' % (self.old_server.host,
1075
self.old_server.port)
1076
new_prefix = 'http://%s:%s' % (self.new_server.host,
1077
self.new_server.port)
1078
self.old_server.redirections = \
1079
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1080
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1081
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1082
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1083
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1085
self.assertEquals('redirected 5 times',t._perform(req).read())
1088
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1089
"""Test transport.do_catching_redirections.
1091
We arbitrarily choose to use urllib transports
1094
_transport = HttpTransport_urllib
1097
super(TestDoCatchRedirections, self).setUp()
1098
self.build_tree_contents([('a', '0123456789'),],)
1100
self.old_transport = self._transport(self.old_server.get_url())
1102
def get_a(self, transport):
1103
return transport.get('a')
1105
def test_no_redirection(self):
1106
t = self._transport(self.new_server.get_url())
1108
# We use None for redirected so that we fail if redirected
1109
self.assertEquals('0123456789',
1110
do_catching_redirections(self.get_a, t, None).read())
1112
def test_one_redirection(self):
1113
self.redirections = 0
1115
def redirected(transport, exception, redirection_notice):
1116
self.redirections += 1
1117
dir, file = urlutils.split(exception.target)
1118
return self._transport(dir)
1120
self.assertEquals('0123456789',
1121
do_catching_redirections(self.get_a,
1125
self.assertEquals(1, self.redirections)
1127
def test_redirection_loop(self):
1129
def redirected(transport, exception, redirection_notice):
1130
# By using the redirected url as a base dir for the
1131
# *old* transport, we create a loop: a => a/a =>
1133
return self.old_transport.clone(exception.target)
1135
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1136
self.get_a, self.old_transport, redirected)
1139
class TestAuth(object):
1140
"""Test some authentication scheme specified by daughter class.
1142
This MUST be used by daughter classes that also inherit from
1143
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1147
"""Set up the test environment
1149
Daughter classes should set up their own environment
1150
(including self.server) and explicitely call this
1151
method. This is needed because we want to reuse the same
1152
tests for proxy and no-proxy accesses which have
1153
different ways of setting self.server.
1155
self.build_tree_contents([('a', 'contents of a\n'),
1156
('b', 'contents of b\n'),])
1157
self.old_factory = ui.ui_factory
1158
self.old_stdout = sys.stdout
1159
sys.stdout = StringIOWrapper()
1160
self.addCleanup(self.restoreUIFactory)
1162
def restoreUIFactory(self):
1163
ui.ui_factory = self.old_factory
1164
sys.stdout = self.old_stdout
1166
def get_user_url(self, user=None, password=None):
1167
"""Build an url embedding user and password"""
1168
url = '%s://' % self.server._url_protocol
1169
if user is not None:
1171
if password is not None:
1172
url += ':' + password
1174
url += '%s:%s/' % (self.server.host, self.server.port)
1177
def test_no_user(self):
1178
self.server.add_user('joe', 'foo')
1179
t = self.get_user_transport()
1180
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1181
# Only one 'Authentication Required' error should occur
1182
self.assertEqual(1, self.server.auth_required_errors)
1184
def test_empty_pass(self):
1185
self.server.add_user('joe', '')
1186
t = self.get_user_transport('joe', '')
1187
self.assertEqual('contents of a\n', t.get('a').read())
1188
# Only one 'Authentication Required' error should occur
1189
self.assertEqual(1, self.server.auth_required_errors)
1191
def test_user_pass(self):
1192
self.server.add_user('joe', 'foo')
1193
t = self.get_user_transport('joe', 'foo')
1194
self.assertEqual('contents of a\n', t.get('a').read())
1195
# Only one 'Authentication Required' error should occur
1196
self.assertEqual(1, self.server.auth_required_errors)
1198
def test_unknown_user(self):
1199
self.server.add_user('joe', 'foo')
1200
t = self.get_user_transport('bill', 'foo')
1201
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1202
# Two 'Authentication Required' errors should occur (the
1203
# initial 'who are you' and 'I don't know you, who are
1205
self.assertEqual(2, self.server.auth_required_errors)
1207
def test_wrong_pass(self):
1208
self.server.add_user('joe', 'foo')
1209
t = self.get_user_transport('joe', 'bar')
1210
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1211
# Two 'Authentication Required' errors should occur (the
1212
# initial 'who are you' and 'this is not you, who are you')
1213
self.assertEqual(2, self.server.auth_required_errors)
1215
def test_prompt_for_password(self):
1216
self.server.add_user('joe', 'foo')
1217
t = self.get_user_transport('joe', None)
1218
ui.ui_factory = TestUIFactory(stdin='foo\n')
1219
self.assertEqual('contents of a\n',t.get('a').read())
1220
# stdin should be empty
1221
self.assertEqual('', ui.ui_factory.stdin.readline())
1222
# And we shouldn't prompt again for a different request
1223
# against the same transport.
1224
self.assertEqual('contents of b\n',t.get('b').read())
1226
# And neither against a clone
1227
self.assertEqual('contents of b\n',t2.get('b').read())
1228
# Only one 'Authentication Required' error should occur
1229
self.assertEqual(1, self.server.auth_required_errors)
1232
class TestHTTPAuth(TestAuth):
1233
"""Test HTTP authentication schemes.
1235
Daughter classes MUST inherit from TestCaseWithWebserver too.
1238
_auth_header = 'Authorization'
1241
TestCaseWithWebserver.setUp(self)
1242
self.server = self.get_readonly_server()
1243
TestAuth.setUp(self)
1245
def get_user_transport(self, user=None, password=None):
1246
return self._transport(self.get_user_url(user, password))
1249
class TestProxyAuth(TestAuth):
1250
"""Test proxy authentication schemes.
1252
Daughter classes MUST also inherit from TestCaseWithWebserver.
1254
_auth_header = 'Proxy-authorization'
1257
TestCaseWithWebserver.setUp(self)
1258
self.server = self.get_readonly_server()
1260
self.addCleanup(self._restore_env)
1261
TestAuth.setUp(self)
1262
# Override the contents to avoid false positives
1263
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1264
('b', 'not proxied contents of b\n'),
1265
('a-proxied', 'contents of a\n'),
1266
('b-proxied', 'contents of b\n'),
1269
def get_user_transport(self, user=None, password=None):
1270
self._install_env({'all_proxy': self.get_user_url(user, password)})
1271
return self._transport(self.server.get_url())
1273
def _install_env(self, env):
1274
for name, value in env.iteritems():
1275
self._old_env[name] = osutils.set_or_unset_env(name, value)
1277
def _restore_env(self):
1278
for name, value in self._old_env.iteritems():
1279
osutils.set_or_unset_env(name, value)
1282
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1283
"""Test http basic authentication scheme"""
1285
_transport = HttpTransport_urllib
1287
def create_transport_readonly_server(self):
1288
return HTTPBasicAuthServer()
1291
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1292
"""Test proxy basic authentication scheme"""
1294
_transport = HttpTransport_urllib
1296
def create_transport_readonly_server(self):
1297
return ProxyBasicAuthServer()
1300
class TestDigestAuth(object):
1301
"""Digest Authentication specific tests"""
1303
def test_changing_nonce(self):
1304
self.server.add_user('joe', 'foo')
1305
t = self.get_user_transport('joe', 'foo')
1306
self.assertEqual('contents of a\n', t.get('a').read())
1307
self.assertEqual('contents of b\n', t.get('b').read())
1308
# Only one 'Authentication Required' error should have
1310
self.assertEqual(1, self.server.auth_required_errors)
1311
# The server invalidates the current nonce
1312
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1313
self.assertEqual('contents of a\n', t.get('a').read())
1314
# Two 'Authentication Required' errors should occur (the
1315
# initial 'who are you' and a second 'who are you' with the new nonce)
1316
self.assertEqual(2, self.server.auth_required_errors)
1319
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1320
"""Test http digest authentication scheme"""
1322
_transport = HttpTransport_urllib
1324
def create_transport_readonly_server(self):
1325
return HTTPDigestAuthServer()
1328
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1329
TestCaseWithWebserver):
1330
"""Test proxy digest authentication scheme"""
1332
_transport = HttpTransport_urllib
1334
def create_transport_readonly_server(self):
1335
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
