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
39
from bzrlib.tests.HttpServer import (
44
from bzrlib.tests.HTTPTestUtil import (
45
BadProtocolRequestHandler,
46
BadStatusRequestHandler,
47
ForbiddenRequestHandler,
50
HTTPServerRedirecting,
51
InvalidStatusRequestHandler,
52
LimitedRangeHTTPServer,
53
NoRangeRequestHandler,
55
ProxyDigestAuthServer,
57
SingleRangeRequestHandler,
58
SingleOnlyRangeRequestHandler,
59
TestCaseWithRedirectedWebserver,
60
TestCaseWithTwoWebservers,
61
TestCaseWithWebserver,
64
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 (
82
class FakeManager(object):
87
def add_password(self, realm, host, username, password):
88
self.credentials.append([realm, host, username, password])
91
class RecordingServer(object):
92
"""A fake HTTP server.
94
It records the bytes sent to it, and replies with a 200.
97
def __init__(self, expect_body_tail=None):
100
:type expect_body_tail: str
101
:param expect_body_tail: a reply won't be sent until this string is
104
self._expect_body_tail = expect_body_tail
107
self.received_bytes = ''
110
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
111
self._sock.bind(('127.0.0.1', 0))
112
self.host, self.port = self._sock.getsockname()
113
self._ready = threading.Event()
114
self._thread = threading.Thread(target=self._accept_read_and_reply)
115
self._thread.setDaemon(True)
119
def _accept_read_and_reply(self):
122
self._sock.settimeout(5)
124
conn, address = self._sock.accept()
125
# On win32, the accepted connection will be non-blocking to start
126
# with because we're using settimeout.
127
conn.setblocking(True)
128
while not self.received_bytes.endswith(self._expect_body_tail):
129
self.received_bytes += conn.recv(4096)
130
conn.sendall('HTTP/1.1 200 OK\r\n')
131
except socket.timeout:
132
# Make sure the client isn't stuck waiting for us to e.g. accept.
135
# The client may have already closed the socket.
142
# We might have already closed it. We don't care.
148
class TestWithTransport_pycurl(object):
149
"""Test case to inherit from if pycurl is present"""
151
def _get_pycurl_maybe(self):
153
from bzrlib.transport.http._pycurl import PyCurlTransport
154
return PyCurlTransport
155
except errors.DependencyNotPresent:
156
raise tests.TestSkipped('pycurl not present')
158
_transport = property(_get_pycurl_maybe)
161
class TestHttpUrls(tests.TestCase):
163
# TODO: This should be moved to authorization tests once they
166
def test_url_parsing(self):
168
url = extract_auth('http://example.com', f)
169
self.assertEquals('http://example.com', url)
170
self.assertEquals(0, len(f.credentials))
171
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
172
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
173
self.assertEquals(1, len(f.credentials))
174
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
178
class TestHttpTransportUrls(object):
179
"""Test the http urls.
181
This MUST be used by daughter classes that also inherit from
184
We can't inherit directly from TestCase or the
185
test framework will try to create an instance which cannot
186
run, its implementation being incomplete.
189
def test_abs_url(self):
190
"""Construction of absolute http URLs"""
191
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
192
eq = self.assertEqualDiff
193
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
194
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
195
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
196
eq(t.abspath('.bzr/1//2/./3'),
197
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
199
def test_invalid_http_urls(self):
200
"""Trap invalid construction of urls"""
201
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
202
self.assertRaises(errors.InvalidURL,
204
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
206
def test_http_root_urls(self):
207
"""Construction of URLs from server root"""
208
t = self._transport('http://bzr.ozlabs.org/')
209
eq = self.assertEqualDiff
210
eq(t.abspath('.bzr/tree-version'),
211
'http://bzr.ozlabs.org/.bzr/tree-version')
213
def test_http_impl_urls(self):
214
"""There are servers which ask for particular clients to connect"""
215
server = self._server()
218
url = server.get_url()
219
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
224
class TestHttpUrls_urllib(TestHttpTransportUrls, tests.TestCase):
225
"""Test http urls with urllib"""
227
_transport = HttpTransport_urllib
228
_server = HttpServer_urllib
229
_qualified_prefix = 'http+urllib'
232
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
234
"""Test http urls with pycurl"""
236
_server = HttpServer_PyCurl
237
_qualified_prefix = 'http+pycurl'
239
# TODO: This should really be moved into another pycurl
240
# specific test. When https tests will be implemented, take
241
# this one into account.
242
def test_pycurl_without_https_support(self):
243
"""Test that pycurl without SSL do not fail with a traceback.
245
For the purpose of the test, we force pycurl to ignore
246
https by supplying a fake version_info that do not
252
raise tests.TestSkipped('pycurl not present')
253
# Now that we have pycurl imported, we can fake its version_info
254
# This was taken from a windows pycurl without SSL
256
pycurl.version_info = lambda : (2,
264
('ftp', 'gopher', 'telnet',
265
'dict', 'ldap', 'http', 'file'),
269
self.assertRaises(errors.DependencyNotPresent, self._transport,
270
'https://launchpad.net')
272
class TestHttpConnections(object):
273
"""Test the http connections.
275
This MUST be used by daughter classes that also inherit from
276
TestCaseWithWebserver.
278
We can't inherit directly from TestCaseWithWebserver or the
279
test framework will try to create an instance which cannot
280
run, its implementation being incomplete.
284
TestCaseWithWebserver.setUp(self)
285
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
286
transport=self.get_transport())
288
def test_http_has(self):
289
server = self.get_readonly_server()
290
t = self._transport(server.get_url())
291
self.assertEqual(t.has('foo/bar'), True)
292
self.assertEqual(len(server.logs), 1)
293
self.assertContainsRe(server.logs[0],
294
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
296
def test_http_has_not_found(self):
297
server = self.get_readonly_server()
298
t = self._transport(server.get_url())
299
self.assertEqual(t.has('not-found'), False)
300
self.assertContainsRe(server.logs[1],
301
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
303
def test_http_get(self):
304
server = self.get_readonly_server()
305
t = self._transport(server.get_url())
306
fp = t.get('foo/bar')
307
self.assertEqualDiff(
309
'contents of foo/bar\n')
310
self.assertEqual(len(server.logs), 1)
311
self.assertTrue(server.logs[0].find(
312
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
313
% bzrlib.__version__) > -1)
315
def test_get_smart_medium(self):
316
# For HTTP, get_smart_medium should return the transport object.
317
server = self.get_readonly_server()
318
http_transport = self._transport(server.get_url())
319
medium = http_transport.get_smart_medium()
320
self.assertIs(medium, http_transport)
322
def test_has_on_bogus_host(self):
323
# Get a free address and don't 'accept' on it, so that we
324
# can be sure there is no http handler there, but set a
325
# reasonable timeout to not slow down tests too much.
326
default_timeout = socket.getdefaulttimeout()
328
socket.setdefaulttimeout(2)
330
s.bind(('localhost', 0))
331
t = self._transport('http://%s:%s/' % s.getsockname())
332
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
334
socket.setdefaulttimeout(default_timeout)
337
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
338
"""Test http connections with urllib"""
340
_transport = HttpTransport_urllib
344
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
346
TestCaseWithWebserver):
347
"""Test http connections with pycurl"""
350
class TestHttpTransportRegistration(tests.TestCase):
351
"""Test registrations of various http implementations"""
353
def test_http_registered(self):
354
# urlllib should always be present
355
t = get_transport('http+urllib://bzr.google.com/')
356
self.assertIsInstance(t, Transport)
357
self.assertIsInstance(t, HttpTransport_urllib)
360
class TestPost(object):
362
def _test_post_body_is_received(self, scheme):
363
server = RecordingServer(expect_body_tail='end-of-body')
365
self.addCleanup(server.tearDown)
366
url = '%s://%s:%s/' % (scheme, server.host, server.port)
368
http_transport = get_transport(url)
369
except errors.UnsupportedProtocol:
370
raise tests.TestSkipped('%s not available' % scheme)
371
code, response = http_transport._post('abc def end-of-body')
373
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
374
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
375
# The transport should not be assuming that the server can accept
376
# chunked encoding the first time it connects, because HTTP/1.1, so we
377
# check for the literal string.
379
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
382
class TestPost_urllib(tests.TestCase, TestPost):
383
"""TestPost for urllib implementation"""
385
_transport = HttpTransport_urllib
387
def test_post_body_is_received_urllib(self):
388
self._test_post_body_is_received('http+urllib')
391
class TestPost_pycurl(TestWithTransport_pycurl, tests.TestCase, TestPost):
392
"""TestPost for pycurl implementation"""
394
def test_post_body_is_received_pycurl(self):
395
self._test_post_body_is_received('http+pycurl')
398
class TestRangeHeader(tests.TestCase):
399
"""Test range_header method"""
401
def check_header(self, value, ranges=[], tail=0):
402
offsets = [ (start, end - start + 1) for start, end in ranges]
403
coalesce = Transport._coalesce_offsets
404
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
405
range_header = HttpTransportBase._range_header
406
self.assertEqual(value, range_header(coalesced, tail))
408
def test_range_header_single(self):
409
self.check_header('0-9', ranges=[(0,9)])
410
self.check_header('100-109', ranges=[(100,109)])
412
def test_range_header_tail(self):
413
self.check_header('-10', tail=10)
414
self.check_header('-50', tail=50)
416
def test_range_header_multi(self):
417
self.check_header('0-9,100-200,300-5000',
418
ranges=[(0,9), (100, 200), (300,5000)])
420
def test_range_header_mixed(self):
421
self.check_header('0-9,300-5000,-50',
422
ranges=[(0,9), (300,5000)],
426
class TestWallServer(object):
427
"""Tests exceptions during the connection phase"""
429
def create_transport_readonly_server(self):
430
return HttpServer(WallRequestHandler)
432
def test_http_has(self):
433
server = self.get_readonly_server()
434
t = self._transport(server.get_url())
435
# Unfortunately httplib (see HTTPResponse._read_status
436
# for details) make no distinction between a closed
437
# socket and badly formatted status line, so we can't
438
# just test for ConnectionError, we have to test
439
# InvalidHttpResponse too.
440
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
443
def test_http_get(self):
444
server = self.get_readonly_server()
445
t = self._transport(server.get_url())
446
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
450
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
451
"""Tests "wall" server for urllib implementation"""
453
_transport = HttpTransport_urllib
456
class TestWallServer_pycurl(TestWithTransport_pycurl,
458
TestCaseWithWebserver):
459
"""Tests "wall" server for pycurl implementation"""
462
class TestBadStatusServer(object):
463
"""Tests bad status from server."""
465
def create_transport_readonly_server(self):
466
return HttpServer(BadStatusRequestHandler)
468
def test_http_has(self):
469
server = self.get_readonly_server()
470
t = self._transport(server.get_url())
471
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
473
def test_http_get(self):
474
server = self.get_readonly_server()
475
t = self._transport(server.get_url())
476
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
479
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
480
"""Tests bad status server for urllib implementation"""
482
_transport = HttpTransport_urllib
485
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
487
TestCaseWithWebserver):
488
"""Tests bad status server for pycurl implementation"""
491
class TestInvalidStatusServer(TestBadStatusServer):
492
"""Tests invalid status from server.
494
Both implementations raises the same error as for a bad status.
497
def create_transport_readonly_server(self):
498
return HttpServer(InvalidStatusRequestHandler)
501
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
502
TestCaseWithWebserver):
503
"""Tests invalid status server for urllib implementation"""
505
_transport = HttpTransport_urllib
508
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
509
TestInvalidStatusServer,
510
TestCaseWithWebserver):
511
"""Tests invalid status server for pycurl implementation"""
514
class TestBadProtocolServer(object):
515
"""Tests bad protocol from server."""
517
def create_transport_readonly_server(self):
518
return HttpServer(BadProtocolRequestHandler)
520
def test_http_has(self):
521
server = self.get_readonly_server()
522
t = self._transport(server.get_url())
523
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
525
def test_http_get(self):
526
server = self.get_readonly_server()
527
t = self._transport(server.get_url())
528
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
531
class TestBadProtocolServer_urllib(TestBadProtocolServer,
532
TestCaseWithWebserver):
533
"""Tests bad protocol server for urllib implementation"""
535
_transport = HttpTransport_urllib
537
# curl don't check the protocol version
538
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
539
# TestBadProtocolServer,
540
# TestCaseWithWebserver):
541
# """Tests bad protocol server for pycurl implementation"""
544
class TestForbiddenServer(object):
545
"""Tests forbidden server"""
547
def create_transport_readonly_server(self):
548
return HttpServer(ForbiddenRequestHandler)
550
def test_http_has(self):
551
server = self.get_readonly_server()
552
t = self._transport(server.get_url())
553
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
555
def test_http_get(self):
556
server = self.get_readonly_server()
557
t = self._transport(server.get_url())
558
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
561
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
562
"""Tests forbidden server for urllib implementation"""
564
_transport = HttpTransport_urllib
567
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
569
TestCaseWithWebserver):
570
"""Tests forbidden server for pycurl implementation"""
573
class TestRecordingServer(tests.TestCase):
575
def test_create(self):
576
server = RecordingServer(expect_body_tail=None)
577
self.assertEqual('', server.received_bytes)
578
self.assertEqual(None, server.host)
579
self.assertEqual(None, server.port)
581
def test_setUp_and_tearDown(self):
582
server = RecordingServer(expect_body_tail=None)
585
self.assertNotEqual(None, server.host)
586
self.assertNotEqual(None, server.port)
589
self.assertEqual(None, server.host)
590
self.assertEqual(None, server.port)
592
def test_send_receive_bytes(self):
593
server = RecordingServer(expect_body_tail='c')
595
self.addCleanup(server.tearDown)
596
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
597
sock.connect((server.host, server.port))
599
self.assertEqual('HTTP/1.1 200 OK\r\n',
600
osutils.recv_all(sock, 4096))
601
self.assertEqual('abc', server.received_bytes)
604
class TestRangeRequestServer(object):
605
"""Tests readv requests against server.
607
This MUST be used by daughter classes that also inherit from
608
TestCaseWithWebserver.
610
We can't inherit directly from TestCaseWithWebserver or the
611
test framework will try to create an instance which cannot
612
run, its implementation being incomplete.
616
TestCaseWithWebserver.setUp(self)
617
self.build_tree_contents([('a', '0123456789')],)
619
def test_readv(self):
620
server = self.get_readonly_server()
621
t = self._transport(server.get_url())
622
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
623
self.assertEqual(l[0], (0, '0'))
624
self.assertEqual(l[1], (1, '1'))
625
self.assertEqual(l[2], (3, '34'))
626
self.assertEqual(l[3], (9, '9'))
628
def test_readv_out_of_order(self):
629
server = self.get_readonly_server()
630
t = self._transport(server.get_url())
631
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
632
self.assertEqual(l[0], (1, '1'))
633
self.assertEqual(l[1], (9, '9'))
634
self.assertEqual(l[2], (0, '0'))
635
self.assertEqual(l[3], (3, '34'))
637
def test_readv_invalid_ranges(self):
638
server = self.get_readonly_server()
639
t = self._transport(server.get_url())
641
# This is intentionally reading off the end of the file
642
# since we are sure that it cannot get there
643
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
644
t.readv, 'a', [(1,1), (8,10)])
646
# This is trying to seek past the end of the file, it should
647
# also raise a special error
648
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
649
t.readv, 'a', [(12,2)])
651
def test_readv_multiple_get_requests(self):
652
server = self.get_readonly_server()
653
t = self._transport(server.get_url())
654
# force transport to issue multiple requests
655
t._max_readv_combine = 1
656
t._max_get_ranges = 1
657
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
658
self.assertEqual(l[0], (0, '0'))
659
self.assertEqual(l[1], (1, '1'))
660
self.assertEqual(l[2], (3, '34'))
661
self.assertEqual(l[3], (9, '9'))
662
# The server should have issued 4 requests
663
self.assertEqual(4, server.GET_request_nb)
665
def test_readv_get_max_size(self):
666
server = self.get_readonly_server()
667
t = self._transport(server.get_url())
668
# force transport to issue multiple requests by limiting the number of
669
# bytes by request. Note that this apply to coalesced offsets only, a
670
# single range ill keep its size even if bigger than the limit.
672
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
673
self.assertEqual(l[0], (0, '0'))
674
self.assertEqual(l[1], (1, '1'))
675
self.assertEqual(l[2], (2, '2345'))
676
self.assertEqual(l[3], (6, '6789'))
677
# The server should have issued 3 requests
678
self.assertEqual(3, server.GET_request_nb)
681
class TestSingleRangeRequestServer(TestRangeRequestServer):
682
"""Test readv against a server which accept only single range requests"""
684
def create_transport_readonly_server(self):
685
return HttpServer(SingleRangeRequestHandler)
688
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
689
TestCaseWithWebserver):
690
"""Tests single range requests accepting server for urllib implementation"""
692
_transport = HttpTransport_urllib
695
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
696
TestSingleRangeRequestServer,
697
TestCaseWithWebserver):
698
"""Tests single range requests accepting server for pycurl implementation"""
701
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
702
"""Test readv against a server which only accept single range requests"""
704
def create_transport_readonly_server(self):
705
return HttpServer(SingleOnlyRangeRequestHandler)
708
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
709
TestCaseWithWebserver):
710
"""Tests single range requests accepting server for urllib implementation"""
712
_transport = HttpTransport_urllib
715
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
716
TestSingleOnlyRangeRequestServer,
717
TestCaseWithWebserver):
718
"""Tests single range requests accepting server for pycurl implementation"""
721
class TestNoRangeRequestServer(TestRangeRequestServer):
722
"""Test readv against a server which do not accept range requests"""
724
def create_transport_readonly_server(self):
725
return HttpServer(NoRangeRequestHandler)
728
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
729
TestCaseWithWebserver):
730
"""Tests range requests refusing server for urllib implementation"""
732
_transport = HttpTransport_urllib
735
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
736
TestNoRangeRequestServer,
737
TestCaseWithWebserver):
738
"""Tests range requests refusing server for pycurl implementation"""
741
class TestLimitedRangeRequestServer(object):
742
"""Tests readv requests against server that errors out on too much ranges.
744
This MUST be used by daughter classes that also inherit from
745
TestCaseWithWebserver.
747
We can't inherit directly from TestCaseWithWebserver or the
748
test framework will try to create an instance which cannot
749
run, its implementation being incomplete.
754
def create_transport_readonly_server(self):
755
# Requests with more range specifiers will error out
756
return LimitedRangeHTTPServer(range_limit=self.range_limit)
758
def get_transport(self):
759
return self._transport(self.get_readonly_server().get_url())
762
TestCaseWithWebserver.setUp(self)
763
# We need to manipulate ranges that correspond to real chunks in the
764
# response, so we build a content appropriately.
765
filler = ''.join(['abcdefghij' for x in range(102)])
766
content = ''.join(['%04d' % v + filler for v in range(16)])
767
self.build_tree_contents([('a', content)],)
769
def test_few_ranges(self):
770
t = self.get_transport()
771
l = list(t.readv('a', ((0, 4), (1024, 4), )))
772
self.assertEqual(l[0], (0, '0000'))
773
self.assertEqual(l[1], (1024, '0001'))
774
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
776
def test_more_ranges(self):
777
t = self.get_transport()
778
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
779
self.assertEqual(l[0], (0, '0000'))
780
self.assertEqual(l[1], (1024, '0001'))
781
self.assertEqual(l[2], (4096, '0004'))
782
self.assertEqual(l[3], (8192, '0008'))
783
# The server will refuse to serve the first request (too much ranges),
784
# a second request will succeeds.
785
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
788
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
789
TestCaseWithWebserver):
790
"""Tests limited range requests server for urllib implementation"""
792
_transport = HttpTransport_urllib
795
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
796
TestLimitedRangeRequestServer,
797
TestCaseWithWebserver):
798
"""Tests limited range requests server for pycurl implementation"""
802
class TestHttpProxyWhiteBox(tests.TestCase):
803
"""Whitebox test proxy http authorization.
805
Only the urllib implementation is tested here.
809
tests.TestCase.setUp(self)
815
def _install_env(self, env):
816
for name, value in env.iteritems():
817
self._old_env[name] = osutils.set_or_unset_env(name, value)
819
def _restore_env(self):
820
for name, value in self._old_env.iteritems():
821
osutils.set_or_unset_env(name, value)
823
def _proxied_request(self):
824
handler = ProxyHandler()
825
request = Request('GET','http://baz/buzzle')
826
handler.set_proxy(request, 'http')
829
def test_empty_user(self):
830
self._install_env({'http_proxy': 'http://bar.com'})
831
request = self._proxied_request()
832
self.assertFalse(request.headers.has_key('Proxy-authorization'))
834
def test_invalid_proxy(self):
835
"""A proxy env variable without scheme"""
836
self._install_env({'http_proxy': 'host:1234'})
837
self.assertRaises(errors.InvalidURL, self._proxied_request)
840
class TestProxyHttpServer(object):
841
"""Tests proxy server.
843
This MUST be used by daughter classes that also inherit from
844
TestCaseWithTwoWebservers.
846
We can't inherit directly from TestCaseWithTwoWebservers or
847
the test framework will try to create an instance which
848
cannot run, its implementation being incomplete.
850
Be aware that we do not setup a real proxy here. Instead, we
851
check that the *connection* goes through the proxy by serving
852
different content (the faked proxy server append '-proxied'
856
# FIXME: We don't have an https server available, so we don't
857
# test https connections.
860
TestCaseWithTwoWebservers.setUp(self)
861
self.build_tree_contents([('foo', 'contents of foo\n'),
862
('foo-proxied', 'proxied contents of foo\n')])
863
# Let's setup some attributes for tests
864
self.server = self.get_readonly_server()
865
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
866
self.no_proxy_host = self.proxy_address
867
# The secondary server is the proxy
868
self.proxy = self.get_secondary_server()
869
self.proxy_url = self.proxy.get_url()
872
def create_transport_secondary_server(self):
873
"""Creates an http server that will serve files with
874
'-proxied' appended to their names.
878
def _install_env(self, env):
879
for name, value in env.iteritems():
880
self._old_env[name] = osutils.set_or_unset_env(name, value)
882
def _restore_env(self):
883
for name, value in self._old_env.iteritems():
884
osutils.set_or_unset_env(name, value)
886
def proxied_in_env(self, env):
887
self._install_env(env)
888
url = self.server.get_url()
889
t = self._transport(url)
891
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
895
def not_proxied_in_env(self, env):
896
self._install_env(env)
897
url = self.server.get_url()
898
t = self._transport(url)
900
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
904
def test_http_proxy(self):
905
self.proxied_in_env({'http_proxy': self.proxy_url})
907
def test_HTTP_PROXY(self):
908
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
910
def test_all_proxy(self):
911
self.proxied_in_env({'all_proxy': self.proxy_url})
913
def test_ALL_PROXY(self):
914
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
916
def test_http_proxy_with_no_proxy(self):
917
self.not_proxied_in_env({'http_proxy': self.proxy_url,
918
'no_proxy': self.no_proxy_host})
920
def test_HTTP_PROXY_with_NO_PROXY(self):
921
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
922
'NO_PROXY': self.no_proxy_host})
924
def test_all_proxy_with_no_proxy(self):
925
self.not_proxied_in_env({'all_proxy': self.proxy_url,
926
'no_proxy': self.no_proxy_host})
928
def test_ALL_PROXY_with_NO_PROXY(self):
929
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
930
'NO_PROXY': self.no_proxy_host})
932
def test_http_proxy_without_scheme(self):
933
self.assertRaises(errors.InvalidURL,
935
{'http_proxy': self.proxy_address})
938
class TestProxyHttpServer_urllib(TestProxyHttpServer,
939
TestCaseWithTwoWebservers):
940
"""Tests proxy server for urllib implementation"""
942
_transport = HttpTransport_urllib
945
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
947
TestCaseWithTwoWebservers):
948
"""Tests proxy server for pycurl implementation"""
951
TestProxyHttpServer.setUp(self)
952
# Oh my ! pycurl does not check for the port as part of
953
# no_proxy :-( So we just test the host part
954
self.no_proxy_host = 'localhost'
956
def test_HTTP_PROXY(self):
957
# pycurl does not check HTTP_PROXY for security reasons
958
# (for use in a CGI context that we do not care
959
# about. Should we ?)
960
raise tests.TestNotApplicable(
961
'pycurl does not check HTTP_PROXY for security reasons')
963
def test_HTTP_PROXY_with_NO_PROXY(self):
964
raise tests.TestNotApplicable(
965
'pycurl does not check HTTP_PROXY for security reasons')
967
def test_http_proxy_without_scheme(self):
968
# pycurl *ignores* invalid proxy env variables. If that
969
# ever change in the future, this test will fail
970
# indicating that pycurl do not ignore anymore such
972
self.not_proxied_in_env({'http_proxy': self.proxy_address})
975
class TestRanges(object):
976
"""Test the Range header in GET methods..
978
This MUST be used by daughter classes that also inherit from
979
TestCaseWithWebserver.
981
We can't inherit directly from TestCaseWithWebserver or the
982
test framework will try to create an instance which cannot
983
run, its implementation being incomplete.
987
TestCaseWithWebserver.setUp(self)
988
self.build_tree_contents([('a', '0123456789')],)
989
server = self.get_readonly_server()
990
self.transport = self._transport(server.get_url())
992
def _file_contents(self, relpath, ranges):
993
offsets = [ (start, end - start + 1) for start, end in ranges]
994
coalesce = self.transport._coalesce_offsets
995
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
996
code, data = self.transport._get(relpath, coalesced)
997
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
998
for start, end in ranges:
1000
yield data.read(end - start + 1)
1002
def _file_tail(self, relpath, tail_amount):
1003
code, data = self.transport._get(relpath, [], tail_amount)
1004
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1005
data.seek(-tail_amount, 2)
1006
return data.read(tail_amount)
1008
def test_range_header(self):
1010
map(self.assertEqual,['0', '234'],
1011
list(self._file_contents('a', [(0,0), (2,4)])),)
1013
self.assertEqual('789', self._file_tail('a', 3))
1014
# Syntactically invalid range
1015
self.assertListRaises(errors.InvalidHttpRange,
1016
self._file_contents, 'a', [(4, 3)])
1017
# Semantically invalid range
1018
self.assertListRaises(errors.InvalidHttpRange,
1019
self._file_contents, 'a', [(42, 128)])
1022
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1023
"""Test the Range header in GET methods for urllib implementation"""
1025
_transport = HttpTransport_urllib
1028
class TestRanges_pycurl(TestWithTransport_pycurl,
1030
TestCaseWithWebserver):
1031
"""Test the Range header in GET methods for pycurl implementation"""
1034
class TestHTTPRedirections(object):
1035
"""Test redirection between http servers.
1037
This MUST be used by daughter classes that also inherit from
1038
TestCaseWithRedirectedWebserver.
1040
We can't inherit directly from TestCaseWithTwoWebservers or the
1041
test framework will try to create an instance which cannot
1042
run, its implementation being incomplete.
1045
def create_transport_secondary_server(self):
1046
"""Create the secondary server redirecting to the primary server"""
1047
new = self.get_readonly_server()
1049
redirecting = HTTPServerRedirecting()
1050
redirecting.redirect_to(new.host, new.port)
1054
super(TestHTTPRedirections, self).setUp()
1055
self.build_tree_contents([('a', '0123456789'),
1057
'# Bazaar revision bundle v0.9\n#\n')
1060
self.old_transport = self._transport(self.old_server.get_url())
1062
def test_redirected(self):
1063
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1064
t = self._transport(self.new_server.get_url())
1065
self.assertEqual('0123456789', t.get('a').read())
1067
def test_read_redirected_bundle_from_url(self):
1068
from bzrlib.bundle import read_bundle_from_url
1069
url = self.old_transport.abspath('bundle')
1070
bundle = read_bundle_from_url(url)
1071
# If read_bundle_from_url was successful we get an empty bundle
1072
self.assertEqual([], bundle.revisions)
1075
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1076
TestCaseWithRedirectedWebserver):
1077
"""Tests redirections for urllib implementation"""
1079
_transport = HttpTransport_urllib
1083
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1084
TestHTTPRedirections,
1085
TestCaseWithRedirectedWebserver):
1086
"""Tests redirections for pycurl implementation"""
1089
class RedirectedRequest(Request):
1090
"""Request following redirections"""
1092
init_orig = Request.__init__
1094
def __init__(self, method, url, *args, **kwargs):
1095
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1096
self.follow_redirections = True
1099
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1100
"""Test redirections provided by urllib.
1102
http implementations do not redirect silently anymore (they
1103
do not redirect at all in fact). The mechanism is still in
1104
place at the _urllib2_wrappers.Request level and these tests
1107
For the pycurl implementation
1108
the redirection have been deleted as we may deprecate pycurl
1109
and I have no place to keep a working implementation.
1113
_transport = HttpTransport_urllib
1116
super(TestHTTPSilentRedirections_urllib, self).setUp()
1117
self.setup_redirected_request()
1118
self.addCleanup(self.cleanup_redirected_request)
1119
self.build_tree_contents([('a','a'),
1121
('1/a', 'redirected once'),
1123
('2/a', 'redirected twice'),
1125
('3/a', 'redirected thrice'),
1127
('4/a', 'redirected 4 times'),
1129
('5/a', 'redirected 5 times'),
1132
self.old_transport = self._transport(self.old_server.get_url())
1134
def setup_redirected_request(self):
1135
self.original_class = _urllib2_wrappers.Request
1136
_urllib2_wrappers.Request = RedirectedRequest
1138
def cleanup_redirected_request(self):
1139
_urllib2_wrappers.Request = self.original_class
1141
def create_transport_secondary_server(self):
1142
"""Create the secondary server, redirections are defined in the tests"""
1143
return HTTPServerRedirecting()
1145
def test_one_redirection(self):
1146
t = self.old_transport
1148
req = RedirectedRequest('GET', t.abspath('a'))
1149
req.follow_redirections = True
1150
new_prefix = 'http://%s:%s' % (self.new_server.host,
1151
self.new_server.port)
1152
self.old_server.redirections = \
1153
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1154
self.assertEquals('redirected once',t._perform(req).read())
1156
def test_five_redirections(self):
1157
t = self.old_transport
1159
req = RedirectedRequest('GET', t.abspath('a'))
1160
req.follow_redirections = True
1161
old_prefix = 'http://%s:%s' % (self.old_server.host,
1162
self.old_server.port)
1163
new_prefix = 'http://%s:%s' % (self.new_server.host,
1164
self.new_server.port)
1165
self.old_server.redirections = \
1166
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1167
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1168
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1169
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1170
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1172
self.assertEquals('redirected 5 times',t._perform(req).read())
1175
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1176
"""Test transport.do_catching_redirections.
1178
We arbitrarily choose to use urllib transports
1181
_transport = HttpTransport_urllib
1184
super(TestDoCatchRedirections, self).setUp()
1185
self.build_tree_contents([('a', '0123456789'),],)
1187
self.old_transport = self._transport(self.old_server.get_url())
1189
def get_a(self, transport):
1190
return transport.get('a')
1192
def test_no_redirection(self):
1193
t = self._transport(self.new_server.get_url())
1195
# We use None for redirected so that we fail if redirected
1196
self.assertEquals('0123456789',
1197
do_catching_redirections(self.get_a, t, None).read())
1199
def test_one_redirection(self):
1200
self.redirections = 0
1202
def redirected(transport, exception, redirection_notice):
1203
self.redirections += 1
1204
dir, file = urlutils.split(exception.target)
1205
return self._transport(dir)
1207
self.assertEquals('0123456789',
1208
do_catching_redirections(self.get_a,
1212
self.assertEquals(1, self.redirections)
1214
def test_redirection_loop(self):
1216
def redirected(transport, exception, redirection_notice):
1217
# By using the redirected url as a base dir for the
1218
# *old* transport, we create a loop: a => a/a =>
1220
return self.old_transport.clone(exception.target)
1222
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1223
self.get_a, self.old_transport, redirected)
1226
class TestAuth(object):
1227
"""Test some authentication scheme specified by daughter class.
1229
This MUST be used by daughter classes that also inherit from
1230
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1233
_password_prompt_prefix = ''
1236
"""Set up the test environment
1238
Daughter classes should set up their own environment
1239
(including self.server) and explicitely call this
1240
method. This is needed because we want to reuse the same
1241
tests for proxy and no-proxy accesses which have
1242
different ways of setting self.server.
1244
self.build_tree_contents([('a', 'contents of a\n'),
1245
('b', 'contents of b\n'),])
1247
def get_user_url(self, user=None, password=None):
1248
"""Build an url embedding user and password"""
1249
url = '%s://' % self.server._url_protocol
1250
if user is not None:
1252
if password is not None:
1253
url += ':' + password
1255
url += '%s:%s/' % (self.server.host, self.server.port)
1258
def test_no_user(self):
1259
self.server.add_user('joe', 'foo')
1260
t = self.get_user_transport()
1261
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1262
# Only one 'Authentication Required' error should occur
1263
self.assertEqual(1, self.server.auth_required_errors)
1265
def test_empty_pass(self):
1266
self.server.add_user('joe', '')
1267
t = self.get_user_transport('joe', '')
1268
self.assertEqual('contents of a\n', t.get('a').read())
1269
# Only one 'Authentication Required' error should occur
1270
self.assertEqual(1, self.server.auth_required_errors)
1272
def test_user_pass(self):
1273
self.server.add_user('joe', 'foo')
1274
t = self.get_user_transport('joe', 'foo')
1275
self.assertEqual('contents of a\n', t.get('a').read())
1276
# Only one 'Authentication Required' error should occur
1277
self.assertEqual(1, self.server.auth_required_errors)
1279
def test_unknown_user(self):
1280
self.server.add_user('joe', 'foo')
1281
t = self.get_user_transport('bill', 'foo')
1282
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1283
# Two 'Authentication Required' errors should occur (the
1284
# initial 'who are you' and 'I don't know you, who are
1286
self.assertEqual(2, self.server.auth_required_errors)
1288
def test_wrong_pass(self):
1289
self.server.add_user('joe', 'foo')
1290
t = self.get_user_transport('joe', 'bar')
1291
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1292
# Two 'Authentication Required' errors should occur (the
1293
# initial 'who are you' and 'this is not you, who are you')
1294
self.assertEqual(2, self.server.auth_required_errors)
1296
def test_prompt_for_password(self):
1297
self.server.add_user('joe', 'foo')
1298
t = self.get_user_transport('joe', None)
1299
stdout = tests.StringIOWrapper()
1300
ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1301
self.assertEqual('contents of a\n',t.get('a').read())
1302
# stdin should be empty
1303
self.assertEqual('', ui.ui_factory.stdin.readline())
1304
self._check_password_prompt(t._unqualified_scheme, 'joe',
1306
# And we shouldn't prompt again for a different request
1307
# against the same transport.
1308
self.assertEqual('contents of b\n',t.get('b').read())
1310
# And neither against a clone
1311
self.assertEqual('contents of b\n',t2.get('b').read())
1312
# Only one 'Authentication Required' error should occur
1313
self.assertEqual(1, self.server.auth_required_errors)
1315
def _check_password_prompt(self, scheme, user, actual_prompt):
1316
expected_prompt = (self._password_prompt_prefix
1317
+ ("%s %s@%s:%d, Realm: '%s' password: "
1319
user, self.server.host, self.server.port,
1320
self.server.auth_realm)))
1321
self.assertEquals(expected_prompt, actual_prompt)
1323
def test_no_prompt_for_password_when_using_auth_config(self):
1326
stdin_content = 'bar\n' # Not the right password
1327
self.server.add_user(user, password)
1328
t = self.get_user_transport(user, None)
1329
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1330
stdout=tests.StringIOWrapper())
1331
# Create a minimal config file with the right password
1332
conf = config.AuthenticationConfig()
1333
conf._get_config().update(
1334
{'httptest': {'scheme': 'http', 'port': self.server.port,
1335
'user': user, 'password': password}})
1337
# Issue a request to the server to connect
1338
self.assertEqual('contents of a\n',t.get('a').read())
1339
# stdin should have been left untouched
1340
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1341
# Only one 'Authentication Required' error should occur
1342
self.assertEqual(1, self.server.auth_required_errors)
1346
class TestHTTPAuth(TestAuth):
1347
"""Test HTTP authentication schemes.
1349
Daughter classes MUST inherit from TestCaseWithWebserver too.
1352
_auth_header = 'Authorization'
1355
TestCaseWithWebserver.setUp(self)
1356
self.server = self.get_readonly_server()
1357
TestAuth.setUp(self)
1359
def get_user_transport(self, user=None, password=None):
1360
return self._transport(self.get_user_url(user, password))
1363
class TestProxyAuth(TestAuth):
1364
"""Test proxy authentication schemes.
1366
Daughter classes MUST also inherit from TestCaseWithWebserver.
1368
_auth_header = 'Proxy-authorization'
1369
_password_prompt_prefix = 'Proxy '
1373
TestCaseWithWebserver.setUp(self)
1374
self.server = self.get_readonly_server()
1376
self.addCleanup(self._restore_env)
1377
TestAuth.setUp(self)
1378
# Override the contents to avoid false positives
1379
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1380
('b', 'not proxied contents of b\n'),
1381
('a-proxied', 'contents of a\n'),
1382
('b-proxied', 'contents of b\n'),
1385
def get_user_transport(self, user=None, password=None):
1386
self._install_env({'all_proxy': self.get_user_url(user, password)})
1387
return self._transport(self.server.get_url())
1389
def _install_env(self, env):
1390
for name, value in env.iteritems():
1391
self._old_env[name] = osutils.set_or_unset_env(name, value)
1393
def _restore_env(self):
1394
for name, value in self._old_env.iteritems():
1395
osutils.set_or_unset_env(name, value)
1398
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1399
"""Test http basic authentication scheme"""
1401
_transport = HttpTransport_urllib
1403
def create_transport_readonly_server(self):
1404
return HTTPBasicAuthServer()
1407
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1408
"""Test proxy basic authentication scheme"""
1410
_transport = HttpTransport_urllib
1412
def create_transport_readonly_server(self):
1413
return ProxyBasicAuthServer()
1416
class TestDigestAuth(object):
1417
"""Digest Authentication specific tests"""
1419
def test_changing_nonce(self):
1420
self.server.add_user('joe', 'foo')
1421
t = self.get_user_transport('joe', 'foo')
1422
self.assertEqual('contents of a\n', t.get('a').read())
1423
self.assertEqual('contents of b\n', t.get('b').read())
1424
# Only one 'Authentication Required' error should have
1426
self.assertEqual(1, self.server.auth_required_errors)
1427
# The server invalidates the current nonce
1428
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1429
self.assertEqual('contents of a\n', t.get('a').read())
1430
# Two 'Authentication Required' errors should occur (the
1431
# initial 'who are you' and a second 'who are you' with the new nonce)
1432
self.assertEqual(2, self.server.auth_required_errors)
1435
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1436
"""Test http digest authentication scheme"""
1438
_transport = HttpTransport_urllib
1440
def create_transport_readonly_server(self):
1441
return HTTPDigestAuthServer()
1444
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1445
TestCaseWithWebserver):
1446
"""Test proxy digest authentication scheme"""
1448
_transport = HttpTransport_urllib
1450
def create_transport_readonly_server(self):
1451
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
