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 ?
34
from bzrlib.tests import (
38
from bzrlib.tests.HttpServer import (
43
from bzrlib.tests.HTTPTestUtil import (
44
BadProtocolRequestHandler,
45
BadStatusRequestHandler,
46
FakeProxyRequestHandler,
47
ForbiddenRequestHandler,
48
HTTPServerRedirecting,
49
InvalidStatusRequestHandler,
50
NoRangeRequestHandler,
51
SingleRangeRequestHandler,
52
TestCaseWithRedirectedWebserver,
53
TestCaseWithTwoWebservers,
54
TestCaseWithWebserver,
57
from bzrlib.transport import (
58
do_catching_redirections,
62
from bzrlib.transport.http import (
67
from bzrlib.transport.http._urllib import HttpTransport_urllib
70
class FakeManager(object):
75
def add_password(self, realm, host, username, password):
76
self.credentials.append([realm, host, username, password])
79
class RecordingServer(object):
80
"""A fake HTTP server.
82
It records the bytes sent to it, and replies with a 200.
85
def __init__(self, expect_body_tail=None):
88
:type expect_body_tail: str
89
:param expect_body_tail: a reply won't be sent until this string is
92
self._expect_body_tail = expect_body_tail
95
self.received_bytes = ''
98
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
99
self._sock.bind(('127.0.0.1', 0))
100
self.host, self.port = self._sock.getsockname()
101
self._ready = threading.Event()
102
self._thread = threading.Thread(target=self._accept_read_and_reply)
103
self._thread.setDaemon(True)
107
def _accept_read_and_reply(self):
110
self._sock.settimeout(5)
112
conn, address = self._sock.accept()
113
# On win32, the accepted connection will be non-blocking to start
114
# with because we're using settimeout.
115
conn.setblocking(True)
116
while not self.received_bytes.endswith(self._expect_body_tail):
117
self.received_bytes += conn.recv(4096)
118
conn.sendall('HTTP/1.1 200 OK\r\n')
119
except socket.timeout:
120
# Make sure the client isn't stuck waiting for us to e.g. accept.
123
# The client may have already closed the socket.
130
# We might have already closed it. We don't care.
136
class TestWithTransport_pycurl(object):
137
"""Test case to inherit from if pycurl is present"""
139
def _get_pycurl_maybe(self):
141
from bzrlib.transport.http._pycurl import PyCurlTransport
142
return PyCurlTransport
143
except errors.DependencyNotPresent:
144
raise TestSkipped('pycurl not present')
146
_transport = property(_get_pycurl_maybe)
149
class TestHttpUrls(TestCase):
151
# TODO: This should be moved to authorization tests once they
154
def test_url_parsing(self):
156
url = extract_auth('http://example.com', f)
157
self.assertEquals('http://example.com', url)
158
self.assertEquals(0, len(f.credentials))
159
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
160
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
161
self.assertEquals(1, len(f.credentials))
162
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
166
class TestHttpTransportUrls(object):
167
"""Test the http urls.
169
This MUST be used by daughter classes that also inherit from
172
We can't inherit directly from TestCase or the
173
test framework will try to create an instance which cannot
174
run, its implementation being incomplete.
177
def test_abs_url(self):
178
"""Construction of absolute http URLs"""
179
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
180
eq = self.assertEqualDiff
181
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
182
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
183
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
184
eq(t.abspath('.bzr/1//2/./3'),
185
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
187
def test_invalid_http_urls(self):
188
"""Trap invalid construction of urls"""
189
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
190
self.assertRaises(ValueError, t.abspath, '.bzr/')
191
t = self._transport('http://http://bazaar-vcs.org/bzr/bzr.dev/')
192
self.assertRaises((errors.InvalidURL, errors.ConnectionError),
195
def test_http_root_urls(self):
196
"""Construction of URLs from server root"""
197
t = self._transport('http://bzr.ozlabs.org/')
198
eq = self.assertEqualDiff
199
eq(t.abspath('.bzr/tree-version'),
200
'http://bzr.ozlabs.org/.bzr/tree-version')
202
def test_http_impl_urls(self):
203
"""There are servers which ask for particular clients to connect"""
204
server = self._server()
207
url = server.get_url()
208
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
213
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
214
"""Test http urls with urllib"""
216
_transport = HttpTransport_urllib
217
_server = HttpServer_urllib
218
_qualified_prefix = 'http+urllib'
221
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
223
"""Test http urls with pycurl"""
225
_server = HttpServer_PyCurl
226
_qualified_prefix = 'http+pycurl'
228
# TODO: This should really be moved into another pycurl
229
# specific test. When https tests will be implemented, take
230
# this one into account.
231
def test_pycurl_without_https_support(self):
232
"""Test that pycurl without SSL do not fail with a traceback.
234
For the purpose of the test, we force pycurl to ignore
235
https by supplying a fake version_info that do not
241
raise TestSkipped('pycurl not present')
242
# Now that we have pycurl imported, we can fake its version_info
243
# This was taken from a windows pycurl without SSL
245
pycurl.version_info = lambda : (2,
253
('ftp', 'gopher', 'telnet',
254
'dict', 'ldap', 'http', 'file'),
258
self.assertRaises(errors.DependencyNotPresent, self._transport,
259
'https://launchpad.net')
261
class TestHttpConnections(object):
262
"""Test the http connections.
264
This MUST be used by daughter classes that also inherit from
265
TestCaseWithWebserver.
267
We can't inherit directly from TestCaseWithWebserver or the
268
test framework will try to create an instance which cannot
269
run, its implementation being incomplete.
273
TestCaseWithWebserver.setUp(self)
274
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
275
transport=self.get_transport())
277
def test_http_has(self):
278
server = self.get_readonly_server()
279
t = self._transport(server.get_url())
280
self.assertEqual(t.has('foo/bar'), True)
281
self.assertEqual(len(server.logs), 1)
282
self.assertContainsRe(server.logs[0],
283
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
285
def test_http_has_not_found(self):
286
server = self.get_readonly_server()
287
t = self._transport(server.get_url())
288
self.assertEqual(t.has('not-found'), False)
289
self.assertContainsRe(server.logs[1],
290
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
292
def test_http_get(self):
293
server = self.get_readonly_server()
294
t = self._transport(server.get_url())
295
fp = t.get('foo/bar')
296
self.assertEqualDiff(
298
'contents of foo/bar\n')
299
self.assertEqual(len(server.logs), 1)
300
self.assertTrue(server.logs[0].find(
301
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
302
% bzrlib.__version__) > -1)
304
def test_get_smart_medium(self):
305
# For HTTP, get_smart_medium should return the transport object.
306
server = self.get_readonly_server()
307
http_transport = self._transport(server.get_url())
308
medium = http_transport.get_smart_medium()
309
self.assertIs(medium, http_transport)
311
def test_has_on_bogus_host(self):
312
# Get a free address and don't 'accept' on it, so that we
313
# can be sure there is no http handler there, but set a
314
# reasonable timeout to not slow down tests too much.
315
default_timeout = socket.getdefaulttimeout()
317
socket.setdefaulttimeout(2)
319
s.bind(('localhost', 0))
320
t = self._transport('http://%s:%s/' % s.getsockname())
321
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
323
socket.setdefaulttimeout(default_timeout)
326
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
327
"""Test http connections with urllib"""
329
_transport = HttpTransport_urllib
333
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
335
TestCaseWithWebserver):
336
"""Test http connections with pycurl"""
339
class TestHttpTransportRegistration(TestCase):
340
"""Test registrations of various http implementations"""
342
def test_http_registered(self):
343
# urlllib should always be present
344
t = get_transport('http+urllib://bzr.google.com/')
345
self.assertIsInstance(t, Transport)
346
self.assertIsInstance(t, HttpTransport_urllib)
349
class TestOffsets(TestCase):
350
"""Test offsets_to_ranges method"""
352
def test_offsets_to_ranges_simple(self):
353
to_range = HttpTransportBase.offsets_to_ranges
354
ranges = to_range([(10, 1)])
355
self.assertEqual([[10, 10]], ranges)
357
ranges = to_range([(0, 1), (1, 1)])
358
self.assertEqual([[0, 1]], ranges)
360
ranges = to_range([(1, 1), (0, 1)])
361
self.assertEqual([[0, 1]], ranges)
363
def test_offset_to_ranges_overlapped(self):
364
to_range = HttpTransportBase.offsets_to_ranges
366
ranges = to_range([(10, 1), (20, 2), (22, 5)])
367
self.assertEqual([[10, 10], [20, 26]], ranges)
369
ranges = to_range([(10, 1), (11, 2), (22, 5)])
370
self.assertEqual([[10, 12], [22, 26]], ranges)
373
class TestPost(object):
375
def _test_post_body_is_received(self, scheme):
376
server = RecordingServer(expect_body_tail='end-of-body')
378
self.addCleanup(server.tearDown)
379
url = '%s://%s:%s/' % (scheme, server.host, server.port)
381
http_transport = get_transport(url)
382
except errors.UnsupportedProtocol:
383
raise TestSkipped('%s not available' % scheme)
384
code, response = http_transport._post('abc def end-of-body')
386
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
387
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
388
# The transport should not be assuming that the server can accept
389
# chunked encoding the first time it connects, because HTTP/1.1, so we
390
# check for the literal string.
392
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
395
class TestPost_urllib(TestCase, TestPost):
396
"""TestPost for urllib implementation"""
398
_transport = HttpTransport_urllib
400
def test_post_body_is_received_urllib(self):
401
self._test_post_body_is_received('http+urllib')
404
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
405
"""TestPost for pycurl implementation"""
407
def test_post_body_is_received_pycurl(self):
408
self._test_post_body_is_received('http+pycurl')
411
class TestRangeHeader(TestCase):
412
"""Test range_header method"""
414
def check_header(self, value, ranges=[], tail=0):
415
range_header = HttpTransportBase.range_header
416
self.assertEqual(value, range_header(ranges, tail))
418
def test_range_header_single(self):
419
self.check_header('0-9', ranges=[[0,9]])
420
self.check_header('100-109', ranges=[[100,109]])
422
def test_range_header_tail(self):
423
self.check_header('-10', tail=10)
424
self.check_header('-50', tail=50)
426
def test_range_header_multi(self):
427
self.check_header('0-9,100-200,300-5000',
428
ranges=[(0,9), (100, 200), (300,5000)])
430
def test_range_header_mixed(self):
431
self.check_header('0-9,300-5000,-50',
432
ranges=[(0,9), (300,5000)],
436
class TestWallServer(object):
437
"""Tests exceptions during the connection phase"""
439
def create_transport_readonly_server(self):
440
return HttpServer(WallRequestHandler)
442
def test_http_has(self):
443
server = self.get_readonly_server()
444
t = self._transport(server.get_url())
445
# Unfortunately httplib (see HTTPResponse._read_status
446
# for details) make no distinction between a closed
447
# socket and badly formatted status line, so we can't
448
# just test for ConnectionError, we have to test
449
# InvalidHttpResponse too.
450
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
453
def test_http_get(self):
454
server = self.get_readonly_server()
455
t = self._transport(server.get_url())
456
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
460
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
461
"""Tests "wall" server for urllib implementation"""
463
_transport = HttpTransport_urllib
466
class TestWallServer_pycurl(TestWithTransport_pycurl,
468
TestCaseWithWebserver):
469
"""Tests "wall" server for pycurl implementation"""
472
class TestBadStatusServer(object):
473
"""Tests bad status from server."""
475
def create_transport_readonly_server(self):
476
return HttpServer(BadStatusRequestHandler)
478
def test_http_has(self):
479
server = self.get_readonly_server()
480
t = self._transport(server.get_url())
481
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
483
def test_http_get(self):
484
server = self.get_readonly_server()
485
t = self._transport(server.get_url())
486
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
489
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
490
"""Tests bad status server for urllib implementation"""
492
_transport = HttpTransport_urllib
495
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
497
TestCaseWithWebserver):
498
"""Tests bad status server for pycurl implementation"""
501
class TestInvalidStatusServer(TestBadStatusServer):
502
"""Tests invalid status from server.
504
Both implementations raises the same error as for a bad status.
507
def create_transport_readonly_server(self):
508
return HttpServer(InvalidStatusRequestHandler)
511
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
512
TestCaseWithWebserver):
513
"""Tests invalid status server for urllib implementation"""
515
_transport = HttpTransport_urllib
518
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
519
TestInvalidStatusServer,
520
TestCaseWithWebserver):
521
"""Tests invalid status server for pycurl implementation"""
524
class TestBadProtocolServer(object):
525
"""Tests bad protocol from server."""
527
def create_transport_readonly_server(self):
528
return HttpServer(BadProtocolRequestHandler)
530
def test_http_has(self):
531
server = self.get_readonly_server()
532
t = self._transport(server.get_url())
533
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
535
def test_http_get(self):
536
server = self.get_readonly_server()
537
t = self._transport(server.get_url())
538
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
541
class TestBadProtocolServer_urllib(TestBadProtocolServer,
542
TestCaseWithWebserver):
543
"""Tests bad protocol server for urllib implementation"""
545
_transport = HttpTransport_urllib
547
# curl don't check the protocol version
548
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
549
# TestBadProtocolServer,
550
# TestCaseWithWebserver):
551
# """Tests bad protocol server for pycurl implementation"""
554
class TestForbiddenServer(object):
555
"""Tests forbidden server"""
557
def create_transport_readonly_server(self):
558
return HttpServer(ForbiddenRequestHandler)
560
def test_http_has(self):
561
server = self.get_readonly_server()
562
t = self._transport(server.get_url())
563
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
565
def test_http_get(self):
566
server = self.get_readonly_server()
567
t = self._transport(server.get_url())
568
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
571
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
572
"""Tests forbidden server for urllib implementation"""
574
_transport = HttpTransport_urllib
577
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
579
TestCaseWithWebserver):
580
"""Tests forbidden server for pycurl implementation"""
583
class TestRecordingServer(TestCase):
585
def test_create(self):
586
server = RecordingServer(expect_body_tail=None)
587
self.assertEqual('', server.received_bytes)
588
self.assertEqual(None, server.host)
589
self.assertEqual(None, server.port)
591
def test_setUp_and_tearDown(self):
592
server = RecordingServer(expect_body_tail=None)
595
self.assertNotEqual(None, server.host)
596
self.assertNotEqual(None, server.port)
599
self.assertEqual(None, server.host)
600
self.assertEqual(None, server.port)
602
def test_send_receive_bytes(self):
603
server = RecordingServer(expect_body_tail='c')
605
self.addCleanup(server.tearDown)
606
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
607
sock.connect((server.host, server.port))
609
self.assertEqual('HTTP/1.1 200 OK\r\n',
610
osutils.recv_all(sock, 4096))
611
self.assertEqual('abc', server.received_bytes)
614
class TestRangeRequestServer(object):
615
"""Tests readv requests against server.
617
This MUST be used by daughter classes that also inherit from
618
TestCaseWithWebserver.
620
We can't inherit directly from TestCaseWithWebserver or the
621
test framework will try to create an instance which cannot
622
run, its implementation being incomplete.
626
TestCaseWithWebserver.setUp(self)
627
self.build_tree_contents([('a', '0123456789')],)
629
def test_readv(self):
630
server = self.get_readonly_server()
631
t = self._transport(server.get_url())
632
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
633
self.assertEqual(l[0], (0, '0'))
634
self.assertEqual(l[1], (1, '1'))
635
self.assertEqual(l[2], (3, '34'))
636
self.assertEqual(l[3], (9, '9'))
638
def test_readv_out_of_order(self):
639
server = self.get_readonly_server()
640
t = self._transport(server.get_url())
641
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
642
self.assertEqual(l[0], (1, '1'))
643
self.assertEqual(l[1], (9, '9'))
644
self.assertEqual(l[2], (0, '0'))
645
self.assertEqual(l[3], (3, '34'))
647
def test_readv_invalid_ranges(self):
648
server = self.get_readonly_server()
649
t = self._transport(server.get_url())
651
# This is intentionally reading off the end of the file
652
# since we are sure that it cannot get there
653
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
654
t.readv, 'a', [(1,1), (8,10)])
656
# This is trying to seek past the end of the file, it should
657
# also raise a special error
658
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
659
t.readv, 'a', [(12,2)])
662
class TestSingleRangeRequestServer(TestRangeRequestServer):
663
"""Test readv against a server which accept only single range requests"""
665
def create_transport_readonly_server(self):
666
return HttpServer(SingleRangeRequestHandler)
669
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
670
TestCaseWithWebserver):
671
"""Tests single range requests accepting server for urllib implementation"""
673
_transport = HttpTransport_urllib
676
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
677
TestSingleRangeRequestServer,
678
TestCaseWithWebserver):
679
"""Tests single range requests accepting server for pycurl implementation"""
682
class TestNoRangeRequestServer(TestRangeRequestServer):
683
"""Test readv against a server which do not accept range requests"""
685
def create_transport_readonly_server(self):
686
return HttpServer(NoRangeRequestHandler)
689
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
690
TestCaseWithWebserver):
691
"""Tests range requests refusing server for urllib implementation"""
693
_transport = HttpTransport_urllib
696
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
697
TestNoRangeRequestServer,
698
TestCaseWithWebserver):
699
"""Tests range requests refusing server for pycurl implementation"""
702
class TestHttpProxyWhiteBox(TestCase):
703
"""Whitebox test proxy http authorization.
705
These tests concern urllib implementation only.
715
def _set_and_capture_env_var(self, name, new_value):
716
"""Set an environment variable, and reset it when finished."""
717
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
719
def _install_env(self, env):
720
for name, value in env.iteritems():
721
self._set_and_capture_env_var(name, value)
723
def _restore_env(self):
724
for name, value in self._old_env.iteritems():
725
osutils.set_or_unset_env(name, value)
727
def _proxied_request(self):
728
from bzrlib.transport.http._urllib2_wrappers import (
733
handler = ProxyHandler()
734
request = Request('GET','http://baz/buzzle')
735
handler.set_proxy(request, 'http')
738
def test_empty_user(self):
739
self._install_env({'http_proxy': 'http://bar.com'})
740
request = self._proxied_request()
741
self.assertFalse(request.headers.has_key('Proxy-authorization'))
743
def test_empty_pass(self):
744
self._install_env({'http_proxy': 'http://joe@bar.com'})
745
request = self._proxied_request()
746
self.assertEqual('Basic ' + 'joe:'.encode('base64').strip(),
747
request.headers['Proxy-authorization'])
748
def test_user_pass(self):
749
self._install_env({'http_proxy': 'http://joe:foo@bar.com'})
750
request = self._proxied_request()
751
self.assertEqual('Basic ' + 'joe:foo'.encode('base64').strip(),
752
request.headers['Proxy-authorization'])
754
def test_invalid_proxy(self):
755
"""A proxy env variable without scheme"""
756
self._install_env({'http_proxy': 'host:1234'})
757
self.assertRaises(errors.InvalidURL, self._proxied_request)
760
class TestProxyHttpServer(object):
761
"""Tests proxy server.
763
This MUST be used by daughter classes that also inherit from
764
TestCaseWithTwoWebservers.
766
We can't inherit directly from TestCaseWithTwoWebservers or
767
the test framework will try to create an instance which
768
cannot run, its implementation being incomplete.
770
Be aware that we do not setup a real proxy here. Instead, we
771
check that the *connection* goes through the proxy by serving
772
different content (the faked proxy server append '-proxied'
776
# FIXME: We don't have an https server available, so we don't
777
# test https connections.
779
# FIXME: Once the test suite is better fitted to test
780
# authorization schemes, test proxy authorizations too (see
784
TestCaseWithTwoWebservers.setUp(self)
785
self.build_tree_contents([('foo', 'contents of foo\n'),
786
('foo-proxied', 'proxied contents of foo\n')])
787
# Let's setup some attributes for tests
788
self.server = self.get_readonly_server()
789
# FIXME: We should not rely on 'localhost' being the hostname
790
self.proxy_address = 'localhost:%d' % self.server.port
791
self.no_proxy_host = self.proxy_address
792
# The secondary server is the proxy
793
self.proxy = self.get_secondary_server()
794
self.proxy_url = self.proxy.get_url()
797
def create_transport_secondary_server(self):
798
"""Creates an http server that will serve files with
799
'-proxied' appended to their names.
801
return HttpServer(FakeProxyRequestHandler)
803
def _set_and_capture_env_var(self, name, new_value):
804
"""Set an environment variable, and reset it when finished."""
805
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
807
def _install_env(self, env):
808
for name, value in env.iteritems():
809
self._set_and_capture_env_var(name, value)
811
def _restore_env(self):
812
for name, value in self._old_env.iteritems():
813
osutils.set_or_unset_env(name, value)
815
def proxied_in_env(self, env):
816
self._install_env(env)
817
url = self.server.get_url()
818
t = self._transport(url)
820
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
824
def not_proxied_in_env(self, env):
825
self._install_env(env)
826
url = self.server.get_url()
827
t = self._transport(url)
829
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
833
def test_http_proxy(self):
834
self.proxied_in_env({'http_proxy': self.proxy_url})
836
def test_HTTP_PROXY(self):
837
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
839
def test_all_proxy(self):
840
self.proxied_in_env({'all_proxy': self.proxy_url})
842
def test_ALL_PROXY(self):
843
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
845
def test_http_proxy_with_no_proxy(self):
846
self.not_proxied_in_env({'http_proxy': self.proxy_url,
847
'no_proxy': self.no_proxy_host})
849
def test_HTTP_PROXY_with_NO_PROXY(self):
850
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
851
'NO_PROXY': self.no_proxy_host})
853
def test_all_proxy_with_no_proxy(self):
854
self.not_proxied_in_env({'all_proxy': self.proxy_url,
855
'no_proxy': self.no_proxy_host})
857
def test_ALL_PROXY_with_NO_PROXY(self):
858
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
859
'NO_PROXY': self.no_proxy_host})
861
def test_http_proxy_without_scheme(self):
862
self.assertRaises(errors.InvalidURL,
864
{'http_proxy': self.proxy_address})
867
class TestProxyHttpServer_urllib(TestProxyHttpServer,
868
TestCaseWithTwoWebservers):
869
"""Tests proxy server for urllib implementation"""
871
_transport = HttpTransport_urllib
874
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
876
TestCaseWithTwoWebservers):
877
"""Tests proxy server for pycurl implementation"""
880
TestProxyHttpServer.setUp(self)
881
# Oh my ! pycurl does not check for the port as part of
882
# no_proxy :-( So we just test the host part
883
self.no_proxy_host = 'localhost'
885
def test_HTTP_PROXY(self):
886
# pycurl do not check HTTP_PROXY for security reasons
887
# (for use in a CGI context that we do not care
888
# about. Should we ?)
891
def test_HTTP_PROXY_with_NO_PROXY(self):
894
def test_http_proxy_without_scheme(self):
895
# pycurl *ignores* invalid proxy env variables. If that
896
# ever change in the future, this test will fail
897
# indicating that pycurl do not ignore anymore such
899
self.not_proxied_in_env({'http_proxy': self.proxy_address})
902
class TestRanges(object):
903
"""Test the Range header in GET methods..
905
This MUST be used by daughter classes that also inherit from
906
TestCaseWithWebserver.
908
We can't inherit directly from TestCaseWithWebserver or the
909
test framework will try to create an instance which cannot
910
run, its implementation being incomplete.
914
TestCaseWithWebserver.setUp(self)
915
self.build_tree_contents([('a', '0123456789')],)
916
server = self.get_readonly_server()
917
self.transport = self._transport(server.get_url())
919
def _file_contents(self, relpath, ranges, tail_amount=0):
920
code, data = self.transport._get(relpath, ranges)
921
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
922
for start, end in ranges:
924
yield data.read(end - start + 1)
926
def _file_tail(self, relpath, tail_amount):
927
code, data = self.transport._get(relpath, [], tail_amount)
928
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
929
data.seek(-tail_amount + 1, 2)
930
return data.read(tail_amount)
932
def test_range_header(self):
934
map(self.assertEqual,['0', '234'],
935
list(self._file_contents('a', [(0,0), (2,4)])),)
937
self.assertEqual('789', self._file_tail('a', 3))
938
# Syntactically invalid range
939
self.assertRaises(errors.InvalidRange,
940
self.transport._get, 'a', [(4, 3)])
941
# Semantically invalid range
942
self.assertRaises(errors.InvalidRange,
943
self.transport._get, 'a', [(42, 128)])
946
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
947
"""Test the Range header in GET methods for urllib implementation"""
949
_transport = HttpTransport_urllib
952
class TestRanges_pycurl(TestWithTransport_pycurl,
954
TestCaseWithWebserver):
955
"""Test the Range header in GET methods for pycurl implementation"""
958
class TestHTTPRedirections(object):
959
"""Test redirection between http servers.
961
This MUST be used by daughter classes that also inherit from
962
TestCaseWithRedirectedWebserver.
964
We can't inherit directly from TestCaseWithTwoWebservers or the
965
test framework will try to create an instance which cannot
966
run, its implementation being incomplete.
969
def create_transport_secondary_server(self):
970
"""Create the secondary server redirecting to the primary server"""
971
new = self.get_readonly_server()
973
redirecting = HTTPServerRedirecting()
974
redirecting.redirect_to(new.host, new.port)
978
super(TestHTTPRedirections, self).setUp()
979
self.build_tree_contents([('a', '0123456789'),
981
'# Bazaar revision bundle v0.9\n#\n')
984
self.old_transport = self._transport(self.old_server.get_url())
986
def test_redirected(self):
987
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
988
t = self._transport(self.new_server.get_url())
989
self.assertEqual('0123456789', t.get('a').read())
991
def test_read_redirected_bundle_from_url(self):
992
from bzrlib.bundle import read_bundle_from_url
993
url = self.old_transport.abspath('bundle')
994
bundle = read_bundle_from_url(url)
995
# If read_bundle_from_url was successful we get an empty bundle
996
self.assertEqual([], bundle.revisions)
999
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1000
TestCaseWithRedirectedWebserver):
1001
"""Tests redirections for urllib implementation"""
1003
_transport = HttpTransport_urllib
1007
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1008
TestHTTPRedirections,
1009
TestCaseWithRedirectedWebserver):
1010
"""Tests redirections for pycurl implementation"""
1013
class RedirectedRequest(_urllib2_wrappers.Request):
1014
"""Request following redirections"""
1016
init_orig = _urllib2_wrappers.Request.__init__
1018
def __init__(self, method, url, *args, **kwargs):
1019
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1020
self.follow_redirections = True
1023
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1024
"""Test redirections provided by urllib.
1026
http implementations do not redirect silently anymore (they
1027
do not redirect at all in fact). The mechanism is still in
1028
place at the _urllib2_wrappers.Request level and these tests
1031
For the pycurl implementation
1032
the redirection have been deleted as we may deprecate pycurl
1033
and I have no place to keep a working implementation.
1037
_transport = HttpTransport_urllib
1040
super(TestHTTPSilentRedirections_urllib, self).setUp()
1041
self.setup_redirected_request()
1042
self.addCleanup(self.cleanup_redirected_request)
1043
self.build_tree_contents([('a','a'),
1045
('1/a', 'redirected once'),
1047
('2/a', 'redirected twice'),
1049
('3/a', 'redirected thrice'),
1051
('4/a', 'redirected 4 times'),
1053
('5/a', 'redirected 5 times'),
1056
self.old_transport = self._transport(self.old_server.get_url())
1058
def setup_redirected_request(self):
1059
self.original_class = _urllib2_wrappers.Request
1060
_urllib2_wrappers.Request = RedirectedRequest
1062
def cleanup_redirected_request(self):
1063
_urllib2_wrappers.Request = self.original_class
1065
def create_transport_secondary_server(self):
1066
"""Create the secondary server, redirections are defined in the tests"""
1067
return HTTPServerRedirecting()
1069
def test_one_redirection(self):
1070
t = self.old_transport
1072
req = RedirectedRequest('GET', t.abspath('a'))
1073
req.follow_redirections = True
1074
new_prefix = 'http://%s:%s' % (self.new_server.host,
1075
self.new_server.port)
1076
self.old_server.redirections = \
1077
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1078
self.assertEquals('redirected once',t._perform(req).read())
1080
def test_five_redirections(self):
1081
t = self.old_transport
1083
req = RedirectedRequest('GET', t.abspath('a'))
1084
req.follow_redirections = True
1085
old_prefix = 'http://%s:%s' % (self.old_server.host,
1086
self.old_server.port)
1087
new_prefix = 'http://%s:%s' % (self.new_server.host,
1088
self.new_server.port)
1089
self.old_server.redirections = \
1090
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1091
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1092
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1093
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1094
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1096
self.assertEquals('redirected 5 times',t._perform(req).read())
1099
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1100
"""Test transport.do_catching_redirections.
1102
We arbitrarily choose to use urllib transports
1105
_transport = HttpTransport_urllib
1108
super(TestDoCatchRedirections, self).setUp()
1109
self.build_tree_contents([('a', '0123456789'),],)
1111
self.old_transport = self._transport(self.old_server.get_url())
1113
def get_a(self, transport):
1114
return transport.get('a')
1116
def test_no_redirection(self):
1117
t = self._transport(self.new_server.get_url())
1119
# We use None for redirected so that we fail if redirected
1120
self.assertEquals('0123456789',
1121
do_catching_redirections(self.get_a, t, None).read())
1123
def test_one_redirection(self):
1124
self.redirections = 0
1126
def redirected(transport, exception, redirection_notice):
1127
self.redirections += 1
1128
dir, file = urlutils.split(exception.target)
1129
return self._transport(dir)
1131
self.assertEquals('0123456789',
1132
do_catching_redirections(self.get_a,
1136
self.assertEquals(1, self.redirections)
1138
def test_redirection_loop(self):
1140
def redirected(transport, exception, redirection_notice):
1141
# By using the redirected url as a base dir for the
1142
# *old* transport, we create a loop: a => a/a =>
1144
return self.old_transport.clone(exception.target)
1146
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1147
self.get_a, self.old_transport, redirected)
added
removed
bzrlib/tests/test_http.py
