314
120
'contents of foo/bar\n')
315
121
self.assertEqual(len(server.logs), 1)
316
122
self.assertTrue(server.logs[0].find(
317
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
318
% bzrlib.__version__) > -1)
320
def test_get_smart_medium(self):
321
# For HTTP, get_smart_medium should return the transport object.
322
server = self.get_readonly_server()
323
http_transport = self._transport(server.get_url())
324
medium = http_transport.get_smart_medium()
325
self.assertIs(medium, http_transport)
327
def test_has_on_bogus_host(self):
328
# Get a free address and don't 'accept' on it, so that we
329
# can be sure there is no http handler there, but set a
330
# reasonable timeout to not slow down tests too much.
331
default_timeout = socket.getdefaulttimeout()
123
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s' % bzrlib.__version__) > -1)
126
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
127
_transport = HttpTransport_urllib
130
TestCaseWithWebserver.setUp(self)
135
class TestHttpConnections_pycurl(TestCaseWithWebserver, TestHttpMixins):
137
def _get_pycurl_maybe(self):
333
socket.setdefaulttimeout(2)
335
s.bind(('localhost', 0))
336
t = self._transport('http://%s:%s/' % s.getsockname())
337
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
339
socket.setdefaulttimeout(default_timeout)
342
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
343
"""Test http connections with urllib"""
345
_transport = HttpTransport_urllib
349
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
351
TestCaseWithWebserver):
352
"""Test http connections with pycurl"""
139
from bzrlib.transport.http._pycurl import PyCurlTransport
140
return PyCurlTransport
141
except DependencyNotPresent:
142
raise TestSkipped('pycurl not present')
144
_transport = property(_get_pycurl_maybe)
147
TestCaseWithWebserver.setUp(self)
355
152
class TestHttpTransportRegistration(TestCase):
356
153
"""Test registrations of various http implementations"""
358
155
def test_http_registered(self):
156
import bzrlib.transport.http._urllib
157
from bzrlib.transport import get_transport
359
158
# urlllib should always be present
360
159
t = get_transport('http+urllib://bzr.google.com/')
361
160
self.assertIsInstance(t, Transport)
362
self.assertIsInstance(t, HttpTransport_urllib)
365
class TestPost(object):
367
def _test_post_body_is_received(self, scheme):
368
server = RecordingServer(expect_body_tail='end-of-body')
370
self.addCleanup(server.tearDown)
371
url = '%s://%s:%s/' % (scheme, server.host, server.port)
373
http_transport = get_transport(url)
374
except errors.UnsupportedProtocol:
375
raise TestSkipped('%s not available' % scheme)
376
code, response = http_transport._post('abc def end-of-body')
378
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
379
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
380
# The transport should not be assuming that the server can accept
381
# chunked encoding the first time it connects, because HTTP/1.1, so we
382
# check for the literal string.
384
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
387
class TestPost_urllib(TestCase, TestPost):
388
"""TestPost for urllib implementation"""
390
_transport = HttpTransport_urllib
392
def test_post_body_is_received_urllib(self):
393
self._test_post_body_is_received('http+urllib')
396
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
397
"""TestPost for pycurl implementation"""
399
def test_post_body_is_received_pycurl(self):
400
self._test_post_body_is_received('http+pycurl')
403
class TestRangeHeader(TestCase):
404
"""Test range_header method"""
406
def check_header(self, value, ranges=[], tail=0):
407
offsets = [ (start, end - start + 1) for start, end in ranges]
408
coalesce = Transport._coalesce_offsets
409
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
410
range_header = HttpTransportBase._range_header
411
self.assertEqual(value, range_header(coalesced, tail))
413
def test_range_header_single(self):
414
self.check_header('0-9', ranges=[(0,9)])
415
self.check_header('100-109', ranges=[(100,109)])
417
def test_range_header_tail(self):
418
self.check_header('-10', tail=10)
419
self.check_header('-50', tail=50)
421
def test_range_header_multi(self):
422
self.check_header('0-9,100-200,300-5000',
423
ranges=[(0,9), (100, 200), (300,5000)])
425
def test_range_header_mixed(self):
426
self.check_header('0-9,300-5000,-50',
427
ranges=[(0,9), (300,5000)],
431
class TestWallServer(object):
432
"""Tests exceptions during the connection phase"""
434
def create_transport_readonly_server(self):
435
return HttpServer(WallRequestHandler)
437
def test_http_has(self):
438
server = self.get_readonly_server()
439
t = self._transport(server.get_url())
440
# Unfortunately httplib (see HTTPResponse._read_status
441
# for details) make no distinction between a closed
442
# socket and badly formatted status line, so we can't
443
# just test for ConnectionError, we have to test
444
# InvalidHttpResponse too.
445
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
448
def test_http_get(self):
449
server = self.get_readonly_server()
450
t = self._transport(server.get_url())
451
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
455
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
456
"""Tests "wall" server for urllib implementation"""
458
_transport = HttpTransport_urllib
461
class TestWallServer_pycurl(TestWithTransport_pycurl,
463
TestCaseWithWebserver):
464
"""Tests "wall" server for pycurl implementation"""
467
class TestBadStatusServer(object):
468
"""Tests bad status from server."""
470
def create_transport_readonly_server(self):
471
return HttpServer(BadStatusRequestHandler)
473
def test_http_has(self):
474
server = self.get_readonly_server()
475
t = self._transport(server.get_url())
476
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
478
def test_http_get(self):
479
server = self.get_readonly_server()
480
t = self._transport(server.get_url())
481
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
484
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
485
"""Tests bad status server for urllib implementation"""
487
_transport = HttpTransport_urllib
490
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
492
TestCaseWithWebserver):
493
"""Tests bad status server for pycurl implementation"""
496
class TestInvalidStatusServer(TestBadStatusServer):
497
"""Tests invalid status from server.
499
Both implementations raises the same error as for a bad status.
502
def create_transport_readonly_server(self):
503
return HttpServer(InvalidStatusRequestHandler)
506
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
507
TestCaseWithWebserver):
508
"""Tests invalid status server for urllib implementation"""
510
_transport = HttpTransport_urllib
513
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
514
TestInvalidStatusServer,
515
TestCaseWithWebserver):
516
"""Tests invalid status server for pycurl implementation"""
519
class TestBadProtocolServer(object):
520
"""Tests bad protocol from server."""
522
def create_transport_readonly_server(self):
523
return HttpServer(BadProtocolRequestHandler)
525
def test_http_has(self):
526
server = self.get_readonly_server()
527
t = self._transport(server.get_url())
528
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
530
def test_http_get(self):
531
server = self.get_readonly_server()
532
t = self._transport(server.get_url())
533
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
536
class TestBadProtocolServer_urllib(TestBadProtocolServer,
537
TestCaseWithWebserver):
538
"""Tests bad protocol server for urllib implementation"""
540
_transport = HttpTransport_urllib
542
# curl don't check the protocol version
543
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
544
# TestBadProtocolServer,
545
# TestCaseWithWebserver):
546
# """Tests bad protocol server for pycurl implementation"""
549
class TestForbiddenServer(object):
550
"""Tests forbidden server"""
552
def create_transport_readonly_server(self):
553
return HttpServer(ForbiddenRequestHandler)
555
def test_http_has(self):
556
server = self.get_readonly_server()
557
t = self._transport(server.get_url())
558
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
560
def test_http_get(self):
561
server = self.get_readonly_server()
562
t = self._transport(server.get_url())
563
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
566
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
567
"""Tests forbidden server for urllib implementation"""
569
_transport = HttpTransport_urllib
572
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
574
TestCaseWithWebserver):
575
"""Tests forbidden server for pycurl implementation"""
578
class TestRecordingServer(TestCase):
580
def test_create(self):
581
server = RecordingServer(expect_body_tail=None)
582
self.assertEqual('', server.received_bytes)
583
self.assertEqual(None, server.host)
584
self.assertEqual(None, server.port)
586
def test_setUp_and_tearDown(self):
587
server = RecordingServer(expect_body_tail=None)
590
self.assertNotEqual(None, server.host)
591
self.assertNotEqual(None, server.port)
594
self.assertEqual(None, server.host)
595
self.assertEqual(None, server.port)
597
def test_send_receive_bytes(self):
598
server = RecordingServer(expect_body_tail='c')
600
self.addCleanup(server.tearDown)
601
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
602
sock.connect((server.host, server.port))
604
self.assertEqual('HTTP/1.1 200 OK\r\n',
605
osutils.recv_all(sock, 4096))
606
self.assertEqual('abc', server.received_bytes)
609
class TestRangeRequestServer(object):
610
"""Tests readv requests against server.
612
This MUST be used by daughter classes that also inherit from
613
TestCaseWithWebserver.
615
We can't inherit directly from TestCaseWithWebserver or the
616
test framework will try to create an instance which cannot
617
run, its implementation being incomplete.
621
TestCaseWithWebserver.setUp(self)
622
self.build_tree_contents([('a', '0123456789')],)
624
def test_readv(self):
625
server = self.get_readonly_server()
626
t = self._transport(server.get_url())
627
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
628
self.assertEqual(l[0], (0, '0'))
629
self.assertEqual(l[1], (1, '1'))
630
self.assertEqual(l[2], (3, '34'))
631
self.assertEqual(l[3], (9, '9'))
633
def test_readv_out_of_order(self):
634
server = self.get_readonly_server()
635
t = self._transport(server.get_url())
636
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
637
self.assertEqual(l[0], (1, '1'))
638
self.assertEqual(l[1], (9, '9'))
639
self.assertEqual(l[2], (0, '0'))
640
self.assertEqual(l[3], (3, '34'))
642
def test_readv_invalid_ranges(self):
643
server = self.get_readonly_server()
644
t = self._transport(server.get_url())
646
# This is intentionally reading off the end of the file
647
# since we are sure that it cannot get there
648
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
649
t.readv, 'a', [(1,1), (8,10)])
651
# This is trying to seek past the end of the file, it should
652
# also raise a special error
653
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
654
t.readv, 'a', [(12,2)])
657
class TestSingleRangeRequestServer(TestRangeRequestServer):
658
"""Test readv against a server which accept only single range requests"""
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"""
677
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
678
"""Test readv against a server which only accept single range requests"""
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"""
697
class TestNoRangeRequestServer(TestRangeRequestServer):
698
"""Test readv against a server which do not accept range requests"""
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.
730
def create_transport_readonly_server(self):
731
# Requests with more range specifiers will error out
732
return LimitedRangeHTTPServer(range_limit=self.range_limit)
734
def get_transport(self):
735
return self._transport(self.get_readonly_server().get_url())
738
TestCaseWithWebserver.setUp(self)
739
# We need to manipulate ranges that correspond to real chunks in the
740
# response, so we build a content appropriately.
741
filler = ''.join(['abcdefghij' for _ in range(102)])
742
content = ''.join(['%04d' % v + filler for v in range(16)])
743
self.build_tree_contents([('a', content)],)
745
def test_few_ranges(self):
746
t = self.get_transport()
747
l = list(t.readv('a', ((0, 4), (1024, 4), )))
748
self.assertEqual(l[0], (0, '0000'))
749
self.assertEqual(l[1], (1024, '0001'))
750
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
752
def test_a_lot_of_ranges(self):
753
t = self.get_transport()
754
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
755
self.assertEqual(l[0], (0, '0000'))
756
self.assertEqual(l[1], (1024, '0001'))
757
self.assertEqual(l[2], (4096, '0004'))
758
self.assertEqual(l[3], (8192, '0008'))
759
# The server will refuse to serve the first request (too much ranges),
760
# a second request will succeeds.
761
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
764
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
765
TestCaseWithWebserver):
766
"""Tests limited range requests server for urllib implementation"""
768
_transport = HttpTransport_urllib
771
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
772
TestLimitedRangeRequestServer,
773
TestCaseWithWebserver):
774
"""Tests limited range requests server for pycurl implementation"""
778
class TestHttpProxyWhiteBox(TestCase):
779
"""Whitebox test proxy http authorization.
781
Only the urllib implementation is tested here.
791
def _install_env(self, env):
792
for name, value in env.iteritems():
793
self._old_env[name] = osutils.set_or_unset_env(name, value)
795
def _restore_env(self):
796
for name, value in self._old_env.iteritems():
797
osutils.set_or_unset_env(name, value)
799
def _proxied_request(self):
800
handler = ProxyHandler(PasswordManager())
801
request = Request('GET','http://baz/buzzle')
802
handler.set_proxy(request, 'http')
805
def test_empty_user(self):
806
self._install_env({'http_proxy': 'http://bar.com'})
807
request = self._proxied_request()
808
self.assertFalse(request.headers.has_key('Proxy-authorization'))
810
def test_invalid_proxy(self):
811
"""A proxy env variable without scheme"""
812
self._install_env({'http_proxy': 'host:1234'})
813
self.assertRaises(errors.InvalidURL, self._proxied_request)
816
class TestProxyHttpServer(object):
817
"""Tests proxy server.
819
This MUST be used by daughter classes that also inherit from
820
TestCaseWithTwoWebservers.
822
We can't inherit directly from TestCaseWithTwoWebservers or
823
the test framework will try to create an instance which
824
cannot run, its implementation being incomplete.
826
Be aware that we do not setup a real proxy here. Instead, we
827
check that the *connection* goes through the proxy by serving
828
different content (the faked proxy server append '-proxied'
832
# FIXME: We don't have an https server available, so we don't
833
# test https connections.
835
# FIXME: Once the test suite is better fitted to test
836
# authorization schemes, test proxy authorizations too (see
840
TestCaseWithTwoWebservers.setUp(self)
841
self.build_tree_contents([('foo', 'contents of foo\n'),
842
('foo-proxied', 'proxied contents of foo\n')])
843
# Let's setup some attributes for tests
844
self.server = self.get_readonly_server()
845
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
846
self.no_proxy_host = self.proxy_address
847
# The secondary server is the proxy
848
self.proxy = self.get_secondary_server()
849
self.proxy_url = self.proxy.get_url()
852
def create_transport_secondary_server(self):
853
"""Creates an http server that will serve files with
854
'-proxied' appended to their names.
858
def _install_env(self, env):
859
for name, value in env.iteritems():
860
self._old_env[name] = osutils.set_or_unset_env(name, value)
862
def _restore_env(self):
863
for name, value in self._old_env.iteritems():
864
osutils.set_or_unset_env(name, value)
866
def proxied_in_env(self, env):
867
self._install_env(env)
868
url = self.server.get_url()
869
t = self._transport(url)
871
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
875
def not_proxied_in_env(self, env):
876
self._install_env(env)
877
url = self.server.get_url()
878
t = self._transport(url)
880
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
884
def test_http_proxy(self):
885
self.proxied_in_env({'http_proxy': self.proxy_url})
887
def test_HTTP_PROXY(self):
888
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
890
def test_all_proxy(self):
891
self.proxied_in_env({'all_proxy': self.proxy_url})
893
def test_ALL_PROXY(self):
894
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
896
def test_http_proxy_with_no_proxy(self):
897
self.not_proxied_in_env({'http_proxy': self.proxy_url,
898
'no_proxy': self.no_proxy_host})
900
def test_HTTP_PROXY_with_NO_PROXY(self):
901
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
902
'NO_PROXY': self.no_proxy_host})
904
def test_all_proxy_with_no_proxy(self):
905
self.not_proxied_in_env({'all_proxy': self.proxy_url,
906
'no_proxy': self.no_proxy_host})
908
def test_ALL_PROXY_with_NO_PROXY(self):
909
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
910
'NO_PROXY': self.no_proxy_host})
912
def test_http_proxy_without_scheme(self):
913
self.assertRaises(errors.InvalidURL,
915
{'http_proxy': self.proxy_address})
918
class TestProxyHttpServer_urllib(TestProxyHttpServer,
919
TestCaseWithTwoWebservers):
920
"""Tests proxy server for urllib implementation"""
922
_transport = HttpTransport_urllib
925
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
927
TestCaseWithTwoWebservers):
928
"""Tests proxy server for pycurl implementation"""
931
TestProxyHttpServer.setUp(self)
932
# Oh my ! pycurl does not check for the port as part of
933
# no_proxy :-( So we just test the host part
934
self.no_proxy_host = 'localhost'
936
def test_HTTP_PROXY(self):
937
# pycurl do not check HTTP_PROXY for security reasons
938
# (for use in a CGI context that we do not care
939
# about. Should we ?)
942
def test_HTTP_PROXY_with_NO_PROXY(self):
945
def test_http_proxy_without_scheme(self):
946
# pycurl *ignores* invalid proxy env variables. If that
947
# ever change in the future, this test will fail
948
# indicating that pycurl do not ignore anymore such
950
self.not_proxied_in_env({'http_proxy': self.proxy_address})
953
class TestRanges(object):
954
"""Test the Range header in GET methods..
956
This MUST be used by daughter classes that also inherit from
957
TestCaseWithWebserver.
959
We can't inherit directly from TestCaseWithWebserver or the
960
test framework will try to create an instance which cannot
961
run, its implementation being incomplete.
965
TestCaseWithWebserver.setUp(self)
966
self.build_tree_contents([('a', '0123456789')],)
967
server = self.get_readonly_server()
968
self.transport = self._transport(server.get_url())
970
def _file_contents(self, relpath, ranges):
971
offsets = [ (start, end - start + 1) for start, end in ranges]
972
coalesce = self.transport._coalesce_offsets
973
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
974
code, data = self.transport._get(relpath, coalesced)
975
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
976
for start, end in ranges:
978
yield data.read(end - start + 1)
980
def _file_tail(self, relpath, tail_amount):
981
code, data = self.transport._get(relpath, [], tail_amount)
982
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
983
data.seek(-tail_amount + 1, 2)
984
return data.read(tail_amount)
986
def test_range_header(self):
988
map(self.assertEqual,['0', '234'],
989
list(self._file_contents('a', [(0,0), (2,4)])),)
991
self.assertEqual('789', self._file_tail('a', 3))
992
# Syntactically invalid range
993
self.assertListRaises(errors.InvalidRange,
994
self._file_contents, 'a', [(4, 3)])
995
# Semantically invalid range
996
self.assertListRaises(errors.InvalidRange,
997
self._file_contents, 'a', [(42, 128)])
1000
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1001
"""Test the Range header in GET methods for urllib implementation"""
1003
_transport = HttpTransport_urllib
1006
class TestRanges_pycurl(TestWithTransport_pycurl,
1008
TestCaseWithWebserver):
1009
"""Test the Range header in GET methods for pycurl implementation"""
1012
class TestHTTPRedirections(object):
1013
"""Test redirection between http servers.
1015
This MUST be used by daughter classes that also inherit from
1016
TestCaseWithRedirectedWebserver.
1018
We can't inherit directly from TestCaseWithTwoWebservers or the
1019
test framework will try to create an instance which cannot
1020
run, its implementation being incomplete.
1023
def create_transport_secondary_server(self):
1024
"""Create the secondary server redirecting to the primary server"""
1025
new = self.get_readonly_server()
1027
redirecting = HTTPServerRedirecting()
1028
redirecting.redirect_to(new.host, new.port)
1032
super(TestHTTPRedirections, self).setUp()
1033
self.build_tree_contents([('a', '0123456789'),
1035
'# Bazaar revision bundle v0.9\n#\n')
1038
self.old_transport = self._transport(self.old_server.get_url())
1040
def test_redirected(self):
1041
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1042
t = self._transport(self.new_server.get_url())
1043
self.assertEqual('0123456789', t.get('a').read())
1045
def test_read_redirected_bundle_from_url(self):
1046
from bzrlib.bundle import read_bundle_from_url
1047
url = self.old_transport.abspath('bundle')
1048
bundle = read_bundle_from_url(url)
1049
# If read_bundle_from_url was successful we get an empty bundle
1050
self.assertEqual([], bundle.revisions)
1053
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1054
TestCaseWithRedirectedWebserver):
1055
"""Tests redirections for urllib implementation"""
1057
_transport = HttpTransport_urllib
1061
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1062
TestHTTPRedirections,
1063
TestCaseWithRedirectedWebserver):
1064
"""Tests redirections for pycurl implementation"""
1067
class RedirectedRequest(Request):
1068
"""Request following redirections"""
1070
init_orig = Request.__init__
1072
def __init__(self, method, url, *args, **kwargs):
1073
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1074
self.follow_redirections = True
1077
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1078
"""Test redirections provided by urllib.
1080
http implementations do not redirect silently anymore (they
1081
do not redirect at all in fact). The mechanism is still in
1082
place at the _urllib2_wrappers.Request level and these tests
1085
For the pycurl implementation
1086
the redirection have been deleted as we may deprecate pycurl
1087
and I have no place to keep a working implementation.
1091
_transport = HttpTransport_urllib
1094
super(TestHTTPSilentRedirections_urllib, self).setUp()
1095
self.setup_redirected_request()
1096
self.addCleanup(self.cleanup_redirected_request)
1097
self.build_tree_contents([('a','a'),
1099
('1/a', 'redirected once'),
1101
('2/a', 'redirected twice'),
1103
('3/a', 'redirected thrice'),
1105
('4/a', 'redirected 4 times'),
1107
('5/a', 'redirected 5 times'),
1110
self.old_transport = self._transport(self.old_server.get_url())
1112
def setup_redirected_request(self):
1113
self.original_class = _urllib2_wrappers.Request
1114
_urllib2_wrappers.Request = RedirectedRequest
1116
def cleanup_redirected_request(self):
1117
_urllib2_wrappers.Request = self.original_class
1119
def create_transport_secondary_server(self):
1120
"""Create the secondary server, redirections are defined in the tests"""
1121
return HTTPServerRedirecting()
1123
def test_one_redirection(self):
1124
t = self.old_transport
1126
req = RedirectedRequest('GET', t.abspath('a'))
1127
req.follow_redirections = True
1128
new_prefix = 'http://%s:%s' % (self.new_server.host,
1129
self.new_server.port)
1130
self.old_server.redirections = \
1131
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1132
self.assertEquals('redirected once',t._perform(req).read())
1134
def test_five_redirections(self):
1135
t = self.old_transport
1137
req = RedirectedRequest('GET', t.abspath('a'))
1138
req.follow_redirections = True
1139
old_prefix = 'http://%s:%s' % (self.old_server.host,
1140
self.old_server.port)
1141
new_prefix = 'http://%s:%s' % (self.new_server.host,
1142
self.new_server.port)
1143
self.old_server.redirections = \
1144
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1145
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1146
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1147
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1148
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1150
self.assertEquals('redirected 5 times',t._perform(req).read())
1153
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1154
"""Test transport.do_catching_redirections.
1156
We arbitrarily choose to use urllib transports
1159
_transport = HttpTransport_urllib
1162
super(TestDoCatchRedirections, self).setUp()
1163
self.build_tree_contents([('a', '0123456789'),],)
1165
self.old_transport = self._transport(self.old_server.get_url())
1167
def get_a(self, transport):
1168
return transport.get('a')
1170
def test_no_redirection(self):
1171
t = self._transport(self.new_server.get_url())
1173
# We use None for redirected so that we fail if redirected
1174
self.assertEquals('0123456789',
1175
do_catching_redirections(self.get_a, t, None).read())
1177
def test_one_redirection(self):
1178
self.redirections = 0
1180
def redirected(transport, exception, redirection_notice):
1181
self.redirections += 1
1182
dir, file = urlutils.split(exception.target)
1183
return self._transport(dir)
1185
self.assertEquals('0123456789',
1186
do_catching_redirections(self.get_a,
1190
self.assertEquals(1, self.redirections)
1192
def test_redirection_loop(self):
1194
def redirected(transport, exception, redirection_notice):
1195
# By using the redirected url as a base dir for the
1196
# *old* transport, we create a loop: a => a/a =>
1198
return self.old_transport.clone(exception.target)
1200
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1201
self.get_a, self.old_transport, redirected)
1204
class TestAuth(object):
1205
"""Test some authentication scheme specified by daughter class.
1207
This MUST be used by daughter classes that also inherit from
1208
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1212
"""Set up the test environment
1214
Daughter classes should set up their own environment
1215
(including self.server) and explicitely call this
1216
method. This is needed because we want to reuse the same
1217
tests for proxy and no-proxy accesses which have
1218
different ways of setting self.server.
1220
self.build_tree_contents([('a', 'contents of a\n'),
1221
('b', 'contents of b\n'),])
1222
self.old_factory = ui.ui_factory
1223
# The following has the unfortunate side-effect of hiding any ouput
1224
# during the tests (including pdb prompts). Feel free to comment them
1225
# for debugging purposes but leave them in place, there are needed to
1226
# run the tests without any console
1227
self.old_stdout = sys.stdout
1228
sys.stdout = StringIOWrapper()
1229
self.addCleanup(self.restoreUIFactory)
1231
def restoreUIFactory(self):
1232
ui.ui_factory = self.old_factory
1233
sys.stdout = self.old_stdout
1235
def get_user_url(self, user=None, password=None):
1236
"""Build an url embedding user and password"""
1237
url = '%s://' % self.server._url_protocol
1238
if user is not None:
1240
if password is not None:
1241
url += ':' + password
1243
url += '%s:%s/' % (self.server.host, self.server.port)
1246
def test_no_user(self):
1247
self.server.add_user('joe', 'foo')
1248
t = self.get_user_transport()
1249
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1250
# Only one 'Authentication Required' error should occur
1251
self.assertEqual(1, self.server.auth_required_errors)
1253
def test_empty_pass(self):
1254
self.server.add_user('joe', '')
1255
t = self.get_user_transport('joe', '')
1256
self.assertEqual('contents of a\n', t.get('a').read())
1257
# Only one 'Authentication Required' error should occur
1258
self.assertEqual(1, self.server.auth_required_errors)
1260
def test_user_pass(self):
1261
self.server.add_user('joe', 'foo')
1262
t = self.get_user_transport('joe', 'foo')
1263
self.assertEqual('contents of a\n', t.get('a').read())
1264
# Only one 'Authentication Required' error should occur
1265
self.assertEqual(1, self.server.auth_required_errors)
1267
def test_unknown_user(self):
1268
self.server.add_user('joe', 'foo')
1269
t = self.get_user_transport('bill', 'foo')
1270
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1271
# Two 'Authentication Required' errors should occur (the
1272
# initial 'who are you' and 'I don't know you, who are
1274
self.assertEqual(2, self.server.auth_required_errors)
1276
def test_wrong_pass(self):
1277
self.server.add_user('joe', 'foo')
1278
t = self.get_user_transport('joe', 'bar')
1279
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1280
# Two 'Authentication Required' errors should occur (the
1281
# initial 'who are you' and 'this is not you, who are you')
1282
self.assertEqual(2, self.server.auth_required_errors)
1284
def test_prompt_for_password(self):
1285
self.server.add_user('joe', 'foo')
1286
t = self.get_user_transport('joe', None)
1287
ui.ui_factory = TestUIFactory(stdin='foo\n')
1288
self.assertEqual('contents of a\n',t.get('a').read())
1289
# stdin should be empty
1290
self.assertEqual('', ui.ui_factory.stdin.readline())
1291
# And we shouldn't prompt again for a different request
1292
# against the same transport.
1293
self.assertEqual('contents of b\n',t.get('b').read())
1295
# And neither against a clone
1296
self.assertEqual('contents of b\n',t2.get('b').read())
1297
# Only one 'Authentication Required' error should occur
1298
self.assertEqual(1, self.server.auth_required_errors)
1301
class TestHTTPAuth(TestAuth):
1302
"""Test HTTP authentication schemes.
1304
Daughter classes MUST inherit from TestCaseWithWebserver too.
1307
_auth_header = 'Authorization'
1310
TestCaseWithWebserver.setUp(self)
1311
self.server = self.get_readonly_server()
1312
TestAuth.setUp(self)
1314
def get_user_transport(self, user=None, password=None):
1315
return self._transport(self.get_user_url(user, password))
1318
class TestProxyAuth(TestAuth):
1319
"""Test proxy authentication schemes.
1321
Daughter classes MUST also inherit from TestCaseWithWebserver.
1323
_auth_header = 'Proxy-authorization'
1326
TestCaseWithWebserver.setUp(self)
1327
self.server = self.get_readonly_server()
1329
self.addCleanup(self._restore_env)
1330
TestAuth.setUp(self)
1331
# Override the contents to avoid false positives
1332
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1333
('b', 'not proxied contents of b\n'),
1334
('a-proxied', 'contents of a\n'),
1335
('b-proxied', 'contents of b\n'),
1338
def get_user_transport(self, user=None, password=None):
1339
self._install_env({'all_proxy': self.get_user_url(user, password)})
1340
return self._transport(self.server.get_url())
1342
def _install_env(self, env):
1343
for name, value in env.iteritems():
1344
self._old_env[name] = osutils.set_or_unset_env(name, value)
1346
def _restore_env(self):
1347
for name, value in self._old_env.iteritems():
1348
osutils.set_or_unset_env(name, value)
1351
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1352
"""Test http basic authentication scheme"""
1354
_transport = HttpTransport_urllib
1356
def create_transport_readonly_server(self):
1357
return HTTPBasicAuthServer()
1360
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1361
"""Test proxy basic authentication scheme"""
1363
_transport = HttpTransport_urllib
1365
def create_transport_readonly_server(self):
1366
return ProxyBasicAuthServer()
1369
class TestDigestAuth(object):
1370
"""Digest Authentication specific tests"""
1372
def test_changing_nonce(self):
1373
self.server.add_user('joe', 'foo')
1374
t = self.get_user_transport('joe', 'foo')
1375
self.assertEqual('contents of a\n', t.get('a').read())
1376
self.assertEqual('contents of b\n', t.get('b').read())
1377
# Only one 'Authentication Required' error should have
1379
self.assertEqual(1, self.server.auth_required_errors)
1380
# The server invalidates the current nonce
1381
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1382
self.assertEqual('contents of a\n', t.get('a').read())
1383
# Two 'Authentication Required' errors should occur (the
1384
# initial 'who are you' and a second 'who are you' with the new nonce)
1385
self.assertEqual(2, self.server.auth_required_errors)
1388
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1389
"""Test http digest authentication scheme"""
1391
_transport = HttpTransport_urllib
1393
def create_transport_readonly_server(self):
1394
return HTTPDigestAuthServer()
1397
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1398
TestCaseWithWebserver):
1399
"""Test proxy digest authentication scheme"""
1401
_transport = HttpTransport_urllib
1403
def create_transport_readonly_server(self):
1404
return ProxyDigestAuthServer()
161
self.assertIsInstance(t, bzrlib.transport.http._urllib.HttpTransport_urllib)
added
removed
bzrlib/tests/test_http.py
