368
220
self.check_header('0-9,300-5000,-50',
369
221
ranges=[(0,9), (300,5000)],
373
class TestWallServer(object):
374
"""Tests exceptions during the connection phase"""
376
def create_transport_readonly_server(self):
377
return HttpServer(WallRequestHandler)
379
def test_http_has(self):
380
server = self.get_readonly_server()
381
t = self._transport(server.get_url())
382
# Unfortunately httplib (see HTTPResponse._read_status
383
# for details) make no distinction between a closed
384
# socket and badly formatted status line, so we can't
385
# just test for ConnectionError, we have to test
386
# InvalidHttpResponse too.
387
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
390
def test_http_get(self):
391
server = self.get_readonly_server()
392
t = self._transport(server.get_url())
393
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
397
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
398
"""Tests "wall" server for urllib implementation"""
400
_transport = HttpTransport_urllib
403
class TestWallServer_pycurl(TestWithTransport_pycurl,
405
TestCaseWithWebserver):
406
"""Tests "wall" server for pycurl implementation"""
409
class TestBadStatusServer(object):
410
"""Tests bad status from server."""
412
def create_transport_readonly_server(self):
413
return HttpServer(BadStatusRequestHandler)
415
def test_http_has(self):
416
server = self.get_readonly_server()
417
t = self._transport(server.get_url())
418
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
420
def test_http_get(self):
421
server = self.get_readonly_server()
422
t = self._transport(server.get_url())
423
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
426
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
427
"""Tests bad status server for urllib implementation"""
429
_transport = HttpTransport_urllib
432
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
434
TestCaseWithWebserver):
435
"""Tests bad status server for pycurl implementation"""
438
class TestInvalidStatusServer(TestBadStatusServer):
439
"""Tests invalid status from server.
441
Both implementations raises the same error as for a bad status.
444
def create_transport_readonly_server(self):
445
return HttpServer(InvalidStatusRequestHandler)
448
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
449
TestCaseWithWebserver):
450
"""Tests invalid status server for urllib implementation"""
452
_transport = HttpTransport_urllib
455
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
456
TestInvalidStatusServer,
457
TestCaseWithWebserver):
458
"""Tests invalid status server for pycurl implementation"""
461
class TestBadProtocolServer(object):
462
"""Tests bad protocol from server."""
464
def create_transport_readonly_server(self):
465
return HttpServer(BadProtocolRequestHandler)
467
def test_http_has(self):
468
server = self.get_readonly_server()
469
t = self._transport(server.get_url())
470
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
472
def test_http_get(self):
473
server = self.get_readonly_server()
474
t = self._transport(server.get_url())
475
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
478
class TestBadProtocolServer_urllib(TestBadProtocolServer,
479
TestCaseWithWebserver):
480
"""Tests bad protocol server for urllib implementation"""
482
_transport = HttpTransport_urllib
484
# curl don't check the protocol version
485
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
486
# TestBadProtocolServer,
487
# TestCaseWithWebserver):
488
# """Tests bad protocol server for pycurl implementation"""
491
class TestForbiddenServer(object):
492
"""Tests forbidden server"""
494
def create_transport_readonly_server(self):
495
return HttpServer(ForbiddenRequestHandler)
497
def test_http_has(self):
498
server = self.get_readonly_server()
499
t = self._transport(server.get_url())
500
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
502
def test_http_get(self):
503
server = self.get_readonly_server()
504
t = self._transport(server.get_url())
505
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
508
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
509
"""Tests forbidden server for urllib implementation"""
511
_transport = HttpTransport_urllib
514
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
516
TestCaseWithWebserver):
517
"""Tests forbidden server for pycurl implementation"""
520
class TestRecordingServer(TestCase):
522
def test_create(self):
523
server = RecordingServer(expect_body_tail=None)
524
self.assertEqual('', server.received_bytes)
525
self.assertEqual(None, server.host)
526
self.assertEqual(None, server.port)
528
def test_setUp_and_tearDown(self):
529
server = RecordingServer(expect_body_tail=None)
532
self.assertNotEqual(None, server.host)
533
self.assertNotEqual(None, server.port)
536
self.assertEqual(None, server.host)
537
self.assertEqual(None, server.port)
539
def test_send_receive_bytes(self):
540
server = RecordingServer(expect_body_tail='c')
542
self.addCleanup(server.tearDown)
543
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
544
sock.connect((server.host, server.port))
546
self.assertEqual('HTTP/1.1 200 OK\r\n',
547
osutils.recv_all(sock, 4096))
548
self.assertEqual('abc', server.received_bytes)
551
class TestRangeRequestServer(object):
552
"""Tests readv requests against server.
554
This MUST be used by daughter classes that also inherit from
555
TestCaseWithWebserver.
557
We can't inherit directly from TestCaseWithWebserver or the
558
test framework will try to create an instance which cannot
559
run, its implementation being incomplete.
563
TestCaseWithWebserver.setUp(self)
564
self.build_tree_contents([('a', '0123456789')],)
566
def test_readv(self):
567
server = self.get_readonly_server()
568
t = self._transport(server.get_url())
569
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
570
self.assertEqual(l[0], (0, '0'))
571
self.assertEqual(l[1], (1, '1'))
572
self.assertEqual(l[2], (3, '34'))
573
self.assertEqual(l[3], (9, '9'))
575
def test_readv_out_of_order(self):
576
server = self.get_readonly_server()
577
t = self._transport(server.get_url())
578
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
579
self.assertEqual(l[0], (1, '1'))
580
self.assertEqual(l[1], (9, '9'))
581
self.assertEqual(l[2], (0, '0'))
582
self.assertEqual(l[3], (3, '34'))
584
def test_readv_invalid_ranges(self):
585
server = self.get_readonly_server()
586
t = self._transport(server.get_url())
588
# This is intentionally reading off the end of the file
589
# since we are sure that it cannot get there
590
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
591
t.readv, 'a', [(1,1), (8,10)])
593
# This is trying to seek past the end of the file, it should
594
# also raise a special error
595
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
596
t.readv, 'a', [(12,2)])
599
class TestSingleRangeRequestServer(TestRangeRequestServer):
600
"""Test readv against a server which accept only single range requests"""
602
def create_transport_readonly_server(self):
603
return HttpServer(SingleRangeRequestHandler)
606
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
607
TestCaseWithWebserver):
608
"""Tests single range requests accepting server for urllib implementation"""
610
_transport = HttpTransport_urllib
613
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
614
TestSingleRangeRequestServer,
615
TestCaseWithWebserver):
616
"""Tests single range requests accepting server for pycurl implementation"""
619
class TestNoRangeRequestServer(TestRangeRequestServer):
620
"""Test readv against a server which do not accept range requests"""
622
def create_transport_readonly_server(self):
623
return HttpServer(NoRangeRequestHandler)
626
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
627
TestCaseWithWebserver):
628
"""Tests range requests refusing server for urllib implementation"""
630
_transport = HttpTransport_urllib
633
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
634
TestNoRangeRequestServer,
635
TestCaseWithWebserver):
636
"""Tests range requests refusing server for pycurl implementation"""
639
class TestHttpProxyWhiteBox(TestCase):
640
"""Whitebox test proxy http authorization."""
649
def _set_and_capture_env_var(self, name, new_value):
650
"""Set an environment variable, and reset it when finished."""
651
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
653
def _install_env(self, env):
654
for name, value in env.iteritems():
655
self._set_and_capture_env_var(name, value)
657
def _restore_env(self):
658
for name, value in self._old_env.iteritems():
659
osutils.set_or_unset_env(name, value)
661
def _proxied_request(self):
662
from bzrlib.transport.http._urllib2_wrappers import (
667
handler = ProxyHandler()
668
request = Request('GET','http://baz/buzzle')
669
handler.set_proxy(request, 'http')
672
def test_empty_user(self):
673
self._install_env({'http_proxy': 'http://bar.com'})
674
request = self._proxied_request()
675
self.assertFalse(request.headers.has_key('Proxy-authorization'))
677
def test_empty_pass(self):
678
self._install_env({'http_proxy': 'http://joe@bar.com'})
679
request = self._proxied_request()
680
self.assertEqual('Basic ' + 'joe:'.encode('base64').strip(),
681
request.headers['Proxy-authorization'])
682
def test_user_pass(self):
683
self._install_env({'http_proxy': 'http://joe:foo@bar.com'})
684
request = self._proxied_request()
685
self.assertEqual('Basic ' + 'joe:foo'.encode('base64').strip(),
686
request.headers['Proxy-authorization'])
690
class TestProxyHttpServer(object):
691
"""Tests proxy server.
693
This MUST be used by daughter classes that also inherit from
694
TestCaseWithTwoWebservers.
696
We can't inherit directly from TestCaseWithTwoWebservers or
697
the test framework will try to create an instance which
698
cannot run, its implementation being incomplete.
700
Be aware that we do not setup a real proxy here. Instead, we
701
check that the *connection* goes through the proxy by serving
702
different content (the faked proxy server append '-proxied'
706
# FIXME: We don't have an https server available, so we don't
707
# test https connections.
709
# FIXME: Once the test suite is better fitted to test
710
# authorization schemes, test proxy authorizations too (see
714
TestCaseWithTwoWebservers.setUp(self)
715
self.build_tree_contents([('foo', 'contents of foo\n'),
716
('foo-proxied', 'proxied contents of foo\n')])
717
# Let's setup some attributes for tests
718
self.server = self.get_readonly_server()
719
self.no_proxy_host = 'localhost:%d' % self.server.port
720
# The secondary server is the proxy
721
self.proxy = self.get_secondary_server()
722
self.proxy_url = self.proxy.get_url()
725
def create_transport_secondary_server(self):
726
"""Creates an http server that will serve files with
727
'-proxied' appended to their names.
729
return HttpServer(FakeProxyRequestHandler)
731
def _set_and_capture_env_var(self, name, new_value):
732
"""Set an environment variable, and reset it when finished."""
733
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
735
def _install_env(self, env):
736
for name, value in env.iteritems():
737
self._set_and_capture_env_var(name, value)
739
def _restore_env(self):
740
for name, value in self._old_env.iteritems():
741
osutils.set_or_unset_env(name, value)
743
def proxied_in_env(self, env):
744
self._install_env(env)
745
url = self.server.get_url()
746
t = self._transport(url)
748
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
752
def not_proxied_in_env(self, env):
753
self._install_env(env)
754
url = self.server.get_url()
755
t = self._transport(url)
757
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
761
def test_http_proxy(self):
762
self.proxied_in_env({'http_proxy': self.proxy_url})
764
def test_HTTP_PROXY(self):
765
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
767
def test_all_proxy(self):
768
self.proxied_in_env({'all_proxy': self.proxy_url})
770
def test_ALL_PROXY(self):
771
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
773
def test_http_proxy_with_no_proxy(self):
774
self.not_proxied_in_env({'http_proxy': self.proxy_url,
775
'no_proxy': self.no_proxy_host})
777
def test_HTTP_PROXY_with_NO_PROXY(self):
778
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
779
'NO_PROXY': self.no_proxy_host})
781
def test_all_proxy_with_no_proxy(self):
782
self.not_proxied_in_env({'all_proxy': self.proxy_url,
783
'no_proxy': self.no_proxy_host})
785
def test_ALL_PROXY_with_NO_PROXY(self):
786
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
787
'NO_PROXY': self.no_proxy_host})
790
class TestProxyHttpServer_urllib(TestProxyHttpServer,
791
TestCaseWithTwoWebservers):
792
"""Tests proxy server for urllib implementation"""
794
_transport = HttpTransport_urllib
797
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
799
TestCaseWithTwoWebservers):
800
"""Tests proxy server for pycurl implementation"""
803
TestProxyHttpServer.setUp(self)
804
# Oh my ! pycurl does not check for the port as part of
805
# no_proxy :-( So we just test the host part
806
self.no_proxy_host = 'localhost'
808
def test_HTTP_PROXY(self):
809
# pycurl do not check HTTP_PROXY for security reasons
810
# (for use in a CGI context that we do not care
811
# about. Should we ?)
814
def test_HTTP_PROXY_with_NO_PROXY(self):
818
class TestRanges(object):
819
"""Test the Range header in GET methods..
821
This MUST be used by daughter classes that also inherit from
822
TestCaseWithWebserver.
824
We can't inherit directly from TestCaseWithWebserver or the
825
test framework will try to create an instance which cannot
826
run, its implementation being incomplete.
830
TestCaseWithWebserver.setUp(self)
831
self.build_tree_contents([('a', '0123456789')],)
832
server = self.get_readonly_server()
833
self.transport = self._transport(server.get_url())
835
def _file_contents(self, relpath, ranges, tail_amount=0):
836
code, data = self.transport._get(relpath, ranges)
837
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
838
for start, end in ranges:
840
yield data.read(end - start + 1)
842
def _file_tail(self, relpath, tail_amount):
843
code, data = self.transport._get(relpath, [], tail_amount)
844
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
845
data.seek(-tail_amount + 1, 2)
846
return data.read(tail_amount)
848
def test_range_header(self):
850
map(self.assertEqual,['0', '234'],
851
list(self._file_contents('a', [(0,0), (2,4)])),)
853
self.assertEqual('789', self._file_tail('a', 3))
854
# Syntactically invalid range
855
self.assertRaises(errors.InvalidRange,
856
self.transport._get, 'a', [(4, 3)])
857
# Semantically invalid range
858
self.assertRaises(errors.InvalidRange,
859
self.transport._get, 'a', [(42, 128)])
862
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
863
"""Test the Range header in GET methods for urllib implementation"""
865
_transport = HttpTransport_urllib
868
class TestRanges_pycurl(TestWithTransport_pycurl,
870
TestCaseWithWebserver):
871
"""Test the Range header in GET methods for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
