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
37
from bzrlib.tests import (
43
from bzrlib.tests.HttpServer import (
48
from bzrlib.tests.HTTPTestUtil import (
49
BadProtocolRequestHandler,
50
BadStatusRequestHandler,
51
ForbiddenRequestHandler,
54
HTTPServerRedirecting,
55
InvalidStatusRequestHandler,
56
NoRangeRequestHandler,
58
ProxyDigestAuthServer,
60
SingleRangeRequestHandler,
61
SingleOnlyRangeRequestHandler,
62
TestCaseWithRedirectedWebserver,
63
TestCaseWithTwoWebservers,
64
TestCaseWithWebserver,
67
from bzrlib.transport import (
68
do_catching_redirections,
72
from bzrlib.transport.http import (
77
from bzrlib.transport.http._urllib import HttpTransport_urllib
78
from bzrlib.transport.http._urllib2_wrappers import (
85
class FakeManager(object):
90
def add_password(self, realm, host, username, password):
91
self.credentials.append([realm, host, username, password])
94
class RecordingServer(object):
95
"""A fake HTTP server.
97
It records the bytes sent to it, and replies with a 200.
100
def __init__(self, expect_body_tail=None):
103
:type expect_body_tail: str
104
:param expect_body_tail: a reply won't be sent until this string is
107
self._expect_body_tail = expect_body_tail
110
self.received_bytes = ''
113
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
114
self._sock.bind(('127.0.0.1', 0))
115
self.host, self.port = self._sock.getsockname()
116
self._ready = threading.Event()
117
self._thread = threading.Thread(target=self._accept_read_and_reply)
118
self._thread.setDaemon(True)
122
def _accept_read_and_reply(self):
125
self._sock.settimeout(5)
127
conn, address = self._sock.accept()
128
# On win32, the accepted connection will be non-blocking to start
129
# with because we're using settimeout.
130
conn.setblocking(True)
131
while not self.received_bytes.endswith(self._expect_body_tail):
132
self.received_bytes += conn.recv(4096)
133
conn.sendall('HTTP/1.1 200 OK\r\n')
134
except socket.timeout:
135
# Make sure the client isn't stuck waiting for us to e.g. accept.
138
# The client may have already closed the socket.
145
# We might have already closed it. We don't care.
151
class TestWithTransport_pycurl(object):
152
"""Test case to inherit from if pycurl is present"""
154
def _get_pycurl_maybe(self):
156
from bzrlib.transport.http._pycurl import PyCurlTransport
157
return PyCurlTransport
158
except errors.DependencyNotPresent:
159
raise TestSkipped('pycurl not present')
161
_transport = property(_get_pycurl_maybe)
164
class TestHttpUrls(TestCase):
166
# TODO: This should be moved to authorization tests once they
169
def test_url_parsing(self):
171
url = extract_auth('http://example.com', f)
172
self.assertEquals('http://example.com', url)
173
self.assertEquals(0, len(f.credentials))
174
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
175
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
176
self.assertEquals(1, len(f.credentials))
177
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
181
class TestHttpTransportUrls(object):
182
"""Test the http urls.
184
This MUST be used by daughter classes that also inherit from
187
We can't inherit directly from TestCase or the
188
test framework will try to create an instance which cannot
189
run, its implementation being incomplete.
192
def test_abs_url(self):
193
"""Construction of absolute http URLs"""
194
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
195
eq = self.assertEqualDiff
196
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
197
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
198
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
199
eq(t.abspath('.bzr/1//2/./3'),
200
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
202
def test_invalid_http_urls(self):
203
"""Trap invalid construction of urls"""
204
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
205
self.assertRaises(ValueError, t.abspath, '.bzr/')
206
t = self._transport('http://http://bazaar-vcs.org/bzr/bzr.dev/')
207
self.assertRaises((errors.InvalidURL, errors.ConnectionError),
210
def test_http_root_urls(self):
211
"""Construction of URLs from server root"""
212
t = self._transport('http://bzr.ozlabs.org/')
213
eq = self.assertEqualDiff
214
eq(t.abspath('.bzr/tree-version'),
215
'http://bzr.ozlabs.org/.bzr/tree-version')
217
def test_http_impl_urls(self):
218
"""There are servers which ask for particular clients to connect"""
219
server = self._server()
222
url = server.get_url()
223
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
228
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
229
"""Test http urls with urllib"""
231
_transport = HttpTransport_urllib
232
_server = HttpServer_urllib
233
_qualified_prefix = 'http+urllib'
236
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
238
"""Test http urls with pycurl"""
240
_server = HttpServer_PyCurl
241
_qualified_prefix = 'http+pycurl'
243
# TODO: This should really be moved into another pycurl
244
# specific test. When https tests will be implemented, take
245
# this one into account.
246
def test_pycurl_without_https_support(self):
247
"""Test that pycurl without SSL do not fail with a traceback.
249
For the purpose of the test, we force pycurl to ignore
250
https by supplying a fake version_info that do not
256
raise TestSkipped('pycurl not present')
257
# Now that we have pycurl imported, we can fake its version_info
258
# This was taken from a windows pycurl without SSL
260
pycurl.version_info = lambda : (2,
268
('ftp', 'gopher', 'telnet',
269
'dict', 'ldap', 'http', 'file'),
273
self.assertRaises(errors.DependencyNotPresent, self._transport,
274
'https://launchpad.net')
276
class TestHttpConnections(object):
277
"""Test the http connections.
279
This MUST be used by daughter classes that also inherit from
280
TestCaseWithWebserver.
282
We can't inherit directly from TestCaseWithWebserver or the
283
test framework will try to create an instance which cannot
284
run, its implementation being incomplete.
288
TestCaseWithWebserver.setUp(self)
289
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
290
transport=self.get_transport())
292
def test_http_has(self):
293
server = self.get_readonly_server()
294
t = self._transport(server.get_url())
295
self.assertEqual(t.has('foo/bar'), True)
296
self.assertEqual(len(server.logs), 1)
297
self.assertContainsRe(server.logs[0],
298
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
300
def test_http_has_not_found(self):
301
server = self.get_readonly_server()
302
t = self._transport(server.get_url())
303
self.assertEqual(t.has('not-found'), False)
304
self.assertContainsRe(server.logs[1],
305
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
307
def test_http_get(self):
308
server = self.get_readonly_server()
309
t = self._transport(server.get_url())
310
fp = t.get('foo/bar')
311
self.assertEqualDiff(
313
'contents of foo/bar\n')
314
self.assertEqual(len(server.logs), 1)
315
self.assertTrue(server.logs[0].find(
316
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
317
% bzrlib.__version__) > -1)
319
def test_get_smart_medium(self):
320
# For HTTP, get_smart_medium should return the transport object.
321
server = self.get_readonly_server()
322
http_transport = self._transport(server.get_url())
323
medium = http_transport.get_smart_medium()
324
self.assertIs(medium, http_transport)
326
def test_has_on_bogus_host(self):
327
# Get a free address and don't 'accept' on it, so that we
328
# can be sure there is no http handler there, but set a
329
# reasonable timeout to not slow down tests too much.
330
default_timeout = socket.getdefaulttimeout()
332
socket.setdefaulttimeout(2)
334
s.bind(('localhost', 0))
335
t = self._transport('http://%s:%s/' % s.getsockname())
336
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
338
socket.setdefaulttimeout(default_timeout)
341
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
342
"""Test http connections with urllib"""
344
_transport = HttpTransport_urllib
348
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
350
TestCaseWithWebserver):
351
"""Test http connections with pycurl"""
354
class TestHttpTransportRegistration(TestCase):
355
"""Test registrations of various http implementations"""
357
def test_http_registered(self):
358
# urlllib should always be present
359
t = get_transport('http+urllib://bzr.google.com/')
360
self.assertIsInstance(t, Transport)
361
self.assertIsInstance(t, HttpTransport_urllib)
364
class TestOffsets(TestCase):
365
"""Test offsets_to_ranges method"""
367
def test_offsets_to_ranges_simple(self):
368
to_range = HttpTransportBase.offsets_to_ranges
369
ranges = to_range([(10, 1)])
370
self.assertEqual([[10, 10]], ranges)
372
ranges = to_range([(0, 1), (1, 1)])
373
self.assertEqual([[0, 1]], ranges)
375
ranges = to_range([(1, 1), (0, 1)])
376
self.assertEqual([[0, 1]], ranges)
378
def test_offset_to_ranges_overlapped(self):
379
to_range = HttpTransportBase.offsets_to_ranges
381
ranges = to_range([(10, 1), (20, 2), (22, 5)])
382
self.assertEqual([[10, 10], [20, 26]], ranges)
384
ranges = to_range([(10, 1), (11, 2), (22, 5)])
385
self.assertEqual([[10, 12], [22, 26]], ranges)
388
class TestPost(object):
390
def _test_post_body_is_received(self, scheme):
391
server = RecordingServer(expect_body_tail='end-of-body')
393
self.addCleanup(server.tearDown)
394
url = '%s://%s:%s/' % (scheme, server.host, server.port)
396
http_transport = get_transport(url)
397
except errors.UnsupportedProtocol:
398
raise TestSkipped('%s not available' % scheme)
399
code, response = http_transport._post('abc def end-of-body')
401
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
402
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
403
# The transport should not be assuming that the server can accept
404
# chunked encoding the first time it connects, because HTTP/1.1, so we
405
# check for the literal string.
407
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
410
class TestPost_urllib(TestCase, TestPost):
411
"""TestPost for urllib implementation"""
413
_transport = HttpTransport_urllib
415
def test_post_body_is_received_urllib(self):
416
self._test_post_body_is_received('http+urllib')
419
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
420
"""TestPost for pycurl implementation"""
422
def test_post_body_is_received_pycurl(self):
423
self._test_post_body_is_received('http+pycurl')
426
class TestRangeHeader(TestCase):
427
"""Test range_header method"""
429
def check_header(self, value, ranges=[], tail=0):
430
range_header = HttpTransportBase.range_header
431
self.assertEqual(value, range_header(ranges, tail))
433
def test_range_header_single(self):
434
self.check_header('0-9', ranges=[[0,9]])
435
self.check_header('100-109', ranges=[[100,109]])
437
def test_range_header_tail(self):
438
self.check_header('-10', tail=10)
439
self.check_header('-50', tail=50)
441
def test_range_header_multi(self):
442
self.check_header('0-9,100-200,300-5000',
443
ranges=[(0,9), (100, 200), (300,5000)])
445
def test_range_header_mixed(self):
446
self.check_header('0-9,300-5000,-50',
447
ranges=[(0,9), (300,5000)],
451
class TestWallServer(object):
452
"""Tests exceptions during the connection phase"""
454
def create_transport_readonly_server(self):
455
return HttpServer(WallRequestHandler)
457
def test_http_has(self):
458
server = self.get_readonly_server()
459
t = self._transport(server.get_url())
460
# Unfortunately httplib (see HTTPResponse._read_status
461
# for details) make no distinction between a closed
462
# socket and badly formatted status line, so we can't
463
# just test for ConnectionError, we have to test
464
# InvalidHttpResponse too.
465
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
468
def test_http_get(self):
469
server = self.get_readonly_server()
470
t = self._transport(server.get_url())
471
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
475
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
476
"""Tests "wall" server for urllib implementation"""
478
_transport = HttpTransport_urllib
481
class TestWallServer_pycurl(TestWithTransport_pycurl,
483
TestCaseWithWebserver):
484
"""Tests "wall" server for pycurl implementation"""
487
class TestBadStatusServer(object):
488
"""Tests bad status from server."""
490
def create_transport_readonly_server(self):
491
return HttpServer(BadStatusRequestHandler)
493
def test_http_has(self):
494
server = self.get_readonly_server()
495
t = self._transport(server.get_url())
496
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
498
def test_http_get(self):
499
server = self.get_readonly_server()
500
t = self._transport(server.get_url())
501
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
504
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
505
"""Tests bad status server for urllib implementation"""
507
_transport = HttpTransport_urllib
510
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
512
TestCaseWithWebserver):
513
"""Tests bad status server for pycurl implementation"""
516
class TestInvalidStatusServer(TestBadStatusServer):
517
"""Tests invalid status from server.
519
Both implementations raises the same error as for a bad status.
522
def create_transport_readonly_server(self):
523
return HttpServer(InvalidStatusRequestHandler)
526
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
527
TestCaseWithWebserver):
528
"""Tests invalid status server for urllib implementation"""
530
_transport = HttpTransport_urllib
533
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
534
TestInvalidStatusServer,
535
TestCaseWithWebserver):
536
"""Tests invalid status server for pycurl implementation"""
539
class TestBadProtocolServer(object):
540
"""Tests bad protocol from server."""
542
def create_transport_readonly_server(self):
543
return HttpServer(BadProtocolRequestHandler)
545
def test_http_has(self):
546
server = self.get_readonly_server()
547
t = self._transport(server.get_url())
548
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
550
def test_http_get(self):
551
server = self.get_readonly_server()
552
t = self._transport(server.get_url())
553
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
556
class TestBadProtocolServer_urllib(TestBadProtocolServer,
557
TestCaseWithWebserver):
558
"""Tests bad protocol server for urllib implementation"""
560
_transport = HttpTransport_urllib
562
# curl don't check the protocol version
563
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
564
# TestBadProtocolServer,
565
# TestCaseWithWebserver):
566
# """Tests bad protocol server for pycurl implementation"""
569
class TestForbiddenServer(object):
570
"""Tests forbidden server"""
572
def create_transport_readonly_server(self):
573
return HttpServer(ForbiddenRequestHandler)
575
def test_http_has(self):
576
server = self.get_readonly_server()
577
t = self._transport(server.get_url())
578
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
580
def test_http_get(self):
581
server = self.get_readonly_server()
582
t = self._transport(server.get_url())
583
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
586
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
587
"""Tests forbidden server for urllib implementation"""
589
_transport = HttpTransport_urllib
592
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
594
TestCaseWithWebserver):
595
"""Tests forbidden server for pycurl implementation"""
598
class TestRecordingServer(TestCase):
600
def test_create(self):
601
server = RecordingServer(expect_body_tail=None)
602
self.assertEqual('', server.received_bytes)
603
self.assertEqual(None, server.host)
604
self.assertEqual(None, server.port)
606
def test_setUp_and_tearDown(self):
607
server = RecordingServer(expect_body_tail=None)
610
self.assertNotEqual(None, server.host)
611
self.assertNotEqual(None, server.port)
614
self.assertEqual(None, server.host)
615
self.assertEqual(None, server.port)
617
def test_send_receive_bytes(self):
618
server = RecordingServer(expect_body_tail='c')
620
self.addCleanup(server.tearDown)
621
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
622
sock.connect((server.host, server.port))
624
self.assertEqual('HTTP/1.1 200 OK\r\n',
625
osutils.recv_all(sock, 4096))
626
self.assertEqual('abc', server.received_bytes)
629
class TestRangeRequestServer(object):
630
"""Tests readv requests against server.
632
This MUST be used by daughter classes that also inherit from
633
TestCaseWithWebserver.
635
We can't inherit directly from TestCaseWithWebserver or the
636
test framework will try to create an instance which cannot
637
run, its implementation being incomplete.
641
TestCaseWithWebserver.setUp(self)
642
self.build_tree_contents([('a', '0123456789')],)
644
def test_readv(self):
645
server = self.get_readonly_server()
646
t = self._transport(server.get_url())
647
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
648
self.assertEqual(l[0], (0, '0'))
649
self.assertEqual(l[1], (1, '1'))
650
self.assertEqual(l[2], (3, '34'))
651
self.assertEqual(l[3], (9, '9'))
653
def test_readv_out_of_order(self):
654
server = self.get_readonly_server()
655
t = self._transport(server.get_url())
656
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
657
self.assertEqual(l[0], (1, '1'))
658
self.assertEqual(l[1], (9, '9'))
659
self.assertEqual(l[2], (0, '0'))
660
self.assertEqual(l[3], (3, '34'))
662
def test_readv_invalid_ranges(self):
663
server = self.get_readonly_server()
664
t = self._transport(server.get_url())
666
# This is intentionally reading off the end of the file
667
# since we are sure that it cannot get there
668
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
669
t.readv, 'a', [(1,1), (8,10)])
671
# This is trying to seek past the end of the file, it should
672
# also raise a special error
673
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
674
t.readv, 'a', [(12,2)])
677
class TestSingleRangeRequestServer(TestRangeRequestServer):
678
"""Test readv against a server which accept only single range requests"""
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleRangeRequestHandler)
684
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
688
_transport = HttpTransport_urllib
691
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleRangeRequestServer,
693
TestCaseWithWebserver):
694
"""Tests single range requests accepting server for pycurl implementation"""
697
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
698
"""Test readv against a server which only accept single range requests"""
700
def create_transport_readonly_server(self):
701
return HttpServer(SingleOnlyRangeRequestHandler)
704
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests single range requests accepting server for urllib implementation"""
708
_transport = HttpTransport_urllib
711
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestSingleOnlyRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests single range requests accepting server for pycurl implementation"""
717
class TestNoRangeRequestServer(TestRangeRequestServer):
718
"""Test readv against a server which do not accept range requests"""
720
def create_transport_readonly_server(self):
721
return HttpServer(NoRangeRequestHandler)
724
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
725
TestCaseWithWebserver):
726
"""Tests range requests refusing server for urllib implementation"""
728
_transport = HttpTransport_urllib
731
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
732
TestNoRangeRequestServer,
733
TestCaseWithWebserver):
734
"""Tests range requests refusing server for pycurl implementation"""
737
class TestHttpProxyWhiteBox(TestCase):
738
"""Whitebox test proxy http authorization.
740
Only the urllib implementation is tested here.
750
def _install_env(self, env):
751
for name, value in env.iteritems():
752
self._old_env[name] = osutils.set_or_unset_env(name, value)
754
def _restore_env(self):
755
for name, value in self._old_env.iteritems():
756
osutils.set_or_unset_env(name, value)
758
def _proxied_request(self):
759
handler = ProxyHandler(PasswordManager())
760
request = Request('GET','http://baz/buzzle')
761
handler.set_proxy(request, 'http')
764
def test_empty_user(self):
765
self._install_env({'http_proxy': 'http://bar.com'})
766
request = self._proxied_request()
767
self.assertFalse(request.headers.has_key('Proxy-authorization'))
769
def test_invalid_proxy(self):
770
"""A proxy env variable without scheme"""
771
self._install_env({'http_proxy': 'host:1234'})
772
self.assertRaises(errors.InvalidURL, self._proxied_request)
775
class TestProxyHttpServer(object):
776
"""Tests proxy server.
778
This MUST be used by daughter classes that also inherit from
779
TestCaseWithTwoWebservers.
781
We can't inherit directly from TestCaseWithTwoWebservers or
782
the test framework will try to create an instance which
783
cannot run, its implementation being incomplete.
785
Be aware that we do not setup a real proxy here. Instead, we
786
check that the *connection* goes through the proxy by serving
787
different content (the faked proxy server append '-proxied'
791
# FIXME: We don't have an https server available, so we don't
792
# test https connections.
794
# FIXME: Once the test suite is better fitted to test
795
# authorization schemes, test proxy authorizations too (see
799
TestCaseWithTwoWebservers.setUp(self)
800
self.build_tree_contents([('foo', 'contents of foo\n'),
801
('foo-proxied', 'proxied contents of foo\n')])
802
# Let's setup some attributes for tests
803
self.server = self.get_readonly_server()
804
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
805
self.no_proxy_host = self.proxy_address
806
# The secondary server is the proxy
807
self.proxy = self.get_secondary_server()
808
self.proxy_url = self.proxy.get_url()
811
def create_transport_secondary_server(self):
812
"""Creates an http server that will serve files with
813
'-proxied' appended to their names.
817
def _install_env(self, env):
818
for name, value in env.iteritems():
819
self._old_env[name] = osutils.set_or_unset_env(name, value)
821
def _restore_env(self):
822
for name, value in self._old_env.iteritems():
823
osutils.set_or_unset_env(name, value)
825
def proxied_in_env(self, env):
826
self._install_env(env)
827
url = self.server.get_url()
828
t = self._transport(url)
830
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
834
def not_proxied_in_env(self, env):
835
self._install_env(env)
836
url = self.server.get_url()
837
t = self._transport(url)
839
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
843
def test_http_proxy(self):
844
self.proxied_in_env({'http_proxy': self.proxy_url})
846
def test_HTTP_PROXY(self):
847
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
849
def test_all_proxy(self):
850
self.proxied_in_env({'all_proxy': self.proxy_url})
852
def test_ALL_PROXY(self):
853
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
855
def test_http_proxy_with_no_proxy(self):
856
self.not_proxied_in_env({'http_proxy': self.proxy_url,
857
'no_proxy': self.no_proxy_host})
859
def test_HTTP_PROXY_with_NO_PROXY(self):
860
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
861
'NO_PROXY': self.no_proxy_host})
863
def test_all_proxy_with_no_proxy(self):
864
self.not_proxied_in_env({'all_proxy': self.proxy_url,
865
'no_proxy': self.no_proxy_host})
867
def test_ALL_PROXY_with_NO_PROXY(self):
868
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
869
'NO_PROXY': self.no_proxy_host})
871
def test_http_proxy_without_scheme(self):
872
self.assertRaises(errors.InvalidURL,
874
{'http_proxy': self.proxy_address})
877
class TestProxyHttpServer_urllib(TestProxyHttpServer,
878
TestCaseWithTwoWebservers):
879
"""Tests proxy server for urllib implementation"""
881
_transport = HttpTransport_urllib
884
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
886
TestCaseWithTwoWebservers):
887
"""Tests proxy server for pycurl implementation"""
890
TestProxyHttpServer.setUp(self)
891
# Oh my ! pycurl does not check for the port as part of
892
# no_proxy :-( So we just test the host part
893
self.no_proxy_host = 'localhost'
895
def test_HTTP_PROXY(self):
896
# pycurl do not check HTTP_PROXY for security reasons
897
# (for use in a CGI context that we do not care
898
# about. Should we ?)
901
def test_HTTP_PROXY_with_NO_PROXY(self):
904
def test_http_proxy_without_scheme(self):
905
# pycurl *ignores* invalid proxy env variables. If that
906
# ever change in the future, this test will fail
907
# indicating that pycurl do not ignore anymore such
909
self.not_proxied_in_env({'http_proxy': self.proxy_address})
912
class TestRanges(object):
913
"""Test the Range header in GET methods..
915
This MUST be used by daughter classes that also inherit from
916
TestCaseWithWebserver.
918
We can't inherit directly from TestCaseWithWebserver or the
919
test framework will try to create an instance which cannot
920
run, its implementation being incomplete.
924
TestCaseWithWebserver.setUp(self)
925
self.build_tree_contents([('a', '0123456789')],)
926
server = self.get_readonly_server()
927
self.transport = self._transport(server.get_url())
929
def _file_contents(self, relpath, ranges, tail_amount=0):
930
code, data = self.transport._get(relpath, ranges)
931
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
932
for start, end in ranges:
934
yield data.read(end - start + 1)
936
def _file_tail(self, relpath, tail_amount):
937
code, data = self.transport._get(relpath, [], tail_amount)
938
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
939
data.seek(-tail_amount + 1, 2)
940
return data.read(tail_amount)
942
def test_range_header(self):
944
map(self.assertEqual,['0', '234'],
945
list(self._file_contents('a', [(0,0), (2,4)])),)
947
self.assertEqual('789', self._file_tail('a', 3))
948
# Syntactically invalid range
949
self.assertRaises(errors.InvalidRange,
950
self.transport._get, 'a', [(4, 3)])
951
# Semantically invalid range
952
self.assertRaises(errors.InvalidRange,
953
self.transport._get, 'a', [(42, 128)])
956
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
957
"""Test the Range header in GET methods for urllib implementation"""
959
_transport = HttpTransport_urllib
962
class TestRanges_pycurl(TestWithTransport_pycurl,
964
TestCaseWithWebserver):
965
"""Test the Range header in GET methods for pycurl implementation"""
968
class TestHTTPRedirections(object):
969
"""Test redirection between http servers.
971
This MUST be used by daughter classes that also inherit from
972
TestCaseWithRedirectedWebserver.
974
We can't inherit directly from TestCaseWithTwoWebservers or the
975
test framework will try to create an instance which cannot
976
run, its implementation being incomplete.
979
def create_transport_secondary_server(self):
980
"""Create the secondary server redirecting to the primary server"""
981
new = self.get_readonly_server()
983
redirecting = HTTPServerRedirecting()
984
redirecting.redirect_to(new.host, new.port)
988
super(TestHTTPRedirections, self).setUp()
989
self.build_tree_contents([('a', '0123456789'),
991
'# Bazaar revision bundle v0.9\n#\n')
994
self.old_transport = self._transport(self.old_server.get_url())
996
def test_redirected(self):
997
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
998
t = self._transport(self.new_server.get_url())
999
self.assertEqual('0123456789', t.get('a').read())
1001
def test_read_redirected_bundle_from_url(self):
1002
from bzrlib.bundle import read_bundle_from_url
1003
url = self.old_transport.abspath('bundle')
1004
bundle = read_bundle_from_url(url)
1005
# If read_bundle_from_url was successful we get an empty bundle
1006
self.assertEqual([], bundle.revisions)
1009
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1010
TestCaseWithRedirectedWebserver):
1011
"""Tests redirections for urllib implementation"""
1013
_transport = HttpTransport_urllib
1017
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1018
TestHTTPRedirections,
1019
TestCaseWithRedirectedWebserver):
1020
"""Tests redirections for pycurl implementation"""
1023
class RedirectedRequest(Request):
1024
"""Request following redirections"""
1026
init_orig = Request.__init__
1028
def __init__(self, method, url, *args, **kwargs):
1029
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1030
self.follow_redirections = True
1033
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1034
"""Test redirections provided by urllib.
1036
http implementations do not redirect silently anymore (they
1037
do not redirect at all in fact). The mechanism is still in
1038
place at the _urllib2_wrappers.Request level and these tests
1041
For the pycurl implementation
1042
the redirection have been deleted as we may deprecate pycurl
1043
and I have no place to keep a working implementation.
1047
_transport = HttpTransport_urllib
1050
super(TestHTTPSilentRedirections_urllib, self).setUp()
1051
self.setup_redirected_request()
1052
self.addCleanup(self.cleanup_redirected_request)
1053
self.build_tree_contents([('a','a'),
1055
('1/a', 'redirected once'),
1057
('2/a', 'redirected twice'),
1059
('3/a', 'redirected thrice'),
1061
('4/a', 'redirected 4 times'),
1063
('5/a', 'redirected 5 times'),
1066
self.old_transport = self._transport(self.old_server.get_url())
1068
def setup_redirected_request(self):
1069
self.original_class = _urllib2_wrappers.Request
1070
_urllib2_wrappers.Request = RedirectedRequest
1072
def cleanup_redirected_request(self):
1073
_urllib2_wrappers.Request = self.original_class
1075
def create_transport_secondary_server(self):
1076
"""Create the secondary server, redirections are defined in the tests"""
1077
return HTTPServerRedirecting()
1079
def test_one_redirection(self):
1080
t = self.old_transport
1082
req = RedirectedRequest('GET', t.abspath('a'))
1083
req.follow_redirections = True
1084
new_prefix = 'http://%s:%s' % (self.new_server.host,
1085
self.new_server.port)
1086
self.old_server.redirections = \
1087
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1088
self.assertEquals('redirected once',t._perform(req).read())
1090
def test_five_redirections(self):
1091
t = self.old_transport
1093
req = RedirectedRequest('GET', t.abspath('a'))
1094
req.follow_redirections = True
1095
old_prefix = 'http://%s:%s' % (self.old_server.host,
1096
self.old_server.port)
1097
new_prefix = 'http://%s:%s' % (self.new_server.host,
1098
self.new_server.port)
1099
self.old_server.redirections = \
1100
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1101
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1102
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1103
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1104
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1106
self.assertEquals('redirected 5 times',t._perform(req).read())
1109
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1110
"""Test transport.do_catching_redirections.
1112
We arbitrarily choose to use urllib transports
1115
_transport = HttpTransport_urllib
1118
super(TestDoCatchRedirections, self).setUp()
1119
self.build_tree_contents([('a', '0123456789'),],)
1121
self.old_transport = self._transport(self.old_server.get_url())
1123
def get_a(self, transport):
1124
return transport.get('a')
1126
def test_no_redirection(self):
1127
t = self._transport(self.new_server.get_url())
1129
# We use None for redirected so that we fail if redirected
1130
self.assertEquals('0123456789',
1131
do_catching_redirections(self.get_a, t, None).read())
1133
def test_one_redirection(self):
1134
self.redirections = 0
1136
def redirected(transport, exception, redirection_notice):
1137
self.redirections += 1
1138
dir, file = urlutils.split(exception.target)
1139
return self._transport(dir)
1141
self.assertEquals('0123456789',
1142
do_catching_redirections(self.get_a,
1146
self.assertEquals(1, self.redirections)
1148
def test_redirection_loop(self):
1150
def redirected(transport, exception, redirection_notice):
1151
# By using the redirected url as a base dir for the
1152
# *old* transport, we create a loop: a => a/a =>
1154
return self.old_transport.clone(exception.target)
1156
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1157
self.get_a, self.old_transport, redirected)
1160
class TestAuth(object):
1161
"""Test some authentication scheme specified by daughter class.
1163
This MUST be used by daughter classes that also inherit from
1164
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1168
"""Set up the test environment
1170
Daughter classes should set up their own environment
1171
(including self.server) and explicitely call this
1172
method. This is needed because we want to reuse the same
1173
tests for proxy and no-proxy accesses which have
1174
different ways of setting self.server.
1176
self.build_tree_contents([('a', 'contents of a\n'),
1177
('b', 'contents of b\n'),])
1178
self.old_factory = ui.ui_factory
1179
# The following has the unfortunate side-effect of hiding any ouput
1180
# during the tests (including pdb prompts). Feel free to comment them
1181
# for debugging purposes but leave them in place, there are needed to
1182
# run the tests without any console
1183
self.old_stdout = sys.stdout
1184
sys.stdout = StringIOWrapper()
1185
self.addCleanup(self.restoreUIFactory)
1187
def restoreUIFactory(self):
1188
ui.ui_factory = self.old_factory
1189
sys.stdout = self.old_stdout
1191
def get_user_url(self, user=None, password=None):
1192
"""Build an url embedding user and password"""
1193
url = '%s://' % self.server._url_protocol
1194
if user is not None:
1196
if password is not None:
1197
url += ':' + password
1199
url += '%s:%s/' % (self.server.host, self.server.port)
1202
def test_no_user(self):
1203
self.server.add_user('joe', 'foo')
1204
t = self.get_user_transport()
1205
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1206
# Only one 'Authentication Required' error should occur
1207
self.assertEqual(1, self.server.auth_required_errors)
1209
def test_empty_pass(self):
1210
self.server.add_user('joe', '')
1211
t = self.get_user_transport('joe', '')
1212
self.assertEqual('contents of a\n', t.get('a').read())
1213
# Only one 'Authentication Required' error should occur
1214
self.assertEqual(1, self.server.auth_required_errors)
1216
def test_user_pass(self):
1217
self.server.add_user('joe', 'foo')
1218
t = self.get_user_transport('joe', 'foo')
1219
self.assertEqual('contents of a\n', t.get('a').read())
1220
# Only one 'Authentication Required' error should occur
1221
self.assertEqual(1, self.server.auth_required_errors)
1223
def test_unknown_user(self):
1224
self.server.add_user('joe', 'foo')
1225
t = self.get_user_transport('bill', 'foo')
1226
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1227
# Two 'Authentication Required' errors should occur (the
1228
# initial 'who are you' and 'I don't know you, who are
1230
self.assertEqual(2, self.server.auth_required_errors)
1232
def test_wrong_pass(self):
1233
self.server.add_user('joe', 'foo')
1234
t = self.get_user_transport('joe', 'bar')
1235
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1236
# Two 'Authentication Required' errors should occur (the
1237
# initial 'who are you' and 'this is not you, who are you')
1238
self.assertEqual(2, self.server.auth_required_errors)
1240
def test_prompt_for_password(self):
1241
self.server.add_user('joe', 'foo')
1242
t = self.get_user_transport('joe', None)
1243
ui.ui_factory = TestUIFactory(stdin='foo\n')
1244
self.assertEqual('contents of a\n',t.get('a').read())
1245
# stdin should be empty
1246
self.assertEqual('', ui.ui_factory.stdin.readline())
1247
# And we shouldn't prompt again for a different request
1248
# against the same transport.
1249
self.assertEqual('contents of b\n',t.get('b').read())
1251
# And neither against a clone
1252
self.assertEqual('contents of b\n',t2.get('b').read())
1253
# Only one 'Authentication Required' error should occur
1254
self.assertEqual(1, self.server.auth_required_errors)
1257
class TestHTTPAuth(TestAuth):
1258
"""Test HTTP authentication schemes.
1260
Daughter classes MUST inherit from TestCaseWithWebserver too.
1263
_auth_header = 'Authorization'
1266
TestCaseWithWebserver.setUp(self)
1267
self.server = self.get_readonly_server()
1268
TestAuth.setUp(self)
1270
def get_user_transport(self, user=None, password=None):
1271
return self._transport(self.get_user_url(user, password))
1274
class TestProxyAuth(TestAuth):
1275
"""Test proxy authentication schemes.
1277
Daughter classes MUST also inherit from TestCaseWithWebserver.
1279
_auth_header = 'Proxy-authorization'
1282
TestCaseWithWebserver.setUp(self)
1283
self.server = self.get_readonly_server()
1285
self.addCleanup(self._restore_env)
1286
TestAuth.setUp(self)
1287
# Override the contents to avoid false positives
1288
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1289
('b', 'not proxied contents of b\n'),
1290
('a-proxied', 'contents of a\n'),
1291
('b-proxied', 'contents of b\n'),
1294
def get_user_transport(self, user=None, password=None):
1295
self._install_env({'all_proxy': self.get_user_url(user, password)})
1296
return self._transport(self.server.get_url())
1298
def _install_env(self, env):
1299
for name, value in env.iteritems():
1300
self._old_env[name] = osutils.set_or_unset_env(name, value)
1302
def _restore_env(self):
1303
for name, value in self._old_env.iteritems():
1304
osutils.set_or_unset_env(name, value)
1307
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1308
"""Test http basic authentication scheme"""
1310
_transport = HttpTransport_urllib
1312
def create_transport_readonly_server(self):
1313
return HTTPBasicAuthServer()
1316
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1317
"""Test proxy basic authentication scheme"""
1319
_transport = HttpTransport_urllib
1321
def create_transport_readonly_server(self):
1322
return ProxyBasicAuthServer()
1325
class TestDigestAuth(object):
1326
"""Digest Authentication specific tests"""
1328
def test_changing_nonce(self):
1329
self.server.add_user('joe', 'foo')
1330
t = self.get_user_transport('joe', 'foo')
1331
self.assertEqual('contents of a\n', t.get('a').read())
1332
self.assertEqual('contents of b\n', t.get('b').read())
1333
# Only one 'Authentication Required' error should have
1335
self.assertEqual(1, self.server.auth_required_errors)
1336
# The server invalidates the current nonce
1337
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1338
self.assertEqual('contents of a\n', t.get('a').read())
1339
# Two 'Authentication Required' errors should occur (the
1340
# initial 'who are you' and a second 'who are you' with the new nonce)
1341
self.assertEqual(2, self.server.auth_required_errors)
1344
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1345
"""Test http digest authentication scheme"""
1347
_transport = HttpTransport_urllib
1349
def create_transport_readonly_server(self):
1350
return HTTPDigestAuthServer()
1353
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1354
TestCaseWithWebserver):
1355
"""Test proxy digest authentication scheme"""
1357
_transport = HttpTransport_urllib
1359
def create_transport_readonly_server(self):
1360
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
