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((errors.InvalidURL, errors.ConnectionError),
207
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
209
def test_http_root_urls(self):
210
"""Construction of URLs from server root"""
211
t = self._transport('http://bzr.ozlabs.org/')
212
eq = self.assertEqualDiff
213
eq(t.abspath('.bzr/tree-version'),
214
'http://bzr.ozlabs.org/.bzr/tree-version')
216
def test_http_impl_urls(self):
217
"""There are servers which ask for particular clients to connect"""
218
server = self._server()
221
url = server.get_url()
222
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
227
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
228
"""Test http urls with urllib"""
230
_transport = HttpTransport_urllib
231
_server = HttpServer_urllib
232
_qualified_prefix = 'http+urllib'
235
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
237
"""Test http urls with pycurl"""
239
_server = HttpServer_PyCurl
240
_qualified_prefix = 'http+pycurl'
242
# TODO: This should really be moved into another pycurl
243
# specific test. When https tests will be implemented, take
244
# this one into account.
245
def test_pycurl_without_https_support(self):
246
"""Test that pycurl without SSL do not fail with a traceback.
248
For the purpose of the test, we force pycurl to ignore
249
https by supplying a fake version_info that do not
255
raise TestSkipped('pycurl not present')
256
# Now that we have pycurl imported, we can fake its version_info
257
# This was taken from a windows pycurl without SSL
259
pycurl.version_info = lambda : (2,
267
('ftp', 'gopher', 'telnet',
268
'dict', 'ldap', 'http', 'file'),
272
self.assertRaises(errors.DependencyNotPresent, self._transport,
273
'https://launchpad.net')
275
class TestHttpConnections(object):
276
"""Test the http connections.
278
This MUST be used by daughter classes that also inherit from
279
TestCaseWithWebserver.
281
We can't inherit directly from TestCaseWithWebserver or the
282
test framework will try to create an instance which cannot
283
run, its implementation being incomplete.
287
TestCaseWithWebserver.setUp(self)
288
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
289
transport=self.get_transport())
291
def test_http_has(self):
292
server = self.get_readonly_server()
293
t = self._transport(server.get_url())
294
self.assertEqual(t.has('foo/bar'), True)
295
self.assertEqual(len(server.logs), 1)
296
self.assertContainsRe(server.logs[0],
297
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
299
def test_http_has_not_found(self):
300
server = self.get_readonly_server()
301
t = self._transport(server.get_url())
302
self.assertEqual(t.has('not-found'), False)
303
self.assertContainsRe(server.logs[1],
304
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
306
def test_http_get(self):
307
server = self.get_readonly_server()
308
t = self._transport(server.get_url())
309
fp = t.get('foo/bar')
310
self.assertEqualDiff(
312
'contents of foo/bar\n')
313
self.assertEqual(len(server.logs), 1)
314
self.assertTrue(server.logs[0].find(
315
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
316
% bzrlib.__version__) > -1)
318
def test_get_smart_medium(self):
319
# For HTTP, get_smart_medium should return the transport object.
320
server = self.get_readonly_server()
321
http_transport = self._transport(server.get_url())
322
medium = http_transport.get_smart_medium()
323
self.assertIs(medium, http_transport)
325
def test_has_on_bogus_host(self):
326
# Get a free address and don't 'accept' on it, so that we
327
# can be sure there is no http handler there, but set a
328
# reasonable timeout to not slow down tests too much.
329
default_timeout = socket.getdefaulttimeout()
331
socket.setdefaulttimeout(2)
333
s.bind(('localhost', 0))
334
t = self._transport('http://%s:%s/' % s.getsockname())
335
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
337
socket.setdefaulttimeout(default_timeout)
340
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
341
"""Test http connections with urllib"""
343
_transport = HttpTransport_urllib
347
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
349
TestCaseWithWebserver):
350
"""Test http connections with pycurl"""
353
class TestHttpTransportRegistration(TestCase):
354
"""Test registrations of various http implementations"""
356
def test_http_registered(self):
357
# urlllib should always be present
358
t = get_transport('http+urllib://bzr.google.com/')
359
self.assertIsInstance(t, Transport)
360
self.assertIsInstance(t, HttpTransport_urllib)
363
class TestOffsets(TestCase):
364
"""Test offsets_to_ranges method"""
366
def test_offsets_to_ranges_simple(self):
367
to_range = HttpTransportBase.offsets_to_ranges
368
ranges = to_range([(10, 1)])
369
self.assertEqual([[10, 10]], ranges)
371
ranges = to_range([(0, 1), (1, 1)])
372
self.assertEqual([[0, 1]], ranges)
374
ranges = to_range([(1, 1), (0, 1)])
375
self.assertEqual([[0, 1]], ranges)
377
def test_offset_to_ranges_overlapped(self):
378
to_range = HttpTransportBase.offsets_to_ranges
380
ranges = to_range([(10, 1), (20, 2), (22, 5)])
381
self.assertEqual([[10, 10], [20, 26]], ranges)
383
ranges = to_range([(10, 1), (11, 2), (22, 5)])
384
self.assertEqual([[10, 12], [22, 26]], ranges)
387
class TestPost(object):
389
def _test_post_body_is_received(self, scheme):
390
server = RecordingServer(expect_body_tail='end-of-body')
392
self.addCleanup(server.tearDown)
393
url = '%s://%s:%s/' % (scheme, server.host, server.port)
395
http_transport = get_transport(url)
396
except errors.UnsupportedProtocol:
397
raise TestSkipped('%s not available' % scheme)
398
code, response = http_transport._post('abc def end-of-body')
400
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
401
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
402
# The transport should not be assuming that the server can accept
403
# chunked encoding the first time it connects, because HTTP/1.1, so we
404
# check for the literal string.
406
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
409
class TestPost_urllib(TestCase, TestPost):
410
"""TestPost for urllib implementation"""
412
_transport = HttpTransport_urllib
414
def test_post_body_is_received_urllib(self):
415
self._test_post_body_is_received('http+urllib')
418
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
419
"""TestPost for pycurl implementation"""
421
def test_post_body_is_received_pycurl(self):
422
self._test_post_body_is_received('http+pycurl')
425
class TestRangeHeader(TestCase):
426
"""Test range_header method"""
428
def check_header(self, value, ranges=[], tail=0):
429
range_header = HttpTransportBase.range_header
430
self.assertEqual(value, range_header(ranges, tail))
432
def test_range_header_single(self):
433
self.check_header('0-9', ranges=[[0,9]])
434
self.check_header('100-109', ranges=[[100,109]])
436
def test_range_header_tail(self):
437
self.check_header('-10', tail=10)
438
self.check_header('-50', tail=50)
440
def test_range_header_multi(self):
441
self.check_header('0-9,100-200,300-5000',
442
ranges=[(0,9), (100, 200), (300,5000)])
444
def test_range_header_mixed(self):
445
self.check_header('0-9,300-5000,-50',
446
ranges=[(0,9), (300,5000)],
450
class TestWallServer(object):
451
"""Tests exceptions during the connection phase"""
453
def create_transport_readonly_server(self):
454
return HttpServer(WallRequestHandler)
456
def test_http_has(self):
457
server = self.get_readonly_server()
458
t = self._transport(server.get_url())
459
# Unfortunately httplib (see HTTPResponse._read_status
460
# for details) make no distinction between a closed
461
# socket and badly formatted status line, so we can't
462
# just test for ConnectionError, we have to test
463
# InvalidHttpResponse too.
464
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
467
def test_http_get(self):
468
server = self.get_readonly_server()
469
t = self._transport(server.get_url())
470
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
474
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
475
"""Tests "wall" server for urllib implementation"""
477
_transport = HttpTransport_urllib
480
class TestWallServer_pycurl(TestWithTransport_pycurl,
482
TestCaseWithWebserver):
483
"""Tests "wall" server for pycurl implementation"""
486
class TestBadStatusServer(object):
487
"""Tests bad status from server."""
489
def create_transport_readonly_server(self):
490
return HttpServer(BadStatusRequestHandler)
492
def test_http_has(self):
493
server = self.get_readonly_server()
494
t = self._transport(server.get_url())
495
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
497
def test_http_get(self):
498
server = self.get_readonly_server()
499
t = self._transport(server.get_url())
500
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
503
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
504
"""Tests bad status server for urllib implementation"""
506
_transport = HttpTransport_urllib
509
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
511
TestCaseWithWebserver):
512
"""Tests bad status server for pycurl implementation"""
515
class TestInvalidStatusServer(TestBadStatusServer):
516
"""Tests invalid status from server.
518
Both implementations raises the same error as for a bad status.
521
def create_transport_readonly_server(self):
522
return HttpServer(InvalidStatusRequestHandler)
525
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
526
TestCaseWithWebserver):
527
"""Tests invalid status server for urllib implementation"""
529
_transport = HttpTransport_urllib
532
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
533
TestInvalidStatusServer,
534
TestCaseWithWebserver):
535
"""Tests invalid status server for pycurl implementation"""
538
class TestBadProtocolServer(object):
539
"""Tests bad protocol from server."""
541
def create_transport_readonly_server(self):
542
return HttpServer(BadProtocolRequestHandler)
544
def test_http_has(self):
545
server = self.get_readonly_server()
546
t = self._transport(server.get_url())
547
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
549
def test_http_get(self):
550
server = self.get_readonly_server()
551
t = self._transport(server.get_url())
552
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
555
class TestBadProtocolServer_urllib(TestBadProtocolServer,
556
TestCaseWithWebserver):
557
"""Tests bad protocol server for urllib implementation"""
559
_transport = HttpTransport_urllib
561
# curl don't check the protocol version
562
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
563
# TestBadProtocolServer,
564
# TestCaseWithWebserver):
565
# """Tests bad protocol server for pycurl implementation"""
568
class TestForbiddenServer(object):
569
"""Tests forbidden server"""
571
def create_transport_readonly_server(self):
572
return HttpServer(ForbiddenRequestHandler)
574
def test_http_has(self):
575
server = self.get_readonly_server()
576
t = self._transport(server.get_url())
577
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
579
def test_http_get(self):
580
server = self.get_readonly_server()
581
t = self._transport(server.get_url())
582
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
585
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
586
"""Tests forbidden server for urllib implementation"""
588
_transport = HttpTransport_urllib
591
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
593
TestCaseWithWebserver):
594
"""Tests forbidden server for pycurl implementation"""
597
class TestRecordingServer(TestCase):
599
def test_create(self):
600
server = RecordingServer(expect_body_tail=None)
601
self.assertEqual('', server.received_bytes)
602
self.assertEqual(None, server.host)
603
self.assertEqual(None, server.port)
605
def test_setUp_and_tearDown(self):
606
server = RecordingServer(expect_body_tail=None)
609
self.assertNotEqual(None, server.host)
610
self.assertNotEqual(None, server.port)
613
self.assertEqual(None, server.host)
614
self.assertEqual(None, server.port)
616
def test_send_receive_bytes(self):
617
server = RecordingServer(expect_body_tail='c')
619
self.addCleanup(server.tearDown)
620
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
621
sock.connect((server.host, server.port))
623
self.assertEqual('HTTP/1.1 200 OK\r\n',
624
osutils.recv_all(sock, 4096))
625
self.assertEqual('abc', server.received_bytes)
628
class TestRangeRequestServer(object):
629
"""Tests readv requests against server.
631
This MUST be used by daughter classes that also inherit from
632
TestCaseWithWebserver.
634
We can't inherit directly from TestCaseWithWebserver or the
635
test framework will try to create an instance which cannot
636
run, its implementation being incomplete.
640
TestCaseWithWebserver.setUp(self)
641
self.build_tree_contents([('a', '0123456789')],)
643
def test_readv(self):
644
server = self.get_readonly_server()
645
t = self._transport(server.get_url())
646
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
647
self.assertEqual(l[0], (0, '0'))
648
self.assertEqual(l[1], (1, '1'))
649
self.assertEqual(l[2], (3, '34'))
650
self.assertEqual(l[3], (9, '9'))
652
def test_readv_out_of_order(self):
653
server = self.get_readonly_server()
654
t = self._transport(server.get_url())
655
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
656
self.assertEqual(l[0], (1, '1'))
657
self.assertEqual(l[1], (9, '9'))
658
self.assertEqual(l[2], (0, '0'))
659
self.assertEqual(l[3], (3, '34'))
661
def test_readv_invalid_ranges(self):
662
server = self.get_readonly_server()
663
t = self._transport(server.get_url())
665
# This is intentionally reading off the end of the file
666
# since we are sure that it cannot get there
667
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
668
t.readv, 'a', [(1,1), (8,10)])
670
# This is trying to seek past the end of the file, it should
671
# also raise a special error
672
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
673
t.readv, 'a', [(12,2)])
676
class TestSingleRangeRequestServer(TestRangeRequestServer):
677
"""Test readv against a server which accept only single range requests"""
679
def create_transport_readonly_server(self):
680
return HttpServer(SingleRangeRequestHandler)
683
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
684
TestCaseWithWebserver):
685
"""Tests single range requests accepting server for urllib implementation"""
687
_transport = HttpTransport_urllib
690
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
691
TestSingleRangeRequestServer,
692
TestCaseWithWebserver):
693
"""Tests single range requests accepting server for pycurl implementation"""
696
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
697
"""Test readv against a server which only accept single range requests"""
699
def create_transport_readonly_server(self):
700
return HttpServer(SingleOnlyRangeRequestHandler)
703
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
704
TestCaseWithWebserver):
705
"""Tests single range requests accepting server for urllib implementation"""
707
_transport = HttpTransport_urllib
710
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
711
TestSingleOnlyRangeRequestServer,
712
TestCaseWithWebserver):
713
"""Tests single range requests accepting server for pycurl implementation"""
716
class TestNoRangeRequestServer(TestRangeRequestServer):
717
"""Test readv against a server which do not accept range requests"""
719
def create_transport_readonly_server(self):
720
return HttpServer(NoRangeRequestHandler)
723
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
724
TestCaseWithWebserver):
725
"""Tests range requests refusing server for urllib implementation"""
727
_transport = HttpTransport_urllib
730
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
731
TestNoRangeRequestServer,
732
TestCaseWithWebserver):
733
"""Tests range requests refusing server for pycurl implementation"""
736
class TestHttpProxyWhiteBox(TestCase):
737
"""Whitebox test proxy http authorization.
739
Only the urllib implementation is tested here.
749
def _install_env(self, env):
750
for name, value in env.iteritems():
751
self._old_env[name] = osutils.set_or_unset_env(name, value)
753
def _restore_env(self):
754
for name, value in self._old_env.iteritems():
755
osutils.set_or_unset_env(name, value)
757
def _proxied_request(self):
758
handler = ProxyHandler(PasswordManager())
759
request = Request('GET','http://baz/buzzle')
760
handler.set_proxy(request, 'http')
763
def test_empty_user(self):
764
self._install_env({'http_proxy': 'http://bar.com'})
765
request = self._proxied_request()
766
self.assertFalse(request.headers.has_key('Proxy-authorization'))
768
def test_invalid_proxy(self):
769
"""A proxy env variable without scheme"""
770
self._install_env({'http_proxy': 'host:1234'})
771
self.assertRaises(errors.InvalidURL, self._proxied_request)
774
class TestProxyHttpServer(object):
775
"""Tests proxy server.
777
This MUST be used by daughter classes that also inherit from
778
TestCaseWithTwoWebservers.
780
We can't inherit directly from TestCaseWithTwoWebservers or
781
the test framework will try to create an instance which
782
cannot run, its implementation being incomplete.
784
Be aware that we do not setup a real proxy here. Instead, we
785
check that the *connection* goes through the proxy by serving
786
different content (the faked proxy server append '-proxied'
790
# FIXME: We don't have an https server available, so we don't
791
# test https connections.
793
# FIXME: Once the test suite is better fitted to test
794
# authorization schemes, test proxy authorizations too (see
798
TestCaseWithTwoWebservers.setUp(self)
799
self.build_tree_contents([('foo', 'contents of foo\n'),
800
('foo-proxied', 'proxied contents of foo\n')])
801
# Let's setup some attributes for tests
802
self.server = self.get_readonly_server()
803
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
804
self.no_proxy_host = self.proxy_address
805
# The secondary server is the proxy
806
self.proxy = self.get_secondary_server()
807
self.proxy_url = self.proxy.get_url()
810
def create_transport_secondary_server(self):
811
"""Creates an http server that will serve files with
812
'-proxied' appended to their names.
816
def _install_env(self, env):
817
for name, value in env.iteritems():
818
self._old_env[name] = osutils.set_or_unset_env(name, value)
820
def _restore_env(self):
821
for name, value in self._old_env.iteritems():
822
osutils.set_or_unset_env(name, value)
824
def 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(), 'proxied contents of foo\n')
833
def not_proxied_in_env(self, env):
834
self._install_env(env)
835
url = self.server.get_url()
836
t = self._transport(url)
838
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
842
def test_http_proxy(self):
843
self.proxied_in_env({'http_proxy': self.proxy_url})
845
def test_HTTP_PROXY(self):
846
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
848
def test_all_proxy(self):
849
self.proxied_in_env({'all_proxy': self.proxy_url})
851
def test_ALL_PROXY(self):
852
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
854
def test_http_proxy_with_no_proxy(self):
855
self.not_proxied_in_env({'http_proxy': self.proxy_url,
856
'no_proxy': self.no_proxy_host})
858
def test_HTTP_PROXY_with_NO_PROXY(self):
859
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
860
'NO_PROXY': self.no_proxy_host})
862
def test_all_proxy_with_no_proxy(self):
863
self.not_proxied_in_env({'all_proxy': self.proxy_url,
864
'no_proxy': self.no_proxy_host})
866
def test_ALL_PROXY_with_NO_PROXY(self):
867
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
868
'NO_PROXY': self.no_proxy_host})
870
def test_http_proxy_without_scheme(self):
871
self.assertRaises(errors.InvalidURL,
873
{'http_proxy': self.proxy_address})
876
class TestProxyHttpServer_urllib(TestProxyHttpServer,
877
TestCaseWithTwoWebservers):
878
"""Tests proxy server for urllib implementation"""
880
_transport = HttpTransport_urllib
883
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
885
TestCaseWithTwoWebservers):
886
"""Tests proxy server for pycurl implementation"""
889
TestProxyHttpServer.setUp(self)
890
# Oh my ! pycurl does not check for the port as part of
891
# no_proxy :-( So we just test the host part
892
self.no_proxy_host = 'localhost'
894
def test_HTTP_PROXY(self):
895
# pycurl do not check HTTP_PROXY for security reasons
896
# (for use in a CGI context that we do not care
897
# about. Should we ?)
900
def test_HTTP_PROXY_with_NO_PROXY(self):
903
def test_http_proxy_without_scheme(self):
904
# pycurl *ignores* invalid proxy env variables. If that
905
# ever change in the future, this test will fail
906
# indicating that pycurl do not ignore anymore such
908
self.not_proxied_in_env({'http_proxy': self.proxy_address})
911
class TestRanges(object):
912
"""Test the Range header in GET methods..
914
This MUST be used by daughter classes that also inherit from
915
TestCaseWithWebserver.
917
We can't inherit directly from TestCaseWithWebserver or the
918
test framework will try to create an instance which cannot
919
run, its implementation being incomplete.
923
TestCaseWithWebserver.setUp(self)
924
self.build_tree_contents([('a', '0123456789')],)
925
server = self.get_readonly_server()
926
self.transport = self._transport(server.get_url())
928
def _file_contents(self, relpath, ranges, tail_amount=0):
929
code, data = self.transport._get(relpath, ranges)
930
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
931
for start, end in ranges:
933
yield data.read(end - start + 1)
935
def _file_tail(self, relpath, tail_amount):
936
code, data = self.transport._get(relpath, [], tail_amount)
937
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
938
data.seek(-tail_amount + 1, 2)
939
return data.read(tail_amount)
941
def test_range_header(self):
943
map(self.assertEqual,['0', '234'],
944
list(self._file_contents('a', [(0,0), (2,4)])),)
946
self.assertEqual('789', self._file_tail('a', 3))
947
# Syntactically invalid range
948
self.assertRaises(errors.InvalidRange,
949
self.transport._get, 'a', [(4, 3)])
950
# Semantically invalid range
951
self.assertRaises(errors.InvalidRange,
952
self.transport._get, 'a', [(42, 128)])
955
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
956
"""Test the Range header in GET methods for urllib implementation"""
958
_transport = HttpTransport_urllib
961
class TestRanges_pycurl(TestWithTransport_pycurl,
963
TestCaseWithWebserver):
964
"""Test the Range header in GET methods for pycurl implementation"""
967
class TestHTTPRedirections(object):
968
"""Test redirection between http servers.
970
This MUST be used by daughter classes that also inherit from
971
TestCaseWithRedirectedWebserver.
973
We can't inherit directly from TestCaseWithTwoWebservers or the
974
test framework will try to create an instance which cannot
975
run, its implementation being incomplete.
978
def create_transport_secondary_server(self):
979
"""Create the secondary server redirecting to the primary server"""
980
new = self.get_readonly_server()
982
redirecting = HTTPServerRedirecting()
983
redirecting.redirect_to(new.host, new.port)
987
super(TestHTTPRedirections, self).setUp()
988
self.build_tree_contents([('a', '0123456789'),
990
'# Bazaar revision bundle v0.9\n#\n')
993
self.old_transport = self._transport(self.old_server.get_url())
995
def test_redirected(self):
996
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
997
t = self._transport(self.new_server.get_url())
998
self.assertEqual('0123456789', t.get('a').read())
1000
def test_read_redirected_bundle_from_url(self):
1001
from bzrlib.bundle import read_bundle_from_url
1002
url = self.old_transport.abspath('bundle')
1003
bundle = read_bundle_from_url(url)
1004
# If read_bundle_from_url was successful we get an empty bundle
1005
self.assertEqual([], bundle.revisions)
1008
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1009
TestCaseWithRedirectedWebserver):
1010
"""Tests redirections for urllib implementation"""
1012
_transport = HttpTransport_urllib
1016
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1017
TestHTTPRedirections,
1018
TestCaseWithRedirectedWebserver):
1019
"""Tests redirections for pycurl implementation"""
1022
class RedirectedRequest(Request):
1023
"""Request following redirections"""
1025
init_orig = Request.__init__
1027
def __init__(self, method, url, *args, **kwargs):
1028
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1029
self.follow_redirections = True
1032
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1033
"""Test redirections provided by urllib.
1035
http implementations do not redirect silently anymore (they
1036
do not redirect at all in fact). The mechanism is still in
1037
place at the _urllib2_wrappers.Request level and these tests
1040
For the pycurl implementation
1041
the redirection have been deleted as we may deprecate pycurl
1042
and I have no place to keep a working implementation.
1046
_transport = HttpTransport_urllib
1049
super(TestHTTPSilentRedirections_urllib, self).setUp()
1050
self.setup_redirected_request()
1051
self.addCleanup(self.cleanup_redirected_request)
1052
self.build_tree_contents([('a','a'),
1054
('1/a', 'redirected once'),
1056
('2/a', 'redirected twice'),
1058
('3/a', 'redirected thrice'),
1060
('4/a', 'redirected 4 times'),
1062
('5/a', 'redirected 5 times'),
1065
self.old_transport = self._transport(self.old_server.get_url())
1067
def setup_redirected_request(self):
1068
self.original_class = _urllib2_wrappers.Request
1069
_urllib2_wrappers.Request = RedirectedRequest
1071
def cleanup_redirected_request(self):
1072
_urllib2_wrappers.Request = self.original_class
1074
def create_transport_secondary_server(self):
1075
"""Create the secondary server, redirections are defined in the tests"""
1076
return HTTPServerRedirecting()
1078
def test_one_redirection(self):
1079
t = self.old_transport
1081
req = RedirectedRequest('GET', t.abspath('a'))
1082
req.follow_redirections = True
1083
new_prefix = 'http://%s:%s' % (self.new_server.host,
1084
self.new_server.port)
1085
self.old_server.redirections = \
1086
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1087
self.assertEquals('redirected once',t._perform(req).read())
1089
def test_five_redirections(self):
1090
t = self.old_transport
1092
req = RedirectedRequest('GET', t.abspath('a'))
1093
req.follow_redirections = True
1094
old_prefix = 'http://%s:%s' % (self.old_server.host,
1095
self.old_server.port)
1096
new_prefix = 'http://%s:%s' % (self.new_server.host,
1097
self.new_server.port)
1098
self.old_server.redirections = \
1099
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1100
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1101
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1102
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1103
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1105
self.assertEquals('redirected 5 times',t._perform(req).read())
1108
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1109
"""Test transport.do_catching_redirections.
1111
We arbitrarily choose to use urllib transports
1114
_transport = HttpTransport_urllib
1117
super(TestDoCatchRedirections, self).setUp()
1118
self.build_tree_contents([('a', '0123456789'),],)
1120
self.old_transport = self._transport(self.old_server.get_url())
1122
def get_a(self, transport):
1123
return transport.get('a')
1125
def test_no_redirection(self):
1126
t = self._transport(self.new_server.get_url())
1128
# We use None for redirected so that we fail if redirected
1129
self.assertEquals('0123456789',
1130
do_catching_redirections(self.get_a, t, None).read())
1132
def test_one_redirection(self):
1133
self.redirections = 0
1135
def redirected(transport, exception, redirection_notice):
1136
self.redirections += 1
1137
dir, file = urlutils.split(exception.target)
1138
return self._transport(dir)
1140
self.assertEquals('0123456789',
1141
do_catching_redirections(self.get_a,
1145
self.assertEquals(1, self.redirections)
1147
def test_redirection_loop(self):
1149
def redirected(transport, exception, redirection_notice):
1150
# By using the redirected url as a base dir for the
1151
# *old* transport, we create a loop: a => a/a =>
1153
return self.old_transport.clone(exception.target)
1155
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1156
self.get_a, self.old_transport, redirected)
1159
class TestAuth(object):
1160
"""Test some authentication scheme specified by daughter class.
1162
This MUST be used by daughter classes that also inherit from
1163
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1167
"""Set up the test environment
1169
Daughter classes should set up their own environment
1170
(including self.server) and explicitely call this
1171
method. This is needed because we want to reuse the same
1172
tests for proxy and no-proxy accesses which have
1173
different ways of setting self.server.
1175
self.build_tree_contents([('a', 'contents of a\n'),
1176
('b', 'contents of b\n'),])
1177
self.old_factory = ui.ui_factory
1178
self.old_stdout = sys.stdout
1179
sys.stdout = StringIOWrapper()
1180
self.addCleanup(self.restoreUIFactory)
1182
def restoreUIFactory(self):
1183
ui.ui_factory = self.old_factory
1184
sys.stdout = self.old_stdout
1186
def get_user_url(self, user=None, password=None):
1187
"""Build an url embedding user and password"""
1188
url = '%s://' % self.server._url_protocol
1189
if user is not None:
1191
if password is not None:
1192
url += ':' + password
1194
url += '%s:%s/' % (self.server.host, self.server.port)
1197
def test_no_user(self):
1198
self.server.add_user('joe', 'foo')
1199
t = self.get_user_transport()
1200
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1201
# Only one 'Authentication Required' error should occur
1202
self.assertEqual(1, self.server.auth_required_errors)
1204
def test_empty_pass(self):
1205
self.server.add_user('joe', '')
1206
t = self.get_user_transport('joe', '')
1207
self.assertEqual('contents of a\n', t.get('a').read())
1208
# Only one 'Authentication Required' error should occur
1209
self.assertEqual(1, self.server.auth_required_errors)
1211
def test_user_pass(self):
1212
self.server.add_user('joe', 'foo')
1213
t = self.get_user_transport('joe', 'foo')
1214
self.assertEqual('contents of a\n', t.get('a').read())
1215
# Only one 'Authentication Required' error should occur
1216
self.assertEqual(1, self.server.auth_required_errors)
1218
def test_unknown_user(self):
1219
self.server.add_user('joe', 'foo')
1220
t = self.get_user_transport('bill', 'foo')
1221
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1222
# Two 'Authentication Required' errors should occur (the
1223
# initial 'who are you' and 'I don't know you, who are
1225
self.assertEqual(2, self.server.auth_required_errors)
1227
def test_wrong_pass(self):
1228
self.server.add_user('joe', 'foo')
1229
t = self.get_user_transport('joe', 'bar')
1230
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1231
# Two 'Authentication Required' errors should occur (the
1232
# initial 'who are you' and 'this is not you, who are you')
1233
self.assertEqual(2, self.server.auth_required_errors)
1235
def test_prompt_for_password(self):
1236
self.server.add_user('joe', 'foo')
1237
t = self.get_user_transport('joe', None)
1238
ui.ui_factory = TestUIFactory(stdin='foo\n')
1239
self.assertEqual('contents of a\n',t.get('a').read())
1240
# stdin should be empty
1241
self.assertEqual('', ui.ui_factory.stdin.readline())
1242
# And we shouldn't prompt again for a different request
1243
# against the same transport.
1244
self.assertEqual('contents of b\n',t.get('b').read())
1246
# And neither against a clone
1247
self.assertEqual('contents of b\n',t2.get('b').read())
1248
# Only one 'Authentication Required' error should occur
1249
self.assertEqual(1, self.server.auth_required_errors)
1252
class TestHTTPAuth(TestAuth):
1253
"""Test HTTP authentication schemes.
1255
Daughter classes MUST inherit from TestCaseWithWebserver too.
1258
_auth_header = 'Authorization'
1261
TestCaseWithWebserver.setUp(self)
1262
self.server = self.get_readonly_server()
1263
TestAuth.setUp(self)
1265
def get_user_transport(self, user=None, password=None):
1266
return self._transport(self.get_user_url(user, password))
1269
class TestProxyAuth(TestAuth):
1270
"""Test proxy authentication schemes.
1272
Daughter classes MUST also inherit from TestCaseWithWebserver.
1274
_auth_header = 'Proxy-authorization'
1277
TestCaseWithWebserver.setUp(self)
1278
self.server = self.get_readonly_server()
1280
self.addCleanup(self._restore_env)
1281
TestAuth.setUp(self)
1282
# Override the contents to avoid false positives
1283
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1284
('b', 'not proxied contents of b\n'),
1285
('a-proxied', 'contents of a\n'),
1286
('b-proxied', 'contents of b\n'),
1289
def get_user_transport(self, user=None, password=None):
1290
self._install_env({'all_proxy': self.get_user_url(user, password)})
1291
return self._transport(self.server.get_url())
1293
def _install_env(self, env):
1294
for name, value in env.iteritems():
1295
self._old_env[name] = osutils.set_or_unset_env(name, value)
1297
def _restore_env(self):
1298
for name, value in self._old_env.iteritems():
1299
osutils.set_or_unset_env(name, value)
1302
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1303
"""Test http basic authentication scheme"""
1305
_transport = HttpTransport_urllib
1307
def create_transport_readonly_server(self):
1308
return HTTPBasicAuthServer()
1311
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1312
"""Test proxy basic authentication scheme"""
1314
_transport = HttpTransport_urllib
1316
def create_transport_readonly_server(self):
1317
return ProxyBasicAuthServer()
1320
class TestDigestAuth(object):
1321
"""Digest Authentication specific tests"""
1323
def test_changing_nonce(self):
1324
self.server.add_user('joe', 'foo')
1325
t = self.get_user_transport('joe', 'foo')
1326
self.assertEqual('contents of a\n', t.get('a').read())
1327
self.assertEqual('contents of b\n', t.get('b').read())
1328
# Only one 'Authentication Required' error should have
1330
self.assertEqual(1, self.server.auth_required_errors)
1331
# The server invalidates the current nonce
1332
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1333
self.assertEqual('contents of a\n', t.get('a').read())
1334
# Two 'Authentication Required' errors should occur (the
1335
# initial 'who are you' and a second 'who are you' with the new nonce)
1336
self.assertEqual(2, self.server.auth_required_errors)
1339
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1340
"""Test http digest authentication scheme"""
1342
_transport = HttpTransport_urllib
1344
def create_transport_readonly_server(self):
1345
return HTTPDigestAuthServer()
1348
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1349
TestCaseWithWebserver):
1350
"""Test proxy digest authentication scheme"""
1352
_transport = HttpTransport_urllib
1354
def create_transport_readonly_server(self):
1355
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
