1
# Copyright (C) 2005, 2006 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
# TODO: What about renaming to bzrlib.tests.transport.http ?
29
from bzrlib import errors
30
from bzrlib import osutils
31
from bzrlib.tests import (
35
from bzrlib.tests.HttpServer import (
40
from bzrlib.tests.HTTPTestUtil import (
41
BadProtocolRequestHandler,
42
BadStatusRequestHandler,
43
FakeProxyRequestHandler,
44
ForbiddenRequestHandler,
45
InvalidStatusRequestHandler,
46
NoRangeRequestHandler,
47
SingleRangeRequestHandler,
48
TestCaseWithTwoWebservers,
49
TestCaseWithWebserver,
52
from bzrlib.transport import (
56
from bzrlib.transport.http import (
60
from bzrlib.transport.http._urllib import HttpTransport_urllib
63
class FakeManager(object):
68
def add_password(self, realm, host, username, password):
69
self.credentials.append([realm, host, username, password])
72
class RecordingServer(object):
73
"""A fake HTTP server.
75
It records the bytes sent to it, and replies with a 200.
78
def __init__(self, expect_body_tail=None):
81
:type expect_body_tail: str
82
:param expect_body_tail: a reply won't be sent until this string is
85
self._expect_body_tail = expect_body_tail
88
self.received_bytes = ''
91
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
92
self._sock.bind(('127.0.0.1', 0))
93
self.host, self.port = self._sock.getsockname()
94
self._ready = threading.Event()
95
self._thread = threading.Thread(target=self._accept_read_and_reply)
96
self._thread.setDaemon(True)
100
def _accept_read_and_reply(self):
103
self._sock.settimeout(5)
105
conn, address = self._sock.accept()
106
# On win32, the accepted connection will be non-blocking to start
107
# with because we're using settimeout.
108
conn.setblocking(True)
109
while not self.received_bytes.endswith(self._expect_body_tail):
110
self.received_bytes += conn.recv(4096)
111
conn.sendall('HTTP/1.1 200 OK\r\n')
112
except socket.timeout:
113
# Make sure the client isn't stuck waiting for us to e.g. accept.
116
# The client may have already closed the socket.
123
# We might have already closed it. We don't care.
129
class TestWithTransport_pycurl(object):
130
"""Test case to inherit from if pycurl is present"""
132
def _get_pycurl_maybe(self):
134
from bzrlib.transport.http._pycurl import PyCurlTransport
135
return PyCurlTransport
136
except errors.DependencyNotPresent:
137
raise TestSkipped('pycurl not present')
139
_transport = property(_get_pycurl_maybe)
142
class TestHttpUrls(TestCase):
144
# TODO: This should be moved to authorization tests once they
147
def test_url_parsing(self):
149
url = extract_auth('http://example.com', f)
150
self.assertEquals('http://example.com', url)
151
self.assertEquals(0, len(f.credentials))
152
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
153
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
154
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.
170
def test_abs_url(self):
171
"""Construction of absolute http URLs"""
172
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
173
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')
177
eq(t.abspath('.bzr/1//2/./3'),
178
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
180
def test_invalid_http_urls(self):
181
"""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),
188
def test_http_root_urls(self):
189
"""Construction of URLs from server root"""
190
t = self._transport('http://bzr.ozlabs.org/')
191
eq = self.assertEqualDiff
192
eq(t.abspath('.bzr/tree-version'),
193
'http://bzr.ozlabs.org/.bzr/tree-version')
195
def test_http_impl_urls(self):
196
"""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
234
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)
267
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
268
transport=self.get_transport())
270
def test_http_has(self):
271
server = self.get_readonly_server()
272
t = self._transport(server.get_url())
273
self.assertEqual(t.has('foo/bar'), True)
274
self.assertEqual(len(server.logs), 1)
275
self.assertContainsRe(server.logs[0],
276
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
278
def test_http_has_not_found(self):
279
server = self.get_readonly_server()
280
t = self._transport(server.get_url())
281
self.assertEqual(t.has('not-found'), False)
282
self.assertContainsRe(server.logs[1],
283
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
285
def test_http_get(self):
286
server = self.get_readonly_server()
287
t = self._transport(server.get_url())
288
fp = t.get('foo/bar')
289
self.assertEqualDiff(
291
'contents of foo/bar\n')
292
self.assertEqual(len(server.logs), 1)
293
self.assertTrue(server.logs[0].find(
294
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
295
% bzrlib.__version__) > -1)
297
def test_get_smart_medium(self):
298
# For HTTP, get_smart_medium should return the transport object.
299
server = self.get_readonly_server()
300
http_transport = self._transport(server.get_url())
301
medium = http_transport.get_smart_medium()
302
self.assertIs(medium, http_transport)
304
def test_has_on_bogus_host(self):
305
# Get a free address and don't 'accept' on it, so that we
306
# can be sure there is no http handler there, but set a
307
# reasonable timeout to not slow down tests too much.
308
default_timeout = socket.getdefaulttimeout()
310
socket.setdefaulttimeout(2)
312
s.bind(('localhost', 0))
313
t = self._transport('http://%s:%s/' % s.getsockname())
314
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
316
socket.setdefaulttimeout(default_timeout)
319
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
320
"""Test http connections with urllib"""
322
_transport = HttpTransport_urllib
326
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
328
TestCaseWithWebserver):
329
"""Test http connections with pycurl"""
332
class TestHttpTransportRegistration(TestCase):
333
"""Test registrations of various http implementations"""
335
def test_http_registered(self):
336
# urlllib should always be present
337
t = get_transport('http+urllib://bzr.google.com/')
338
self.assertIsInstance(t, Transport)
339
self.assertIsInstance(t, HttpTransport_urllib)
342
class TestOffsets(TestCase):
343
"""Test offsets_to_ranges method"""
345
def test_offsets_to_ranges_simple(self):
346
to_range = HttpTransportBase.offsets_to_ranges
347
ranges = to_range([(10, 1)])
348
self.assertEqual([[10, 10]], ranges)
350
ranges = to_range([(0, 1), (1, 1)])
351
self.assertEqual([[0, 1]], ranges)
353
ranges = to_range([(1, 1), (0, 1)])
354
self.assertEqual([[0, 1]], ranges)
356
def test_offset_to_ranges_overlapped(self):
357
to_range = HttpTransportBase.offsets_to_ranges
359
ranges = to_range([(10, 1), (20, 2), (22, 5)])
360
self.assertEqual([[10, 10], [20, 26]], ranges)
362
ranges = to_range([(10, 1), (11, 2), (22, 5)])
363
self.assertEqual([[10, 12], [22, 26]], ranges)
366
class TestPost(object):
368
def _test_post_body_is_received(self, scheme):
369
server = RecordingServer(expect_body_tail='end-of-body')
371
self.addCleanup(server.tearDown)
372
url = '%s://%s:%s/' % (scheme, server.host, server.port)
374
http_transport = get_transport(url)
375
except errors.UnsupportedProtocol:
376
raise TestSkipped('%s not available' % scheme)
377
code, response = http_transport._post('abc def end-of-body')
379
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
380
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
381
# The transport should not be assuming that the server can accept
382
# chunked encoding the first time it connects, because HTTP/1.1, so we
383
# check for the literal string.
385
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
388
class TestPost_urllib(TestCase, TestPost):
389
"""TestPost for urllib implementation"""
391
_transport = HttpTransport_urllib
393
def test_post_body_is_received_urllib(self):
394
self._test_post_body_is_received('http+urllib')
397
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
398
"""TestPost for pycurl implementation"""
400
def test_post_body_is_received_pycurl(self):
401
self._test_post_body_is_received('http+pycurl')
404
class TestRangeHeader(TestCase):
405
"""Test range_header method"""
407
def check_header(self, value, ranges=[], tail=0):
408
range_header = HttpTransportBase.range_header
409
self.assertEqual(value, range_header(ranges, tail))
411
def test_range_header_single(self):
412
self.check_header('0-9', ranges=[[0,9]])
413
self.check_header('100-109', ranges=[[100,109]])
415
def test_range_header_tail(self):
416
self.check_header('-10', tail=10)
417
self.check_header('-50', tail=50)
419
def test_range_header_multi(self):
420
self.check_header('0-9,100-200,300-5000',
421
ranges=[(0,9), (100, 200), (300,5000)])
423
def test_range_header_mixed(self):
424
self.check_header('0-9,300-5000,-50',
425
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."""
705
def _set_and_capture_env_var(self, name, new_value):
706
"""Set an environment variable, and reset it when finished."""
707
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
709
def _install_env(self, env):
710
for name, value in env.iteritems():
711
self._set_and_capture_env_var(name, value)
713
def _restore_env(self):
714
for name, value in self._old_env.iteritems():
715
osutils.set_or_unset_env(name, value)
717
def _proxied_request(self):
718
from bzrlib.transport.http._urllib2_wrappers import (
723
handler = ProxyHandler()
724
request = Request('GET','http://baz/buzzle')
725
handler.set_proxy(request, 'http')
728
def test_empty_user(self):
729
self._install_env({'http_proxy': 'http://bar.com'})
730
request = self._proxied_request()
731
self.assertFalse(request.headers.has_key('Proxy-authorization'))
733
def test_empty_pass(self):
734
self._install_env({'http_proxy': 'http://joe@bar.com'})
735
request = self._proxied_request()
736
self.assertEqual('Basic ' + 'joe:'.encode('base64').strip(),
737
request.headers['Proxy-authorization'])
738
def test_user_pass(self):
739
self._install_env({'http_proxy': 'http://joe:foo@bar.com'})
740
request = self._proxied_request()
741
self.assertEqual('Basic ' + 'joe:foo'.encode('base64').strip(),
742
request.headers['Proxy-authorization'])
746
class TestProxyHttpServer(object):
747
"""Tests proxy server.
749
This MUST be used by daughter classes that also inherit from
750
TestCaseWithTwoWebservers.
752
We can't inherit directly from TestCaseWithTwoWebservers or
753
the test framework will try to create an instance which
754
cannot run, its implementation being incomplete.
756
Be aware that we do not setup a real proxy here. Instead, we
757
check that the *connection* goes through the proxy by serving
758
different content (the faked proxy server append '-proxied'
762
# FIXME: We don't have an https server available, so we don't
763
# test https connections.
765
# FIXME: Once the test suite is better fitted to test
766
# authorization schemes, test proxy authorizations too (see
770
TestCaseWithTwoWebservers.setUp(self)
771
self.build_tree_contents([('foo', 'contents of foo\n'),
772
('foo-proxied', 'proxied contents of foo\n')])
773
# Let's setup some attributes for tests
774
self.server = self.get_readonly_server()
775
self.no_proxy_host = 'localhost:%d' % self.server.port
776
# The secondary server is the proxy
777
self.proxy = self.get_secondary_server()
778
self.proxy_url = self.proxy.get_url()
781
def create_transport_secondary_server(self):
782
"""Creates an http server that will serve files with
783
'-proxied' appended to their names.
785
return HttpServer(FakeProxyRequestHandler)
787
def _set_and_capture_env_var(self, name, new_value):
788
"""Set an environment variable, and reset it when finished."""
789
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
791
def _install_env(self, env):
792
for name, value in env.iteritems():
793
self._set_and_capture_env_var(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_in_env(self, env):
800
self._install_env(env)
801
url = self.server.get_url()
802
t = self._transport(url)
804
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
808
def not_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(), 'contents of foo\n')
817
def test_http_proxy(self):
818
self.proxied_in_env({'http_proxy': self.proxy_url})
820
def test_HTTP_PROXY(self):
821
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
823
def test_all_proxy(self):
824
self.proxied_in_env({'all_proxy': self.proxy_url})
826
def test_ALL_PROXY(self):
827
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
829
def test_http_proxy_with_no_proxy(self):
830
self.not_proxied_in_env({'http_proxy': self.proxy_url,
831
'no_proxy': self.no_proxy_host})
833
def test_HTTP_PROXY_with_NO_PROXY(self):
834
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
835
'NO_PROXY': self.no_proxy_host})
837
def test_all_proxy_with_no_proxy(self):
838
self.not_proxied_in_env({'all_proxy': self.proxy_url,
839
'no_proxy': self.no_proxy_host})
841
def test_ALL_PROXY_with_NO_PROXY(self):
842
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
843
'NO_PROXY': self.no_proxy_host})
846
class TestProxyHttpServer_urllib(TestProxyHttpServer,
847
TestCaseWithTwoWebservers):
848
"""Tests proxy server for urllib implementation"""
850
_transport = HttpTransport_urllib
853
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
855
TestCaseWithTwoWebservers):
856
"""Tests proxy server for pycurl implementation"""
859
TestProxyHttpServer.setUp(self)
860
# Oh my ! pycurl does not check for the port as part of
861
# no_proxy :-( So we just test the host part
862
self.no_proxy_host = 'localhost'
864
def test_HTTP_PROXY(self):
865
# pycurl do not check HTTP_PROXY for security reasons
866
# (for use in a CGI context that we do not care
867
# about. Should we ?)
870
def test_HTTP_PROXY_with_NO_PROXY(self):
874
class TestRanges(object):
875
"""Test the Range header in GET methods..
877
This MUST be used by daughter classes that also inherit from
878
TestCaseWithWebserver.
880
We can't inherit directly from TestCaseWithWebserver or the
881
test framework will try to create an instance which cannot
882
run, its implementation being incomplete.
886
TestCaseWithWebserver.setUp(self)
887
self.build_tree_contents([('a', '0123456789')],)
888
server = self.get_readonly_server()
889
self.transport = self._transport(server.get_url())
891
def _file_contents(self, relpath, ranges, tail_amount=0):
892
code, data = self.transport._get(relpath, ranges)
893
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
894
for start, end in ranges:
896
yield data.read(end - start + 1)
898
def _file_tail(self, relpath, tail_amount):
899
code, data = self.transport._get(relpath, [], tail_amount)
900
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
901
data.seek(-tail_amount + 1, 2)
902
return data.read(tail_amount)
904
def test_range_header(self):
906
map(self.assertEqual,['0', '234'],
907
list(self._file_contents('a', [(0,0), (2,4)])),)
909
self.assertEqual('789', self._file_tail('a', 3))
910
# Syntactically invalid range
911
self.assertRaises(errors.InvalidRange,
912
self.transport._get, 'a', [(4, 3)])
913
# Semantically invalid range
914
self.assertRaises(errors.InvalidRange,
915
self.transport._get, 'a', [(42, 128)])
918
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
919
"""Test the Range header in GET methods for urllib implementation"""
921
_transport = HttpTransport_urllib
924
class TestRanges_pycurl(TestWithTransport_pycurl,
926
TestCaseWithWebserver):
927
"""Test the Range header in GET methods for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
