446
220
self.check_header('0-9,300-5000,-50',
447
221
ranges=[(0,9), (300,5000)],
451
class TestWallServer(object):
452
"""Tests exceptions during the connection phase"""
454
def create_transport_readonly_server(self):
455
return HttpServer(WallRequestHandler)
457
def test_http_has(self):
458
server = self.get_readonly_server()
459
t = self._transport(server.get_url())
460
# Unfortunately httplib (see HTTPResponse._read_status
461
# for details) make no distinction between a closed
462
# socket and badly formatted status line, so we can't
463
# just test for ConnectionError, we have to test
464
# InvalidHttpResponse too.
465
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
468
def test_http_get(self):
469
server = self.get_readonly_server()
470
t = self._transport(server.get_url())
471
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
475
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
476
"""Tests "wall" server for urllib implementation"""
478
_transport = HttpTransport_urllib
481
class TestWallServer_pycurl(TestWithTransport_pycurl,
483
TestCaseWithWebserver):
484
"""Tests "wall" server for pycurl implementation"""
487
class TestBadStatusServer(object):
488
"""Tests bad status from server."""
490
def create_transport_readonly_server(self):
491
return HttpServer(BadStatusRequestHandler)
493
def test_http_has(self):
494
server = self.get_readonly_server()
495
t = self._transport(server.get_url())
496
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
498
def test_http_get(self):
499
server = self.get_readonly_server()
500
t = self._transport(server.get_url())
501
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
504
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
505
"""Tests bad status server for urllib implementation"""
507
_transport = HttpTransport_urllib
510
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
512
TestCaseWithWebserver):
513
"""Tests bad status server for pycurl implementation"""
516
class TestInvalidStatusServer(TestBadStatusServer):
517
"""Tests invalid status from server.
519
Both implementations raises the same error as for a bad status.
522
def create_transport_readonly_server(self):
523
return HttpServer(InvalidStatusRequestHandler)
526
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
527
TestCaseWithWebserver):
528
"""Tests invalid status server for urllib implementation"""
530
_transport = HttpTransport_urllib
533
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
534
TestInvalidStatusServer,
535
TestCaseWithWebserver):
536
"""Tests invalid status server for pycurl implementation"""
539
class TestBadProtocolServer(object):
540
"""Tests bad protocol from server."""
542
def create_transport_readonly_server(self):
543
return HttpServer(BadProtocolRequestHandler)
545
def test_http_has(self):
546
server = self.get_readonly_server()
547
t = self._transport(server.get_url())
548
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
550
def test_http_get(self):
551
server = self.get_readonly_server()
552
t = self._transport(server.get_url())
553
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
556
class TestBadProtocolServer_urllib(TestBadProtocolServer,
557
TestCaseWithWebserver):
558
"""Tests bad protocol server for urllib implementation"""
560
_transport = HttpTransport_urllib
562
# curl don't check the protocol version
563
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
564
# TestBadProtocolServer,
565
# TestCaseWithWebserver):
566
# """Tests bad protocol server for pycurl implementation"""
569
class TestForbiddenServer(object):
570
"""Tests forbidden server"""
572
def create_transport_readonly_server(self):
573
return HttpServer(ForbiddenRequestHandler)
575
def test_http_has(self):
576
server = self.get_readonly_server()
577
t = self._transport(server.get_url())
578
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
580
def test_http_get(self):
581
server = self.get_readonly_server()
582
t = self._transport(server.get_url())
583
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
586
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
587
"""Tests forbidden server for urllib implementation"""
589
_transport = HttpTransport_urllib
592
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
594
TestCaseWithWebserver):
595
"""Tests forbidden server for pycurl implementation"""
598
class TestRecordingServer(TestCase):
600
def test_create(self):
601
server = RecordingServer(expect_body_tail=None)
602
self.assertEqual('', server.received_bytes)
603
self.assertEqual(None, server.host)
604
self.assertEqual(None, server.port)
606
def test_setUp_and_tearDown(self):
607
server = RecordingServer(expect_body_tail=None)
610
self.assertNotEqual(None, server.host)
611
self.assertNotEqual(None, server.port)
614
self.assertEqual(None, server.host)
615
self.assertEqual(None, server.port)
617
def test_send_receive_bytes(self):
618
server = RecordingServer(expect_body_tail='c')
620
self.addCleanup(server.tearDown)
621
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
622
sock.connect((server.host, server.port))
624
self.assertEqual('HTTP/1.1 200 OK\r\n',
625
osutils.recv_all(sock, 4096))
626
self.assertEqual('abc', server.received_bytes)
629
class TestRangeRequestServer(object):
630
"""Tests readv requests against server.
632
This MUST be used by daughter classes that also inherit from
633
TestCaseWithWebserver.
635
We can't inherit directly from TestCaseWithWebserver or the
636
test framework will try to create an instance which cannot
637
run, its implementation being incomplete.
641
TestCaseWithWebserver.setUp(self)
642
self.build_tree_contents([('a', '0123456789')],)
644
def test_readv(self):
645
server = self.get_readonly_server()
646
t = self._transport(server.get_url())
647
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
648
self.assertEqual(l[0], (0, '0'))
649
self.assertEqual(l[1], (1, '1'))
650
self.assertEqual(l[2], (3, '34'))
651
self.assertEqual(l[3], (9, '9'))
653
def test_readv_out_of_order(self):
654
server = self.get_readonly_server()
655
t = self._transport(server.get_url())
656
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
657
self.assertEqual(l[0], (1, '1'))
658
self.assertEqual(l[1], (9, '9'))
659
self.assertEqual(l[2], (0, '0'))
660
self.assertEqual(l[3], (3, '34'))
662
def test_readv_invalid_ranges(self):
663
server = self.get_readonly_server()
664
t = self._transport(server.get_url())
666
# This is intentionally reading off the end of the file
667
# since we are sure that it cannot get there
668
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
669
t.readv, 'a', [(1,1), (8,10)])
671
# This is trying to seek past the end of the file, it should
672
# also raise a special error
673
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
674
t.readv, 'a', [(12,2)])
677
class TestSingleRangeRequestServer(TestRangeRequestServer):
678
"""Test readv against a server which accept only single range requests"""
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleRangeRequestHandler)
684
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
688
_transport = HttpTransport_urllib
691
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleRangeRequestServer,
693
TestCaseWithWebserver):
694
"""Tests single range requests accepting server for pycurl implementation"""
697
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
698
"""Test readv against a server which only accept single range requests"""
700
def create_transport_readonly_server(self):
701
return HttpServer(SingleOnlyRangeRequestHandler)
704
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests single range requests accepting server for urllib implementation"""
708
_transport = HttpTransport_urllib
711
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestSingleOnlyRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests single range requests accepting server for pycurl implementation"""
717
class TestNoRangeRequestServer(TestRangeRequestServer):
718
"""Test readv against a server which do not accept range requests"""
720
def create_transport_readonly_server(self):
721
return HttpServer(NoRangeRequestHandler)
724
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
725
TestCaseWithWebserver):
726
"""Tests range requests refusing server for urllib implementation"""
728
_transport = HttpTransport_urllib
731
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
732
TestNoRangeRequestServer,
733
TestCaseWithWebserver):
734
"""Tests range requests refusing server for pycurl implementation"""
737
class TestHttpProxyWhiteBox(TestCase):
738
"""Whitebox test proxy http authorization.
740
Only the urllib implementation is tested here.
750
def _install_env(self, env):
751
for name, value in env.iteritems():
752
self._old_env[name] = osutils.set_or_unset_env(name, value)
754
def _restore_env(self):
755
for name, value in self._old_env.iteritems():
756
osutils.set_or_unset_env(name, value)
758
def _proxied_request(self):
759
handler = ProxyHandler(PasswordManager())
760
request = Request('GET','http://baz/buzzle')
761
handler.set_proxy(request, 'http')
764
def test_empty_user(self):
765
self._install_env({'http_proxy': 'http://bar.com'})
766
request = self._proxied_request()
767
self.assertFalse(request.headers.has_key('Proxy-authorization'))
769
def test_invalid_proxy(self):
770
"""A proxy env variable without scheme"""
771
self._install_env({'http_proxy': 'host:1234'})
772
self.assertRaises(errors.InvalidURL, self._proxied_request)
775
class TestProxyHttpServer(object):
776
"""Tests proxy server.
778
This MUST be used by daughter classes that also inherit from
779
TestCaseWithTwoWebservers.
781
We can't inherit directly from TestCaseWithTwoWebservers or
782
the test framework will try to create an instance which
783
cannot run, its implementation being incomplete.
785
Be aware that we do not setup a real proxy here. Instead, we
786
check that the *connection* goes through the proxy by serving
787
different content (the faked proxy server append '-proxied'
791
# FIXME: We don't have an https server available, so we don't
792
# test https connections.
794
# FIXME: Once the test suite is better fitted to test
795
# authorization schemes, test proxy authorizations too (see
799
TestCaseWithTwoWebservers.setUp(self)
800
self.build_tree_contents([('foo', 'contents of foo\n'),
801
('foo-proxied', 'proxied contents of foo\n')])
802
# Let's setup some attributes for tests
803
self.server = self.get_readonly_server()
804
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
805
self.no_proxy_host = self.proxy_address
806
# The secondary server is the proxy
807
self.proxy = self.get_secondary_server()
808
self.proxy_url = self.proxy.get_url()
811
def create_transport_secondary_server(self):
812
"""Creates an http server that will serve files with
813
'-proxied' appended to their names.
817
def _install_env(self, env):
818
for name, value in env.iteritems():
819
self._old_env[name] = osutils.set_or_unset_env(name, value)
821
def _restore_env(self):
822
for name, value in self._old_env.iteritems():
823
osutils.set_or_unset_env(name, value)
825
def proxied_in_env(self, env):
826
self._install_env(env)
827
url = self.server.get_url()
828
t = self._transport(url)
830
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
834
def not_proxied_in_env(self, env):
835
self._install_env(env)
836
url = self.server.get_url()
837
t = self._transport(url)
839
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
843
def test_http_proxy(self):
844
self.proxied_in_env({'http_proxy': self.proxy_url})
846
def test_HTTP_PROXY(self):
847
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
849
def test_all_proxy(self):
850
self.proxied_in_env({'all_proxy': self.proxy_url})
852
def test_ALL_PROXY(self):
853
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
855
def test_http_proxy_with_no_proxy(self):
856
self.not_proxied_in_env({'http_proxy': self.proxy_url,
857
'no_proxy': self.no_proxy_host})
859
def test_HTTP_PROXY_with_NO_PROXY(self):
860
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
861
'NO_PROXY': self.no_proxy_host})
863
def test_all_proxy_with_no_proxy(self):
864
self.not_proxied_in_env({'all_proxy': self.proxy_url,
865
'no_proxy': self.no_proxy_host})
867
def test_ALL_PROXY_with_NO_PROXY(self):
868
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
869
'NO_PROXY': self.no_proxy_host})
871
def test_http_proxy_without_scheme(self):
872
self.assertRaises(errors.InvalidURL,
874
{'http_proxy': self.proxy_address})
877
class TestProxyHttpServer_urllib(TestProxyHttpServer,
878
TestCaseWithTwoWebservers):
879
"""Tests proxy server for urllib implementation"""
881
_transport = HttpTransport_urllib
884
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
886
TestCaseWithTwoWebservers):
887
"""Tests proxy server for pycurl implementation"""
890
TestProxyHttpServer.setUp(self)
891
# Oh my ! pycurl does not check for the port as part of
892
# no_proxy :-( So we just test the host part
893
self.no_proxy_host = 'localhost'
895
def test_HTTP_PROXY(self):
896
# pycurl do not check HTTP_PROXY for security reasons
897
# (for use in a CGI context that we do not care
898
# about. Should we ?)
901
def test_HTTP_PROXY_with_NO_PROXY(self):
904
def test_http_proxy_without_scheme(self):
905
# pycurl *ignores* invalid proxy env variables. If that
906
# ever change in the future, this test will fail
907
# indicating that pycurl do not ignore anymore such
909
self.not_proxied_in_env({'http_proxy': self.proxy_address})
912
class TestRanges(object):
913
"""Test the Range header in GET methods..
915
This MUST be used by daughter classes that also inherit from
916
TestCaseWithWebserver.
918
We can't inherit directly from TestCaseWithWebserver or the
919
test framework will try to create an instance which cannot
920
run, its implementation being incomplete.
924
TestCaseWithWebserver.setUp(self)
925
self.build_tree_contents([('a', '0123456789')],)
926
server = self.get_readonly_server()
927
self.transport = self._transport(server.get_url())
929
def _file_contents(self, relpath, ranges, tail_amount=0):
930
code, data = self.transport._get(relpath, ranges)
931
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
932
for start, end in ranges:
934
yield data.read(end - start + 1)
936
def _file_tail(self, relpath, tail_amount):
937
code, data = self.transport._get(relpath, [], tail_amount)
938
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
939
data.seek(-tail_amount + 1, 2)
940
return data.read(tail_amount)
942
def test_range_header(self):
944
map(self.assertEqual,['0', '234'],
945
list(self._file_contents('a', [(0,0), (2,4)])),)
947
self.assertEqual('789', self._file_tail('a', 3))
948
# Syntactically invalid range
949
self.assertRaises(errors.InvalidRange,
950
self.transport._get, 'a', [(4, 3)])
951
# Semantically invalid range
952
self.assertRaises(errors.InvalidRange,
953
self.transport._get, 'a', [(42, 128)])
956
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
957
"""Test the Range header in GET methods for urllib implementation"""
959
_transport = HttpTransport_urllib
962
class TestRanges_pycurl(TestWithTransport_pycurl,
964
TestCaseWithWebserver):
965
"""Test the Range header in GET methods for pycurl implementation"""
968
class TestHTTPRedirections(object):
969
"""Test redirection between http servers.
971
This MUST be used by daughter classes that also inherit from
972
TestCaseWithRedirectedWebserver.
974
We can't inherit directly from TestCaseWithTwoWebservers or the
975
test framework will try to create an instance which cannot
976
run, its implementation being incomplete.
979
def create_transport_secondary_server(self):
980
"""Create the secondary server redirecting to the primary server"""
981
new = self.get_readonly_server()
983
redirecting = HTTPServerRedirecting()
984
redirecting.redirect_to(new.host, new.port)
988
super(TestHTTPRedirections, self).setUp()
989
self.build_tree_contents([('a', '0123456789'),
991
'# Bazaar revision bundle v0.9\n#\n')
994
self.old_transport = self._transport(self.old_server.get_url())
996
def test_redirected(self):
997
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
998
t = self._transport(self.new_server.get_url())
999
self.assertEqual('0123456789', t.get('a').read())
1001
def test_read_redirected_bundle_from_url(self):
1002
from bzrlib.bundle import read_bundle_from_url
1003
url = self.old_transport.abspath('bundle')
1004
bundle = read_bundle_from_url(url)
1005
# If read_bundle_from_url was successful we get an empty bundle
1006
self.assertEqual([], bundle.revisions)
1009
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1010
TestCaseWithRedirectedWebserver):
1011
"""Tests redirections for urllib implementation"""
1013
_transport = HttpTransport_urllib
1017
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1018
TestHTTPRedirections,
1019
TestCaseWithRedirectedWebserver):
1020
"""Tests redirections for pycurl implementation"""
1023
class RedirectedRequest(Request):
1024
"""Request following redirections"""
1026
init_orig = Request.__init__
1028
def __init__(self, method, url, *args, **kwargs):
1029
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1030
self.follow_redirections = True
1033
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1034
"""Test redirections provided by urllib.
1036
http implementations do not redirect silently anymore (they
1037
do not redirect at all in fact). The mechanism is still in
1038
place at the _urllib2_wrappers.Request level and these tests
1041
For the pycurl implementation
1042
the redirection have been deleted as we may deprecate pycurl
1043
and I have no place to keep a working implementation.
1047
_transport = HttpTransport_urllib
1050
super(TestHTTPSilentRedirections_urllib, self).setUp()
1051
self.setup_redirected_request()
1052
self.addCleanup(self.cleanup_redirected_request)
1053
self.build_tree_contents([('a','a'),
1055
('1/a', 'redirected once'),
1057
('2/a', 'redirected twice'),
1059
('3/a', 'redirected thrice'),
1061
('4/a', 'redirected 4 times'),
1063
('5/a', 'redirected 5 times'),
1066
self.old_transport = self._transport(self.old_server.get_url())
1068
def setup_redirected_request(self):
1069
self.original_class = _urllib2_wrappers.Request
1070
_urllib2_wrappers.Request = RedirectedRequest
1072
def cleanup_redirected_request(self):
1073
_urllib2_wrappers.Request = self.original_class
1075
def create_transport_secondary_server(self):
1076
"""Create the secondary server, redirections are defined in the tests"""
1077
return HTTPServerRedirecting()
1079
def test_one_redirection(self):
1080
t = self.old_transport
1082
req = RedirectedRequest('GET', t.abspath('a'))
1083
req.follow_redirections = True
1084
new_prefix = 'http://%s:%s' % (self.new_server.host,
1085
self.new_server.port)
1086
self.old_server.redirections = \
1087
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1088
self.assertEquals('redirected once',t._perform(req).read())
1090
def test_five_redirections(self):
1091
t = self.old_transport
1093
req = RedirectedRequest('GET', t.abspath('a'))
1094
req.follow_redirections = True
1095
old_prefix = 'http://%s:%s' % (self.old_server.host,
1096
self.old_server.port)
1097
new_prefix = 'http://%s:%s' % (self.new_server.host,
1098
self.new_server.port)
1099
self.old_server.redirections = \
1100
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1101
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1102
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1103
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1104
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1106
self.assertEquals('redirected 5 times',t._perform(req).read())
1109
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1110
"""Test transport.do_catching_redirections.
1112
We arbitrarily choose to use urllib transports
1115
_transport = HttpTransport_urllib
1118
super(TestDoCatchRedirections, self).setUp()
1119
self.build_tree_contents([('a', '0123456789'),],)
1121
self.old_transport = self._transport(self.old_server.get_url())
1123
def get_a(self, transport):
1124
return transport.get('a')
1126
def test_no_redirection(self):
1127
t = self._transport(self.new_server.get_url())
1129
# We use None for redirected so that we fail if redirected
1130
self.assertEquals('0123456789',
1131
do_catching_redirections(self.get_a, t, None).read())
1133
def test_one_redirection(self):
1134
self.redirections = 0
1136
def redirected(transport, exception, redirection_notice):
1137
self.redirections += 1
1138
dir, file = urlutils.split(exception.target)
1139
return self._transport(dir)
1141
self.assertEquals('0123456789',
1142
do_catching_redirections(self.get_a,
1146
self.assertEquals(1, self.redirections)
1148
def test_redirection_loop(self):
1150
def redirected(transport, exception, redirection_notice):
1151
# By using the redirected url as a base dir for the
1152
# *old* transport, we create a loop: a => a/a =>
1154
return self.old_transport.clone(exception.target)
1156
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1157
self.get_a, self.old_transport, redirected)
1160
class TestAuth(object):
1161
"""Test some authentication scheme specified by daughter class.
1163
This MUST be used by daughter classes that also inherit from
1164
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1168
"""Set up the test environment
1170
Daughter classes should set up their own environment
1171
(including self.server) and explicitely call this
1172
method. This is needed because we want to reuse the same
1173
tests for proxy and no-proxy accesses which have
1174
different ways of setting self.server.
1176
self.build_tree_contents([('a', 'contents of a\n'),
1177
('b', 'contents of b\n'),])
1178
self.old_factory = ui.ui_factory
1179
self.old_stdout = sys.stdout
1180
sys.stdout = StringIOWrapper()
1181
self.addCleanup(self.restoreUIFactory)
1183
def restoreUIFactory(self):
1184
ui.ui_factory = self.old_factory
1185
sys.stdout = self.old_stdout
1187
def get_user_url(self, user=None, password=None):
1188
"""Build an url embedding user and password"""
1189
url = '%s://' % self.server._url_protocol
1190
if user is not None:
1192
if password is not None:
1193
url += ':' + password
1195
url += '%s:%s/' % (self.server.host, self.server.port)
1198
def test_no_user(self):
1199
self.server.add_user('joe', 'foo')
1200
t = self.get_user_transport()
1201
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1202
# Only one 'Authentication Required' error should occur
1203
self.assertEqual(1, self.server.auth_required_errors)
1205
def test_empty_pass(self):
1206
self.server.add_user('joe', '')
1207
t = self.get_user_transport('joe', '')
1208
self.assertEqual('contents of a\n', t.get('a').read())
1209
# Only one 'Authentication Required' error should occur
1210
self.assertEqual(1, self.server.auth_required_errors)
1212
def test_user_pass(self):
1213
self.server.add_user('joe', 'foo')
1214
t = self.get_user_transport('joe', 'foo')
1215
self.assertEqual('contents of a\n', t.get('a').read())
1216
# Only one 'Authentication Required' error should occur
1217
self.assertEqual(1, self.server.auth_required_errors)
1219
def test_unknown_user(self):
1220
self.server.add_user('joe', 'foo')
1221
t = self.get_user_transport('bill', 'foo')
1222
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1223
# Two 'Authentication Required' errors should occur (the
1224
# initial 'who are you' and 'I don't know you, who are
1226
self.assertEqual(2, self.server.auth_required_errors)
1228
def test_wrong_pass(self):
1229
self.server.add_user('joe', 'foo')
1230
t = self.get_user_transport('joe', 'bar')
1231
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1232
# Two 'Authentication Required' errors should occur (the
1233
# initial 'who are you' and 'this is not you, who are you')
1234
self.assertEqual(2, self.server.auth_required_errors)
1236
def test_prompt_for_password(self):
1237
self.server.add_user('joe', 'foo')
1238
t = self.get_user_transport('joe', None)
1239
ui.ui_factory = TestUIFactory(stdin='foo\n')
1240
self.assertEqual('contents of a\n',t.get('a').read())
1241
# stdin should be empty
1242
self.assertEqual('', ui.ui_factory.stdin.readline())
1243
# And we shouldn't prompt again for a different request
1244
# against the same transport.
1245
self.assertEqual('contents of b\n',t.get('b').read())
1247
# And neither against a clone
1248
self.assertEqual('contents of b\n',t2.get('b').read())
1249
# Only one 'Authentication Required' error should occur
1250
self.assertEqual(1, self.server.auth_required_errors)
1253
class TestHTTPAuth(TestAuth):
1254
"""Test HTTP authentication schemes.
1256
Daughter classes MUST inherit from TestCaseWithWebserver too.
1259
_auth_header = 'Authorization'
1262
TestCaseWithWebserver.setUp(self)
1263
self.server = self.get_readonly_server()
1264
TestAuth.setUp(self)
1266
def get_user_transport(self, user=None, password=None):
1267
return self._transport(self.get_user_url(user, password))
1270
class TestProxyAuth(TestAuth):
1271
"""Test proxy authentication schemes.
1273
Daughter classes MUST also inherit from TestCaseWithWebserver.
1275
_auth_header = 'Proxy-authorization'
1278
TestCaseWithWebserver.setUp(self)
1279
self.server = self.get_readonly_server()
1281
self.addCleanup(self._restore_env)
1282
TestAuth.setUp(self)
1283
# Override the contents to avoid false positives
1284
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1285
('b', 'not proxied contents of b\n'),
1286
('a-proxied', 'contents of a\n'),
1287
('b-proxied', 'contents of b\n'),
1290
def get_user_transport(self, user=None, password=None):
1291
self._install_env({'all_proxy': self.get_user_url(user, password)})
1292
return self._transport(self.server.get_url())
1294
def _install_env(self, env):
1295
for name, value in env.iteritems():
1296
self._old_env[name] = osutils.set_or_unset_env(name, value)
1298
def _restore_env(self):
1299
for name, value in self._old_env.iteritems():
1300
osutils.set_or_unset_env(name, value)
1303
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1304
"""Test http basic authentication scheme"""
1306
_transport = HttpTransport_urllib
1308
def create_transport_readonly_server(self):
1309
return HTTPBasicAuthServer()
1312
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1313
"""Test proxy basic authentication scheme"""
1315
_transport = HttpTransport_urllib
1317
def create_transport_readonly_server(self):
1318
return ProxyBasicAuthServer()
1321
class TestDigestAuth(object):
1322
"""Digest Authentication specific tests"""
1324
def test_changing_nonce(self):
1325
self.server.add_user('joe', 'foo')
1326
t = self.get_user_transport('joe', 'foo')
1327
self.assertEqual('contents of a\n', t.get('a').read())
1328
self.assertEqual('contents of b\n', t.get('b').read())
1329
# Only one 'Authentication Required' error should have
1331
self.assertEqual(1, self.server.auth_required_errors)
1332
# The server invalidates the current nonce
1333
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1334
self.assertEqual('contents of a\n', t.get('a').read())
1335
# Two 'Authentication Required' errors should occur (the
1336
# initial 'who are you' and a second 'who are you' with the new nonce)
1337
self.assertEqual(2, self.server.auth_required_errors)
1340
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1341
"""Test http digest authentication scheme"""
1343
_transport = HttpTransport_urllib
1345
def create_transport_readonly_server(self):
1346
return HTTPDigestAuthServer()
1349
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1350
TestCaseWithWebserver):
1351
"""Test proxy digest authentication scheme"""
1353
_transport = HttpTransport_urllib
1355
def create_transport_readonly_server(self):
1356
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
