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 TestHttpUrls(TestCase):
131
# FIXME: Some of these tests should be done for both
134
def test_url_parsing(self):
136
url = extract_auth('http://example.com', f)
137
self.assertEquals('http://example.com', url)
138
self.assertEquals(0, len(f.credentials))
139
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
140
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
141
self.assertEquals(1, len(f.credentials))
142
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
145
def test_abs_url(self):
146
"""Construction of absolute http URLs"""
147
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
148
eq = self.assertEqualDiff
150
'http://bazaar-vcs.org/bzr/bzr.dev')
151
eq(t.abspath('foo/bar'),
152
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
153
eq(t.abspath('.bzr'),
154
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
155
eq(t.abspath('.bzr/1//2/./3'),
156
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
158
def test_invalid_http_urls(self):
159
"""Trap invalid construction of urls"""
160
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
161
self.assertRaises(ValueError,
164
t = HttpTransport_urllib('http://http://bazaar-vcs.org/bzr/bzr.dev/')
165
self.assertRaises(errors.InvalidURL, t.has, 'foo/bar')
167
def test_http_root_urls(self):
168
"""Construction of URLs from server root"""
169
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
170
eq = self.assertEqualDiff
171
eq(t.abspath('.bzr/tree-version'),
172
'http://bzr.ozlabs.org/.bzr/tree-version')
174
def test_http_impl_urls(self):
175
"""There are servers which ask for particular clients to connect"""
176
server = HttpServer_PyCurl()
179
url = server.get_url()
180
self.assertTrue(url.startswith('http+pycurl://'))
185
class TestHttpConnections(object):
186
"""Test the http connections.
188
This MUST be used by daughter classes that also inherit from
189
TestCaseWithWebserver.
191
We can't inherit directly from TestCaseWithWebserver or the
192
test framework will try to create an instance which cannot
193
run, its implementation being incomplete.
197
TestCaseWithWebserver.setUp(self)
198
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
199
transport=self.get_transport())
201
def test_http_has(self):
202
server = self.get_readonly_server()
203
t = self._transport(server.get_url())
204
self.assertEqual(t.has('foo/bar'), True)
205
self.assertEqual(len(server.logs), 1)
206
self.assertContainsRe(server.logs[0],
207
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
209
def test_http_has_not_found(self):
210
server = self.get_readonly_server()
211
t = self._transport(server.get_url())
212
self.assertEqual(t.has('not-found'), False)
213
self.assertContainsRe(server.logs[1],
214
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
216
def test_http_get(self):
217
server = self.get_readonly_server()
218
t = self._transport(server.get_url())
219
fp = t.get('foo/bar')
220
self.assertEqualDiff(
222
'contents of foo/bar\n')
223
self.assertEqual(len(server.logs), 1)
224
self.assertTrue(server.logs[0].find(
225
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
226
% bzrlib.__version__) > -1)
228
def test_get_smart_medium(self):
229
# For HTTP, get_smart_medium should return the transport object.
230
server = self.get_readonly_server()
231
http_transport = self._transport(server.get_url())
232
medium = http_transport.get_smart_medium()
233
self.assertIs(medium, http_transport)
235
def test_has_on_bogus_host(self):
236
# Get a free address and don't 'accept' on it, so that we
237
# can be sure there is no http handler there, but set a
238
# reasonable timeout to not slow down tests too much.
239
default_timeout = socket.getdefaulttimeout()
241
socket.setdefaulttimeout(2)
243
s.bind(('localhost', 0))
244
t = self._transport('http://%s:%s/' % s.getsockname())
245
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
247
socket.setdefaulttimeout(default_timeout)
250
class TestWithTransport_pycurl(object):
251
"""Test case to inherit from if pycurl is present"""
253
def _get_pycurl_maybe(self):
255
from bzrlib.transport.http._pycurl import PyCurlTransport
256
return PyCurlTransport
257
except errors.DependencyNotPresent:
258
raise TestSkipped('pycurl not present')
260
_transport = property(_get_pycurl_maybe)
263
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
264
"""Test http connections with urllib"""
266
_transport = HttpTransport_urllib
270
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
272
TestCaseWithWebserver):
273
"""Test http connections with pycurl"""
276
class TestHttpTransportRegistration(TestCase):
277
"""Test registrations of various http implementations"""
279
def test_http_registered(self):
280
# urlllib should always be present
281
t = get_transport('http+urllib://bzr.google.com/')
282
self.assertIsInstance(t, Transport)
283
self.assertIsInstance(t, HttpTransport_urllib)
286
class TestOffsets(TestCase):
287
"""Test offsets_to_ranges method"""
289
def test_offsets_to_ranges_simple(self):
290
to_range = HttpTransportBase.offsets_to_ranges
291
ranges = to_range([(10, 1)])
292
self.assertEqual([[10, 10]], ranges)
294
ranges = to_range([(0, 1), (1, 1)])
295
self.assertEqual([[0, 1]], ranges)
297
ranges = to_range([(1, 1), (0, 1)])
298
self.assertEqual([[0, 1]], ranges)
300
def test_offset_to_ranges_overlapped(self):
301
to_range = HttpTransportBase.offsets_to_ranges
303
ranges = to_range([(10, 1), (20, 2), (22, 5)])
304
self.assertEqual([[10, 10], [20, 26]], ranges)
306
ranges = to_range([(10, 1), (11, 2), (22, 5)])
307
self.assertEqual([[10, 12], [22, 26]], ranges)
310
class TestPost(object):
312
def _test_post_body_is_received(self, scheme):
313
server = RecordingServer(expect_body_tail='end-of-body')
315
self.addCleanup(server.tearDown)
316
url = '%s://%s:%s/' % (scheme, server.host, server.port)
318
http_transport = get_transport(url)
319
except errors.UnsupportedProtocol:
320
raise TestSkipped('%s not available' % scheme)
321
code, response = http_transport._post('abc def end-of-body')
323
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
324
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
325
# The transport should not be assuming that the server can accept
326
# chunked encoding the first time it connects, because HTTP/1.1, so we
327
# check for the literal string.
329
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
332
class TestPost_urllib(TestCase, TestPost):
333
"""TestPost for urllib implementation"""
335
_transport = HttpTransport_urllib
337
def test_post_body_is_received_urllib(self):
338
self._test_post_body_is_received('http+urllib')
341
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
342
"""TestPost for pycurl implementation"""
344
def test_post_body_is_received_pycurl(self):
345
self._test_post_body_is_received('http+pycurl')
348
class TestRangeHeader(TestCase):
349
"""Test range_header method"""
351
def check_header(self, value, ranges=[], tail=0):
352
range_header = HttpTransportBase.range_header
353
self.assertEqual(value, range_header(ranges, tail))
355
def test_range_header_single(self):
356
self.check_header('0-9', ranges=[[0,9]])
357
self.check_header('100-109', ranges=[[100,109]])
359
def test_range_header_tail(self):
360
self.check_header('-10', tail=10)
361
self.check_header('-50', tail=50)
363
def test_range_header_multi(self):
364
self.check_header('0-9,100-200,300-5000',
365
ranges=[(0,9), (100, 200), (300,5000)])
367
def test_range_header_mixed(self):
368
self.check_header('0-9,300-5000,-50',
369
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
