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.
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
