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 ?
23
from cStringIO import StringIO
36
from bzrlib.tests import (
42
from bzrlib.tests.HttpServer import (
47
from bzrlib.tests.HTTPTestUtil import (
48
BadProtocolRequestHandler,
49
BadStatusRequestHandler,
50
ForbiddenRequestHandler,
53
HTTPServerRedirecting,
54
InvalidStatusRequestHandler,
55
NoRangeRequestHandler,
57
ProxyDigestAuthServer,
59
SingleRangeRequestHandler,
60
TestCaseWithRedirectedWebserver,
61
TestCaseWithTwoWebservers,
62
TestCaseWithWebserver,
65
from bzrlib.transport import (
66
do_catching_redirections,
70
from bzrlib.transport.http import (
75
from bzrlib.transport.http._urllib import HttpTransport_urllib
76
from bzrlib.transport.http._urllib2_wrappers import (
83
class FakeManager(object):
88
def add_password(self, realm, host, username, password):
89
self.credentials.append([realm, host, username, password])
92
class RecordingServer(object):
93
"""A fake HTTP server.
95
It records the bytes sent to it, and replies with a 200.
98
def __init__(self, expect_body_tail=None):
101
:type expect_body_tail: str
102
:param expect_body_tail: a reply won't be sent until this string is
105
self._expect_body_tail = expect_body_tail
108
self.received_bytes = ''
111
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
112
self._sock.bind(('127.0.0.1', 0))
113
self.host, self.port = self._sock.getsockname()
114
self._ready = threading.Event()
115
self._thread = threading.Thread(target=self._accept_read_and_reply)
116
self._thread.setDaemon(True)
120
def _accept_read_and_reply(self):
123
self._sock.settimeout(5)
125
conn, address = self._sock.accept()
126
# On win32, the accepted connection will be non-blocking to start
127
# with because we're using settimeout.
128
conn.setblocking(True)
129
while not self.received_bytes.endswith(self._expect_body_tail):
130
self.received_bytes += conn.recv(4096)
131
conn.sendall('HTTP/1.1 200 OK\r\n')
132
except socket.timeout:
133
# Make sure the client isn't stuck waiting for us to e.g. accept.
136
# The client may have already closed the socket.
143
# We might have already closed it. We don't care.
149
class TestWithTransport_pycurl(object):
150
"""Test case to inherit from if pycurl is present"""
152
def _get_pycurl_maybe(self):
154
from bzrlib.transport.http._pycurl import PyCurlTransport
155
return PyCurlTransport
156
except errors.DependencyNotPresent:
157
raise TestSkipped('pycurl not present')
159
_transport = property(_get_pycurl_maybe)
162
class TestHttpUrls(TestCase):
164
# TODO: This should be moved to authorization tests once they
167
def test_url_parsing(self):
169
url = extract_auth('http://example.com', f)
170
self.assertEquals('http://example.com', url)
171
self.assertEquals(0, len(f.credentials))
172
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
173
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
174
self.assertEquals(1, len(f.credentials))
175
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
179
class TestHttpTransportUrls(object):
180
"""Test the http urls.
182
This MUST be used by daughter classes that also inherit from
185
We can't inherit directly from TestCase or the
186
test framework will try to create an instance which cannot
187
run, its implementation being incomplete.
190
def test_abs_url(self):
191
"""Construction of absolute http URLs"""
192
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
193
eq = self.assertEqualDiff
194
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
195
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
196
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
197
eq(t.abspath('.bzr/1//2/./3'),
198
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
200
def test_invalid_http_urls(self):
201
"""Trap invalid construction of urls"""
202
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
203
self.assertRaises(ValueError, t.abspath, '.bzr/')
204
t = self._transport('http://http://bazaar-vcs.org/bzr/bzr.dev/')
205
self.assertRaises((errors.InvalidURL, errors.ConnectionError),
208
def test_http_root_urls(self):
209
"""Construction of URLs from server root"""
210
t = self._transport('http://bzr.ozlabs.org/')
211
eq = self.assertEqualDiff
212
eq(t.abspath('.bzr/tree-version'),
213
'http://bzr.ozlabs.org/.bzr/tree-version')
215
def test_http_impl_urls(self):
216
"""There are servers which ask for particular clients to connect"""
217
server = self._server()
220
url = server.get_url()
221
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
226
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
227
"""Test http urls with urllib"""
229
_transport = HttpTransport_urllib
230
_server = HttpServer_urllib
231
_qualified_prefix = 'http+urllib'
234
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
236
"""Test http urls with pycurl"""
238
_server = HttpServer_PyCurl
239
_qualified_prefix = 'http+pycurl'
241
# TODO: This should really be moved into another pycurl
242
# specific test. When https tests will be implemented, take
243
# this one into account.
244
def test_pycurl_without_https_support(self):
245
"""Test that pycurl without SSL do not fail with a traceback.
247
For the purpose of the test, we force pycurl to ignore
248
https by supplying a fake version_info that do not
254
raise TestSkipped('pycurl not present')
255
# Now that we have pycurl imported, we can fake its version_info
256
# This was taken from a windows pycurl without SSL
258
pycurl.version_info = lambda : (2,
266
('ftp', 'gopher', 'telnet',
267
'dict', 'ldap', 'http', 'file'),
271
self.assertRaises(errors.DependencyNotPresent, self._transport,
272
'https://launchpad.net')
274
class TestHttpConnections(object):
275
"""Test the http connections.
277
This MUST be used by daughter classes that also inherit from
278
TestCaseWithWebserver.
280
We can't inherit directly from TestCaseWithWebserver or the
281
test framework will try to create an instance which cannot
282
run, its implementation being incomplete.
286
TestCaseWithWebserver.setUp(self)
287
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
288
transport=self.get_transport())
290
def test_http_has(self):
291
server = self.get_readonly_server()
292
t = self._transport(server.get_url())
293
self.assertEqual(t.has('foo/bar'), True)
294
self.assertEqual(len(server.logs), 1)
295
self.assertContainsRe(server.logs[0],
296
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
298
def test_http_has_not_found(self):
299
server = self.get_readonly_server()
300
t = self._transport(server.get_url())
301
self.assertEqual(t.has('not-found'), False)
302
self.assertContainsRe(server.logs[1],
303
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
305
def test_http_get(self):
306
server = self.get_readonly_server()
307
t = self._transport(server.get_url())
308
fp = t.get('foo/bar')
309
self.assertEqualDiff(
311
'contents of foo/bar\n')
312
self.assertEqual(len(server.logs), 1)
313
self.assertTrue(server.logs[0].find(
314
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
315
% bzrlib.__version__) > -1)
317
def test_get_smart_medium(self):
318
# For HTTP, get_smart_medium should return the transport object.
319
server = self.get_readonly_server()
320
http_transport = self._transport(server.get_url())
321
medium = http_transport.get_smart_medium()
322
self.assertIs(medium, http_transport)
324
def test_has_on_bogus_host(self):
325
# Get a free address and don't 'accept' on it, so that we
326
# can be sure there is no http handler there, but set a
327
# reasonable timeout to not slow down tests too much.
328
default_timeout = socket.getdefaulttimeout()
330
socket.setdefaulttimeout(2)
332
s.bind(('localhost', 0))
333
t = self._transport('http://%s:%s/' % s.getsockname())
334
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
336
socket.setdefaulttimeout(default_timeout)
339
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
340
"""Test http connections with urllib"""
342
_transport = HttpTransport_urllib
346
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
348
TestCaseWithWebserver):
349
"""Test http connections with pycurl"""
352
class TestHttpTransportRegistration(TestCase):
353
"""Test registrations of various http implementations"""
355
def test_http_registered(self):
356
# urlllib should always be present
357
t = get_transport('http+urllib://bzr.google.com/')
358
self.assertIsInstance(t, Transport)
359
self.assertIsInstance(t, HttpTransport_urllib)
362
class TestOffsets(TestCase):
363
"""Test offsets_to_ranges method"""
365
def test_offsets_to_ranges_simple(self):
366
to_range = HttpTransportBase.offsets_to_ranges
367
ranges = to_range([(10, 1)])
368
self.assertEqual([[10, 10]], ranges)
370
ranges = to_range([(0, 1), (1, 1)])
371
self.assertEqual([[0, 1]], ranges)
373
ranges = to_range([(1, 1), (0, 1)])
374
self.assertEqual([[0, 1]], ranges)
376
def test_offset_to_ranges_overlapped(self):
377
to_range = HttpTransportBase.offsets_to_ranges
379
ranges = to_range([(10, 1), (20, 2), (22, 5)])
380
self.assertEqual([[10, 10], [20, 26]], ranges)
382
ranges = to_range([(10, 1), (11, 2), (22, 5)])
383
self.assertEqual([[10, 12], [22, 26]], ranges)
386
class TestPost(object):
388
def _test_post_body_is_received(self, scheme):
389
server = RecordingServer(expect_body_tail='end-of-body')
391
self.addCleanup(server.tearDown)
392
url = '%s://%s:%s/' % (scheme, server.host, server.port)
394
http_transport = get_transport(url)
395
except errors.UnsupportedProtocol:
396
raise TestSkipped('%s not available' % scheme)
397
code, response = http_transport._post('abc def end-of-body')
399
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
400
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
401
# The transport should not be assuming that the server can accept
402
# chunked encoding the first time it connects, because HTTP/1.1, so we
403
# check for the literal string.
405
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
408
class TestPost_urllib(TestCase, TestPost):
409
"""TestPost for urllib implementation"""
411
_transport = HttpTransport_urllib
413
def test_post_body_is_received_urllib(self):
414
self._test_post_body_is_received('http+urllib')
417
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
418
"""TestPost for pycurl implementation"""
420
def test_post_body_is_received_pycurl(self):
421
self._test_post_body_is_received('http+pycurl')
424
class TestRangeHeader(TestCase):
425
"""Test range_header method"""
427
def check_header(self, value, ranges=[], tail=0):
428
range_header = HttpTransportBase.range_header
429
self.assertEqual(value, range_header(ranges, tail))
431
def test_range_header_single(self):
432
self.check_header('0-9', ranges=[[0,9]])
433
self.check_header('100-109', ranges=[[100,109]])
435
def test_range_header_tail(self):
436
self.check_header('-10', tail=10)
437
self.check_header('-50', tail=50)
439
def test_range_header_multi(self):
440
self.check_header('0-9,100-200,300-5000',
441
ranges=[(0,9), (100, 200), (300,5000)])
443
def test_range_header_mixed(self):
444
self.check_header('0-9,300-5000,-50',
445
ranges=[(0,9), (300,5000)],
449
class TestWallServer(object):
450
"""Tests exceptions during the connection phase"""
452
def create_transport_readonly_server(self):
453
return HttpServer(WallRequestHandler)
455
def test_http_has(self):
456
server = self.get_readonly_server()
457
t = self._transport(server.get_url())
458
# Unfortunately httplib (see HTTPResponse._read_status
459
# for details) make no distinction between a closed
460
# socket and badly formatted status line, so we can't
461
# just test for ConnectionError, we have to test
462
# InvalidHttpResponse too.
463
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
466
def test_http_get(self):
467
server = self.get_readonly_server()
468
t = self._transport(server.get_url())
469
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
473
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
474
"""Tests "wall" server for urllib implementation"""
476
_transport = HttpTransport_urllib
479
class TestWallServer_pycurl(TestWithTransport_pycurl,
481
TestCaseWithWebserver):
482
"""Tests "wall" server for pycurl implementation"""
485
class TestBadStatusServer(object):
486
"""Tests bad status from server."""
488
def create_transport_readonly_server(self):
489
return HttpServer(BadStatusRequestHandler)
491
def test_http_has(self):
492
server = self.get_readonly_server()
493
t = self._transport(server.get_url())
494
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
496
def test_http_get(self):
497
server = self.get_readonly_server()
498
t = self._transport(server.get_url())
499
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
502
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
503
"""Tests bad status server for urllib implementation"""
505
_transport = HttpTransport_urllib
508
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
510
TestCaseWithWebserver):
511
"""Tests bad status server for pycurl implementation"""
514
class TestInvalidStatusServer(TestBadStatusServer):
515
"""Tests invalid status from server.
517
Both implementations raises the same error as for a bad status.
520
def create_transport_readonly_server(self):
521
return HttpServer(InvalidStatusRequestHandler)
524
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
525
TestCaseWithWebserver):
526
"""Tests invalid status server for urllib implementation"""
528
_transport = HttpTransport_urllib
531
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
532
TestInvalidStatusServer,
533
TestCaseWithWebserver):
534
"""Tests invalid status server for pycurl implementation"""
537
class TestBadProtocolServer(object):
538
"""Tests bad protocol from server."""
540
def create_transport_readonly_server(self):
541
return HttpServer(BadProtocolRequestHandler)
543
def test_http_has(self):
544
server = self.get_readonly_server()
545
t = self._transport(server.get_url())
546
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
548
def test_http_get(self):
549
server = self.get_readonly_server()
550
t = self._transport(server.get_url())
551
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
554
class TestBadProtocolServer_urllib(TestBadProtocolServer,
555
TestCaseWithWebserver):
556
"""Tests bad protocol server for urllib implementation"""
558
_transport = HttpTransport_urllib
560
# curl don't check the protocol version
561
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
562
# TestBadProtocolServer,
563
# TestCaseWithWebserver):
564
# """Tests bad protocol server for pycurl implementation"""
567
class TestForbiddenServer(object):
568
"""Tests forbidden server"""
570
def create_transport_readonly_server(self):
571
return HttpServer(ForbiddenRequestHandler)
573
def test_http_has(self):
574
server = self.get_readonly_server()
575
t = self._transport(server.get_url())
576
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
578
def test_http_get(self):
579
server = self.get_readonly_server()
580
t = self._transport(server.get_url())
581
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
584
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
585
"""Tests forbidden server for urllib implementation"""
587
_transport = HttpTransport_urllib
590
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
592
TestCaseWithWebserver):
593
"""Tests forbidden server for pycurl implementation"""
596
class TestRecordingServer(TestCase):
598
def test_create(self):
599
server = RecordingServer(expect_body_tail=None)
600
self.assertEqual('', server.received_bytes)
601
self.assertEqual(None, server.host)
602
self.assertEqual(None, server.port)
604
def test_setUp_and_tearDown(self):
605
server = RecordingServer(expect_body_tail=None)
608
self.assertNotEqual(None, server.host)
609
self.assertNotEqual(None, server.port)
612
self.assertEqual(None, server.host)
613
self.assertEqual(None, server.port)
615
def test_send_receive_bytes(self):
616
server = RecordingServer(expect_body_tail='c')
618
self.addCleanup(server.tearDown)
619
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
620
sock.connect((server.host, server.port))
622
self.assertEqual('HTTP/1.1 200 OK\r\n',
623
osutils.recv_all(sock, 4096))
624
self.assertEqual('abc', server.received_bytes)
627
class TestRangeRequestServer(object):
628
"""Tests readv requests against server.
630
This MUST be used by daughter classes that also inherit from
631
TestCaseWithWebserver.
633
We can't inherit directly from TestCaseWithWebserver or the
634
test framework will try to create an instance which cannot
635
run, its implementation being incomplete.
639
TestCaseWithWebserver.setUp(self)
640
self.build_tree_contents([('a', '0123456789')],)
642
def test_readv(self):
643
server = self.get_readonly_server()
644
t = self._transport(server.get_url())
645
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
646
self.assertEqual(l[0], (0, '0'))
647
self.assertEqual(l[1], (1, '1'))
648
self.assertEqual(l[2], (3, '34'))
649
self.assertEqual(l[3], (9, '9'))
651
def test_readv_out_of_order(self):
652
server = self.get_readonly_server()
653
t = self._transport(server.get_url())
654
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
655
self.assertEqual(l[0], (1, '1'))
656
self.assertEqual(l[1], (9, '9'))
657
self.assertEqual(l[2], (0, '0'))
658
self.assertEqual(l[3], (3, '34'))
660
def test_readv_invalid_ranges(self):
661
server = self.get_readonly_server()
662
t = self._transport(server.get_url())
664
# This is intentionally reading off the end of the file
665
# since we are sure that it cannot get there
666
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
667
t.readv, 'a', [(1,1), (8,10)])
669
# This is trying to seek past the end of the file, it should
670
# also raise a special error
671
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
672
t.readv, 'a', [(12,2)])
675
class TestSingleRangeRequestServer(TestRangeRequestServer):
676
"""Test readv against a server which accept only single range requests"""
678
def create_transport_readonly_server(self):
679
return HttpServer(SingleRangeRequestHandler)
682
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
683
TestCaseWithWebserver):
684
"""Tests single range requests accepting server for urllib implementation"""
686
_transport = HttpTransport_urllib
689
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
690
TestSingleRangeRequestServer,
691
TestCaseWithWebserver):
692
"""Tests single range requests accepting server for pycurl implementation"""
695
class TestNoRangeRequestServer(TestRangeRequestServer):
696
"""Test readv against a server which do not accept range requests"""
698
def create_transport_readonly_server(self):
699
return HttpServer(NoRangeRequestHandler)
702
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
703
TestCaseWithWebserver):
704
"""Tests range requests refusing server for urllib implementation"""
706
_transport = HttpTransport_urllib
709
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
710
TestNoRangeRequestServer,
711
TestCaseWithWebserver):
712
"""Tests range requests refusing server for pycurl implementation"""
715
class TestHttpProxyWhiteBox(TestCase):
716
"""Whitebox test proxy http authorization.
718
Only the urllib implementation is tested here.
728
def _install_env(self, env):
729
for name, value in env.iteritems():
730
self._old_env[name] = osutils.set_or_unset_env(name, value)
732
def _restore_env(self):
733
for name, value in self._old_env.iteritems():
734
osutils.set_or_unset_env(name, value)
736
def _proxied_request(self):
737
handler = ProxyHandler(PasswordManager())
738
request = Request('GET','http://baz/buzzle')
739
handler.set_proxy(request, 'http')
742
def test_empty_user(self):
743
self._install_env({'http_proxy': 'http://bar.com'})
744
request = self._proxied_request()
745
self.assertFalse(request.headers.has_key('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
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
783
self.no_proxy_host = self.proxy_address
784
# The secondary server is the proxy
785
self.proxy = self.get_secondary_server()
786
self.proxy_url = self.proxy.get_url()
789
def create_transport_secondary_server(self):
790
"""Creates an http server that will serve files with
791
'-proxied' appended to their names.
795
def _install_env(self, env):
796
for name, value in env.iteritems():
797
self._old_env[name] = osutils.set_or_unset_env(name, value)
799
def _restore_env(self):
800
for name, value in self._old_env.iteritems():
801
osutils.set_or_unset_env(name, value)
803
def proxied_in_env(self, env):
804
self._install_env(env)
805
url = self.server.get_url()
806
t = self._transport(url)
808
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
812
def not_proxied_in_env(self, env):
813
self._install_env(env)
814
url = self.server.get_url()
815
t = self._transport(url)
817
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
821
def test_http_proxy(self):
822
self.proxied_in_env({'http_proxy': self.proxy_url})
824
def test_HTTP_PROXY(self):
825
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
827
def test_all_proxy(self):
828
self.proxied_in_env({'all_proxy': self.proxy_url})
830
def test_ALL_PROXY(self):
831
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
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_HTTP_PROXY_with_NO_PROXY(self):
838
self.not_proxied_in_env({'HTTP_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})
845
def test_ALL_PROXY_with_NO_PROXY(self):
846
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
847
'NO_PROXY': self.no_proxy_host})
849
def test_http_proxy_without_scheme(self):
850
self.assertRaises(errors.InvalidURL,
852
{'http_proxy': self.proxy_address})
855
class TestProxyHttpServer_urllib(TestProxyHttpServer,
856
TestCaseWithTwoWebservers):
857
"""Tests proxy server for urllib implementation"""
859
_transport = HttpTransport_urllib
862
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
864
TestCaseWithTwoWebservers):
865
"""Tests proxy server for pycurl implementation"""
868
TestProxyHttpServer.setUp(self)
869
# Oh my ! pycurl does not check for the port as part of
870
# no_proxy :-( So we just test the host part
871
self.no_proxy_host = 'localhost'
873
def test_HTTP_PROXY(self):
874
# pycurl do not check HTTP_PROXY for security reasons
875
# (for use in a CGI context that we do not care
876
# about. Should we ?)
879
def test_HTTP_PROXY_with_NO_PROXY(self):
882
def test_http_proxy_without_scheme(self):
883
# pycurl *ignores* invalid proxy env variables. If that
884
# ever change in the future, this test will fail
885
# indicating that pycurl do not ignore anymore such
887
self.not_proxied_in_env({'http_proxy': self.proxy_address})
890
class TestRanges(object):
891
"""Test the Range header in GET methods..
893
This MUST be used by daughter classes that also inherit from
894
TestCaseWithWebserver.
896
We can't inherit directly from TestCaseWithWebserver or the
897
test framework will try to create an instance which cannot
898
run, its implementation being incomplete.
902
TestCaseWithWebserver.setUp(self)
903
self.build_tree_contents([('a', '0123456789')],)
904
server = self.get_readonly_server()
905
self.transport = self._transport(server.get_url())
907
def _file_contents(self, relpath, ranges, tail_amount=0):
908
code, data = self.transport._get(relpath, ranges)
909
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
910
for start, end in ranges:
912
yield data.read(end - start + 1)
914
def _file_tail(self, relpath, tail_amount):
915
code, data = self.transport._get(relpath, [], tail_amount)
916
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
917
data.seek(-tail_amount + 1, 2)
918
return data.read(tail_amount)
920
def test_range_header(self):
922
map(self.assertEqual,['0', '234'],
923
list(self._file_contents('a', [(0,0), (2,4)])),)
925
self.assertEqual('789', self._file_tail('a', 3))
926
# Syntactically invalid range
927
self.assertRaises(errors.InvalidRange,
928
self.transport._get, 'a', [(4, 3)])
929
# Semantically invalid range
930
self.assertRaises(errors.InvalidRange,
931
self.transport._get, 'a', [(42, 128)])
934
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
935
"""Test the Range header in GET methods for urllib implementation"""
937
_transport = HttpTransport_urllib
940
class TestRanges_pycurl(TestWithTransport_pycurl,
942
TestCaseWithWebserver):
943
"""Test the Range header in GET methods for pycurl implementation"""
946
class TestHTTPRedirections(object):
947
"""Test redirection between http servers.
949
This MUST be used by daughter classes that also inherit from
950
TestCaseWithRedirectedWebserver.
952
We can't inherit directly from TestCaseWithTwoWebservers or the
953
test framework will try to create an instance which cannot
954
run, its implementation being incomplete.
957
def create_transport_secondary_server(self):
958
"""Create the secondary server redirecting to the primary server"""
959
new = self.get_readonly_server()
961
redirecting = HTTPServerRedirecting()
962
redirecting.redirect_to(new.host, new.port)
966
super(TestHTTPRedirections, self).setUp()
967
self.build_tree_contents([('a', '0123456789'),
969
'# Bazaar revision bundle v0.9\n#\n')
972
self.old_transport = self._transport(self.old_server.get_url())
974
def test_redirected(self):
975
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
976
t = self._transport(self.new_server.get_url())
977
self.assertEqual('0123456789', t.get('a').read())
979
def test_read_redirected_bundle_from_url(self):
980
from bzrlib.bundle import read_bundle_from_url
981
url = self.old_transport.abspath('bundle')
982
bundle = read_bundle_from_url(url)
983
# If read_bundle_from_url was successful we get an empty bundle
984
self.assertEqual([], bundle.revisions)
987
class TestHTTPRedirections_urllib(TestHTTPRedirections,
988
TestCaseWithRedirectedWebserver):
989
"""Tests redirections for urllib implementation"""
991
_transport = HttpTransport_urllib
995
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
996
TestHTTPRedirections,
997
TestCaseWithRedirectedWebserver):
998
"""Tests redirections for pycurl implementation"""
1001
class RedirectedRequest(Request):
1002
"""Request following redirections"""
1004
init_orig = Request.__init__
1006
def __init__(self, method, url, *args, **kwargs):
1007
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1008
self.follow_redirections = True
1011
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1012
"""Test redirections provided by urllib.
1014
http implementations do not redirect silently anymore (they
1015
do not redirect at all in fact). The mechanism is still in
1016
place at the _urllib2_wrappers.Request level and these tests
1019
For the pycurl implementation
1020
the redirection have been deleted as we may deprecate pycurl
1021
and I have no place to keep a working implementation.
1025
_transport = HttpTransport_urllib
1028
super(TestHTTPSilentRedirections_urllib, self).setUp()
1029
self.setup_redirected_request()
1030
self.addCleanup(self.cleanup_redirected_request)
1031
self.build_tree_contents([('a','a'),
1033
('1/a', 'redirected once'),
1035
('2/a', 'redirected twice'),
1037
('3/a', 'redirected thrice'),
1039
('4/a', 'redirected 4 times'),
1041
('5/a', 'redirected 5 times'),
1044
self.old_transport = self._transport(self.old_server.get_url())
1046
def setup_redirected_request(self):
1047
self.original_class = _urllib2_wrappers.Request
1048
_urllib2_wrappers.Request = RedirectedRequest
1050
def cleanup_redirected_request(self):
1051
_urllib2_wrappers.Request = self.original_class
1053
def create_transport_secondary_server(self):
1054
"""Create the secondary server, redirections are defined in the tests"""
1055
return HTTPServerRedirecting()
1057
def test_one_redirection(self):
1058
t = self.old_transport
1060
req = RedirectedRequest('GET', t.abspath('a'))
1061
req.follow_redirections = True
1062
new_prefix = 'http://%s:%s' % (self.new_server.host,
1063
self.new_server.port)
1064
self.old_server.redirections = \
1065
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1066
self.assertEquals('redirected once',t._perform(req).read())
1068
def test_five_redirections(self):
1069
t = self.old_transport
1071
req = RedirectedRequest('GET', t.abspath('a'))
1072
req.follow_redirections = True
1073
old_prefix = 'http://%s:%s' % (self.old_server.host,
1074
self.old_server.port)
1075
new_prefix = 'http://%s:%s' % (self.new_server.host,
1076
self.new_server.port)
1077
self.old_server.redirections = \
1078
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1079
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1080
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1081
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1082
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1084
self.assertEquals('redirected 5 times',t._perform(req).read())
1087
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1088
"""Test transport.do_catching_redirections.
1090
We arbitrarily choose to use urllib transports
1093
_transport = HttpTransport_urllib
1096
super(TestDoCatchRedirections, self).setUp()
1097
self.build_tree_contents([('a', '0123456789'),],)
1099
self.old_transport = self._transport(self.old_server.get_url())
1101
def get_a(self, transport):
1102
return transport.get('a')
1104
def test_no_redirection(self):
1105
t = self._transport(self.new_server.get_url())
1107
# We use None for redirected so that we fail if redirected
1108
self.assertEquals('0123456789',
1109
do_catching_redirections(self.get_a, t, None).read())
1111
def test_one_redirection(self):
1112
self.redirections = 0
1114
def redirected(transport, exception, redirection_notice):
1115
self.redirections += 1
1116
dir, file = urlutils.split(exception.target)
1117
return self._transport(dir)
1119
self.assertEquals('0123456789',
1120
do_catching_redirections(self.get_a,
1124
self.assertEquals(1, self.redirections)
1126
def test_redirection_loop(self):
1128
def redirected(transport, exception, redirection_notice):
1129
# By using the redirected url as a base dir for the
1130
# *old* transport, we create a loop: a => a/a =>
1132
return self.old_transport.clone(exception.target)
1134
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1135
self.get_a, self.old_transport, redirected)
1138
class TestAuth(object):
1139
"""Test some authentication scheme specified by daughter class.
1141
This MUST be used by daughter classes that also inherit from
1142
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1146
"""Set up the test environment
1148
Daughter classes should set up their own environment
1149
(including self.server) and explicitely call this
1150
method. This is needed because we want to reuse the same
1151
tests for proxy and no-proxy accesses which have
1152
different ways of setting self.server.
1154
self.build_tree_contents([('a', 'contents of a\n'),
1155
('b', 'contents of b\n'),])
1156
self.old_factory = ui.ui_factory
1157
self.addCleanup(self.restoreUIFactory)
1159
def restoreUIFactory(self):
1160
ui.ui_factory = self.old_factory
1162
def get_user_url(self, user=None, password=None):
1163
"""Build an url embedding user and password"""
1164
url = '%s://' % self.server._url_protocol
1165
if user is not None:
1167
if password is not None:
1168
url += ':' + password
1170
url += '%s:%s/' % (self.server.host, self.server.port)
1173
def test_no_user(self):
1174
self.server.add_user('joe', 'foo')
1175
t = self.get_user_transport()
1176
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1177
# Only one 'Authentication Required' error should occur
1178
self.assertEqual(1, self.server.auth_required_errors)
1180
def test_empty_pass(self):
1181
self.server.add_user('joe', '')
1182
t = self.get_user_transport('joe', '')
1183
self.assertEqual('contents of a\n', t.get('a').read())
1184
# Only one 'Authentication Required' error should occur
1185
self.assertEqual(1, self.server.auth_required_errors)
1187
def test_user_pass(self):
1188
self.server.add_user('joe', 'foo')
1189
t = self.get_user_transport('joe', 'foo')
1190
self.assertEqual('contents of a\n', t.get('a').read())
1191
# Only one 'Authentication Required' error should occur
1192
self.assertEqual(1, self.server.auth_required_errors)
1194
def test_unknown_user(self):
1195
self.server.add_user('joe', 'foo')
1196
t = self.get_user_transport('bill', 'foo')
1197
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1198
# Two 'Authentication Required' errors should occur (the
1199
# initial 'who are you' and 'I don't know you, who are
1201
self.assertEqual(2, self.server.auth_required_errors)
1203
def test_wrong_pass(self):
1204
self.server.add_user('joe', 'foo')
1205
t = self.get_user_transport('joe', 'bar')
1206
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1207
# Two 'Authentication Required' error should occur (the
1208
# initial 'who are you' and 'this is not you, who are you')
1209
self.assertEqual(2, self.server.auth_required_errors)
1211
def test_prompt_for_password(self):
1212
self.server.add_user('joe', 'foo')
1213
t = self.get_user_transport('joe', None)
1214
ui.ui_factory = TestUIFactory(stdin='foo\n', stdout=StringIOWrapper())
1215
self.assertEqual('contents of a\n',t.get('a').read())
1216
# stdin should be empty
1217
self.assertEqual('', ui.ui_factory.stdin.readline())
1218
# And we shouldn't prompt again for a different request
1219
# against the same transport.
1220
self.assertEqual('contents of b\n',t.get('b').read())
1222
# And neither against a clone
1223
self.assertEqual('contents of b\n',t2.get('b').read())
1224
# Only one 'Authentication Required' error should occur
1225
self.assertEqual(1, self.server.auth_required_errors)
1228
class TestHTTPAuth(TestAuth):
1229
"""Test HTTP authentication schemes.
1231
Daughter classes MUST inherit from TestCaseWithWebserver too.
1234
_auth_header = 'Authorization'
1237
TestCaseWithWebserver.setUp(self)
1238
self.server = self.get_readonly_server()
1239
TestAuth.setUp(self)
1241
def get_user_transport(self, user=None, password=None):
1242
return self._transport(self.get_user_url(user, password))
1245
class TestProxyAuth(TestAuth):
1246
"""Test proxy authentication schemes.
1248
Daughter classes MUST also inherit from TestCaseWithWebserver.
1250
_auth_header = 'Proxy-authorization'
1253
TestCaseWithWebserver.setUp(self)
1254
self.server = self.get_readonly_server()
1256
self.addCleanup(self._restore_env)
1257
TestAuth.setUp(self)
1258
# Override the contents to avoid false positives
1259
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1260
('b', 'not proxied contents of b\n'),
1261
('a-proxied', 'contents of a\n'),
1262
('b-proxied', 'contents of b\n'),
1265
def get_user_transport(self, user=None, password=None):
1266
self._install_env({'all_proxy': self.get_user_url(user, password)})
1267
return self._transport(self.server.get_url())
1269
def _install_env(self, env):
1270
for name, value in env.iteritems():
1271
self._old_env[name] = osutils.set_or_unset_env(name, value)
1273
def _restore_env(self):
1274
for name, value in self._old_env.iteritems():
1275
osutils.set_or_unset_env(name, value)
1278
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1279
"""Test http basic authentication scheme"""
1281
_transport = HttpTransport_urllib
1283
def create_transport_readonly_server(self):
1284
return HTTPBasicAuthServer()
1287
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1288
"""Test proxy basic authentication scheme"""
1290
_transport = HttpTransport_urllib
1292
def create_transport_readonly_server(self):
1293
return ProxyBasicAuthServer()
1296
class TestHTTPDigestAuth(TestHTTPAuth, TestCaseWithWebserver):
1297
"""Test http digest authentication scheme"""
1299
_transport = HttpTransport_urllib
1301
def create_transport_readonly_server(self):
1302
return HTTPDigestAuthServer()
1305
class TestHTTPProxyDigestAuth(TestProxyAuth, TestCaseWithWebserver):
1306
"""Test proxy digest authentication scheme"""
1308
_transport = HttpTransport_urllib
1310
def create_transport_readonly_server(self):
1311
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
