152
49
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
153
50
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
154
51
self.assertEquals(1, len(f.credentials))
155
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
159
class TestHttpTransportUrls(object):
160
"""Test the http urls.
162
This MUST be used by daughter classes that also inherit from
165
We can't inherit directly from TestCase or the
166
test framework will try to create an instance which cannot
167
run, its implementation being incomplete.
52
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'], f.credentials[0])
170
54
def test_abs_url(self):
171
55
"""Construction of absolute http URLs"""
172
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
56
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
173
57
eq = self.assertEqualDiff
174
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
175
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
176
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
59
'http://bazaar-vcs.org/bzr/bzr.dev')
60
eq(t.abspath('foo/bar'),
61
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
63
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
177
64
eq(t.abspath('.bzr/1//2/./3'),
178
65
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
180
67
def test_invalid_http_urls(self):
181
68
"""Trap invalid construction of urls"""
182
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
183
self.assertRaises(ValueError, t.abspath, '.bzr/')
184
t = self._transport('http://http://bazaar-vcs.org/bzr/bzr.dev/')
185
self.assertRaises((errors.InvalidURL, errors.ConnectionError),
69
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
70
self.assertRaises(ValueError,
188
74
def test_http_root_urls(self):
189
75
"""Construction of URLs from server root"""
190
t = self._transport('http://bzr.ozlabs.org/')
76
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
191
77
eq = self.assertEqualDiff
192
78
eq(t.abspath('.bzr/tree-version'),
193
79
'http://bzr.ozlabs.org/.bzr/tree-version')
195
81
def test_http_impl_urls(self):
196
82
"""There are servers which ask for particular clients to connect"""
197
server = self._server()
200
url = server.get_url()
201
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
206
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
207
"""Test http urls with urllib"""
209
_transport = HttpTransport_urllib
210
_server = HttpServer_urllib
211
_qualified_prefix = 'http+urllib'
214
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
216
"""Test http urls with pycurl"""
218
_server = HttpServer_PyCurl
219
_qualified_prefix = 'http+pycurl'
221
# TODO: This should really be moved into another pycurl
222
# specific test. When https tests will be implemented, take
223
# this one into account.
224
def test_pycurl_without_https_support(self):
225
"""Test that pycurl without SSL do not fail with a traceback.
227
For the purpose of the test, we force pycurl to ignore
228
https by supplying a fake version_info that do not
84
from bzrlib.transport.http._pycurl import HttpServer_PyCurl
85
server = HttpServer_PyCurl()
88
url = server.get_url()
89
self.assertTrue(url.startswith('http+pycurl://'))
92
except DependencyNotPresent:
234
93
raise TestSkipped('pycurl not present')
235
# Now that we have pycurl imported, we can fake its version_info
236
# This was taken from a windows pycurl without SSL
238
pycurl.version_info = lambda : (2,
246
('ftp', 'gopher', 'telnet',
247
'dict', 'ldap', 'http', 'file'),
251
self.assertRaises(errors.DependencyNotPresent, self._transport,
252
'https://launchpad.net')
254
class TestHttpConnections(object):
255
"""Test the http connections.
257
This MUST be used by daughter classes that also inherit from
258
TestCaseWithWebserver.
260
We can't inherit directly from TestCaseWithWebserver or the
261
test framework will try to create an instance which cannot
262
run, its implementation being incomplete.
266
TestCaseWithWebserver.setUp(self)
96
class TestHttpMixins(object):
267
99
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
268
100
transport=self.get_transport())
424
220
self.check_header('0-9,300-5000,-50',
425
221
ranges=[(0,9), (300,5000)],
429
class TestWallServer(object):
430
"""Tests exceptions during the connection phase"""
432
def create_transport_readonly_server(self):
433
return HttpServer(WallRequestHandler)
435
def test_http_has(self):
436
server = self.get_readonly_server()
437
t = self._transport(server.get_url())
438
# Unfortunately httplib (see HTTPResponse._read_status
439
# for details) make no distinction between a closed
440
# socket and badly formatted status line, so we can't
441
# just test for ConnectionError, we have to test
442
# InvalidHttpResponse too.
443
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
446
def test_http_get(self):
447
server = self.get_readonly_server()
448
t = self._transport(server.get_url())
449
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
453
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
454
"""Tests "wall" server for urllib implementation"""
456
_transport = HttpTransport_urllib
459
class TestWallServer_pycurl(TestWithTransport_pycurl,
461
TestCaseWithWebserver):
462
"""Tests "wall" server for pycurl implementation"""
465
class TestBadStatusServer(object):
466
"""Tests bad status from server."""
468
def create_transport_readonly_server(self):
469
return HttpServer(BadStatusRequestHandler)
471
def test_http_has(self):
472
server = self.get_readonly_server()
473
t = self._transport(server.get_url())
474
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
476
def test_http_get(self):
477
server = self.get_readonly_server()
478
t = self._transport(server.get_url())
479
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
482
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
483
"""Tests bad status server for urllib implementation"""
485
_transport = HttpTransport_urllib
488
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
490
TestCaseWithWebserver):
491
"""Tests bad status server for pycurl implementation"""
494
class TestInvalidStatusServer(TestBadStatusServer):
495
"""Tests invalid status from server.
497
Both implementations raises the same error as for a bad status.
500
def create_transport_readonly_server(self):
501
return HttpServer(InvalidStatusRequestHandler)
504
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
505
TestCaseWithWebserver):
506
"""Tests invalid status server for urllib implementation"""
508
_transport = HttpTransport_urllib
511
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
512
TestInvalidStatusServer,
513
TestCaseWithWebserver):
514
"""Tests invalid status server for pycurl implementation"""
517
class TestBadProtocolServer(object):
518
"""Tests bad protocol from server."""
520
def create_transport_readonly_server(self):
521
return HttpServer(BadProtocolRequestHandler)
523
def test_http_has(self):
524
server = self.get_readonly_server()
525
t = self._transport(server.get_url())
526
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
528
def test_http_get(self):
529
server = self.get_readonly_server()
530
t = self._transport(server.get_url())
531
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
534
class TestBadProtocolServer_urllib(TestBadProtocolServer,
535
TestCaseWithWebserver):
536
"""Tests bad protocol server for urllib implementation"""
538
_transport = HttpTransport_urllib
540
# curl don't check the protocol version
541
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
542
# TestBadProtocolServer,
543
# TestCaseWithWebserver):
544
# """Tests bad protocol server for pycurl implementation"""
547
class TestForbiddenServer(object):
548
"""Tests forbidden server"""
550
def create_transport_readonly_server(self):
551
return HttpServer(ForbiddenRequestHandler)
553
def test_http_has(self):
554
server = self.get_readonly_server()
555
t = self._transport(server.get_url())
556
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
558
def test_http_get(self):
559
server = self.get_readonly_server()
560
t = self._transport(server.get_url())
561
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
564
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
565
"""Tests forbidden server for urllib implementation"""
567
_transport = HttpTransport_urllib
570
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
572
TestCaseWithWebserver):
573
"""Tests forbidden server for pycurl implementation"""
576
class TestRecordingServer(TestCase):
578
def test_create(self):
579
server = RecordingServer(expect_body_tail=None)
580
self.assertEqual('', server.received_bytes)
581
self.assertEqual(None, server.host)
582
self.assertEqual(None, server.port)
584
def test_setUp_and_tearDown(self):
585
server = RecordingServer(expect_body_tail=None)
588
self.assertNotEqual(None, server.host)
589
self.assertNotEqual(None, server.port)
592
self.assertEqual(None, server.host)
593
self.assertEqual(None, server.port)
595
def test_send_receive_bytes(self):
596
server = RecordingServer(expect_body_tail='c')
598
self.addCleanup(server.tearDown)
599
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
600
sock.connect((server.host, server.port))
602
self.assertEqual('HTTP/1.1 200 OK\r\n',
603
osutils.recv_all(sock, 4096))
604
self.assertEqual('abc', server.received_bytes)
607
class TestRangeRequestServer(object):
608
"""Tests readv requests against server.
610
This MUST be used by daughter classes that also inherit from
611
TestCaseWithWebserver.
613
We can't inherit directly from TestCaseWithWebserver or the
614
test framework will try to create an instance which cannot
615
run, its implementation being incomplete.
619
TestCaseWithWebserver.setUp(self)
620
self.build_tree_contents([('a', '0123456789')],)
622
def test_readv(self):
623
server = self.get_readonly_server()
624
t = self._transport(server.get_url())
625
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
626
self.assertEqual(l[0], (0, '0'))
627
self.assertEqual(l[1], (1, '1'))
628
self.assertEqual(l[2], (3, '34'))
629
self.assertEqual(l[3], (9, '9'))
631
def test_readv_out_of_order(self):
632
server = self.get_readonly_server()
633
t = self._transport(server.get_url())
634
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
635
self.assertEqual(l[0], (1, '1'))
636
self.assertEqual(l[1], (9, '9'))
637
self.assertEqual(l[2], (0, '0'))
638
self.assertEqual(l[3], (3, '34'))
640
def test_readv_invalid_ranges(self):
641
server = self.get_readonly_server()
642
t = self._transport(server.get_url())
644
# This is intentionally reading off the end of the file
645
# since we are sure that it cannot get there
646
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
647
t.readv, 'a', [(1,1), (8,10)])
649
# This is trying to seek past the end of the file, it should
650
# also raise a special error
651
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
652
t.readv, 'a', [(12,2)])
655
class TestSingleRangeRequestServer(TestRangeRequestServer):
656
"""Test readv against a server which accept only single range requests"""
658
def create_transport_readonly_server(self):
659
return HttpServer(SingleRangeRequestHandler)
662
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
663
TestCaseWithWebserver):
664
"""Tests single range requests accepting server for urllib implementation"""
666
_transport = HttpTransport_urllib
669
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
670
TestSingleRangeRequestServer,
671
TestCaseWithWebserver):
672
"""Tests single range requests accepting server for pycurl implementation"""
675
class TestNoRangeRequestServer(TestRangeRequestServer):
676
"""Test readv against a server which do not accept range requests"""
678
def create_transport_readonly_server(self):
679
return HttpServer(NoRangeRequestHandler)
682
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
683
TestCaseWithWebserver):
684
"""Tests range requests refusing server for urllib implementation"""
686
_transport = HttpTransport_urllib
689
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
690
TestNoRangeRequestServer,
691
TestCaseWithWebserver):
692
"""Tests range requests refusing server for pycurl implementation"""
695
class TestHttpProxyWhiteBox(TestCase):
696
"""Whitebox test proxy http authorization.
698
These tests concern urllib implementation only.
708
def _set_and_capture_env_var(self, name, new_value):
709
"""Set an environment variable, and reset it when finished."""
710
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
712
def _install_env(self, env):
713
for name, value in env.iteritems():
714
self._set_and_capture_env_var(name, value)
716
def _restore_env(self):
717
for name, value in self._old_env.iteritems():
718
osutils.set_or_unset_env(name, value)
720
def _proxied_request(self):
721
from bzrlib.transport.http._urllib2_wrappers import (
726
handler = ProxyHandler()
727
request = Request('GET','http://baz/buzzle')
728
handler.set_proxy(request, 'http')
731
def test_empty_user(self):
732
self._install_env({'http_proxy': 'http://bar.com'})
733
request = self._proxied_request()
734
self.assertFalse(request.headers.has_key('Proxy-authorization'))
736
def test_empty_pass(self):
737
self._install_env({'http_proxy': 'http://joe@bar.com'})
738
request = self._proxied_request()
739
self.assertEqual('Basic ' + 'joe:'.encode('base64').strip(),
740
request.headers['Proxy-authorization'])
741
def test_user_pass(self):
742
self._install_env({'http_proxy': 'http://joe:foo@bar.com'})
743
request = self._proxied_request()
744
self.assertEqual('Basic ' + 'joe:foo'.encode('base64').strip(),
745
request.headers['Proxy-authorization'])
747
def test_invalid_proxy(self):
748
"""A proxy env variable without scheme"""
749
self._install_env({'http_proxy': 'host:1234'})
750
self.assertRaises(errors.InvalidURL, self._proxied_request)
753
class TestProxyHttpServer(object):
754
"""Tests proxy server.
756
This MUST be used by daughter classes that also inherit from
757
TestCaseWithTwoWebservers.
759
We can't inherit directly from TestCaseWithTwoWebservers or
760
the test framework will try to create an instance which
761
cannot run, its implementation being incomplete.
763
Be aware that we do not setup a real proxy here. Instead, we
764
check that the *connection* goes through the proxy by serving
765
different content (the faked proxy server append '-proxied'
769
# FIXME: We don't have an https server available, so we don't
770
# test https connections.
772
# FIXME: Once the test suite is better fitted to test
773
# authorization schemes, test proxy authorizations too (see
777
TestCaseWithTwoWebservers.setUp(self)
778
self.build_tree_contents([('foo', 'contents of foo\n'),
779
('foo-proxied', 'proxied contents of foo\n')])
780
# Let's setup some attributes for tests
781
self.server = self.get_readonly_server()
782
# FIXME: We should not rely on 'localhost' being the hostname
783
self.proxy_address = 'localhost:%d' % 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.
794
return HttpServer(FakeProxyRequestHandler)
796
def _set_and_capture_env_var(self, name, new_value):
797
"""Set an environment variable, and reset it when finished."""
798
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
800
def _install_env(self, env):
801
for name, value in env.iteritems():
802
self._set_and_capture_env_var(name, value)
804
def _restore_env(self):
805
for name, value in self._old_env.iteritems():
806
osutils.set_or_unset_env(name, value)
808
def proxied_in_env(self, env):
809
self._install_env(env)
810
url = self.server.get_url()
811
t = self._transport(url)
813
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
817
def not_proxied_in_env(self, env):
818
self._install_env(env)
819
url = self.server.get_url()
820
t = self._transport(url)
822
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
826
def test_http_proxy(self):
827
self.proxied_in_env({'http_proxy': self.proxy_url})
829
def test_HTTP_PROXY(self):
830
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
832
def test_all_proxy(self):
833
self.proxied_in_env({'all_proxy': self.proxy_url})
835
def test_ALL_PROXY(self):
836
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
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_HTTP_PROXY_with_NO_PROXY(self):
843
self.not_proxied_in_env({'HTTP_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_ALL_PROXY_with_NO_PROXY(self):
851
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
852
'NO_PROXY': self.no_proxy_host})
854
def test_http_proxy_without_scheme(self):
855
self.assertRaises(errors.InvalidURL,
857
{'http_proxy': self.proxy_address})
860
class TestProxyHttpServer_urllib(TestProxyHttpServer,
861
TestCaseWithTwoWebservers):
862
"""Tests proxy server for urllib implementation"""
864
_transport = HttpTransport_urllib
867
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
869
TestCaseWithTwoWebservers):
870
"""Tests proxy server for pycurl implementation"""
873
TestProxyHttpServer.setUp(self)
874
# Oh my ! pycurl does not check for the port as part of
875
# no_proxy :-( So we just test the host part
876
self.no_proxy_host = 'localhost'
878
def test_HTTP_PROXY(self):
879
# pycurl do not check HTTP_PROXY for security reasons
880
# (for use in a CGI context that we do not care
881
# about. Should we ?)
884
def test_HTTP_PROXY_with_NO_PROXY(self):
887
def test_http_proxy_without_scheme(self):
888
# pycurl *ignores* invalid proxy env variables. If that
889
# ever change in the future, this test will fail
890
# indicating that pycurl do not ignore anymore such
892
self.not_proxied_in_env({'http_proxy': self.proxy_address})
895
class TestRanges(object):
896
"""Test the Range header in GET methods..
898
This MUST be used by daughter classes that also inherit from
899
TestCaseWithWebserver.
901
We can't inherit directly from TestCaseWithWebserver or the
902
test framework will try to create an instance which cannot
903
run, its implementation being incomplete.
907
TestCaseWithWebserver.setUp(self)
908
self.build_tree_contents([('a', '0123456789')],)
909
server = self.get_readonly_server()
910
self.transport = self._transport(server.get_url())
912
def _file_contents(self, relpath, ranges, tail_amount=0):
913
code, data = self.transport._get(relpath, ranges)
914
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
915
for start, end in ranges:
917
yield data.read(end - start + 1)
919
def _file_tail(self, relpath, tail_amount):
920
code, data = self.transport._get(relpath, [], tail_amount)
921
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
922
data.seek(-tail_amount + 1, 2)
923
return data.read(tail_amount)
925
def test_range_header(self):
927
map(self.assertEqual,['0', '234'],
928
list(self._file_contents('a', [(0,0), (2,4)])),)
930
self.assertEqual('789', self._file_tail('a', 3))
931
# Syntactically invalid range
932
self.assertRaises(errors.InvalidRange,
933
self.transport._get, 'a', [(4, 3)])
934
# Semantically invalid range
935
self.assertRaises(errors.InvalidRange,
936
self.transport._get, 'a', [(42, 128)])
939
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
940
"""Test the Range header in GET methods for urllib implementation"""
942
_transport = HttpTransport_urllib
945
class TestRanges_pycurl(TestWithTransport_pycurl,
947
TestCaseWithWebserver):
948
"""Test the Range header in GET methods for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
