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, self.get_readonly_server().GET_request_nb)
666
class TestSingleRangeRequestServer(TestRangeRequestServer):
667
"""Test readv against a server which accept only single range requests"""
669
def create_transport_readonly_server(self):
670
return HttpServer(SingleRangeRequestHandler)
673
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
674
TestCaseWithWebserver):
675
"""Tests single range requests accepting server for urllib implementation"""
677
_transport = HttpTransport_urllib
680
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
681
TestSingleRangeRequestServer,
682
TestCaseWithWebserver):
683
"""Tests single range requests accepting server for pycurl implementation"""
686
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
687
"""Test readv against a server which only accept single range requests"""
689
def create_transport_readonly_server(self):
690
return HttpServer(SingleOnlyRangeRequestHandler)
693
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
694
TestCaseWithWebserver):
695
"""Tests single range requests accepting server for urllib implementation"""
697
_transport = HttpTransport_urllib
700
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
701
TestSingleOnlyRangeRequestServer,
702
TestCaseWithWebserver):
703
"""Tests single range requests accepting server for pycurl implementation"""
706
class TestNoRangeRequestServer(TestRangeRequestServer):
707
"""Test readv against a server which do not accept range requests"""
709
def create_transport_readonly_server(self):
710
return HttpServer(NoRangeRequestHandler)
713
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
714
TestCaseWithWebserver):
715
"""Tests range requests refusing server for urllib implementation"""
717
_transport = HttpTransport_urllib
720
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
721
TestNoRangeRequestServer,
722
TestCaseWithWebserver):
723
"""Tests range requests refusing server for pycurl implementation"""
726
class TestLimitedRangeRequestServer(object):
727
"""Tests readv requests against server that errors out on too much ranges.
729
This MUST be used by daughter classes that also inherit from
730
TestCaseWithWebserver.
732
We can't inherit directly from TestCaseWithWebserver or the
733
test framework will try to create an instance which cannot
734
run, its implementation being incomplete.
739
def create_transport_readonly_server(self):
740
# Requests with more range specifiers will error out
741
return LimitedRangeHTTPServer(range_limit=self.range_limit)
743
def get_transport(self):
744
return self._transport(self.get_readonly_server().get_url())
747
TestCaseWithWebserver.setUp(self)
748
# We need to manipulate ranges that correspond to real chunks in the
749
# response, so we build a content appropriately.
750
filler = ''.join(['abcdefghij' for x in range(102)])
751
content = ''.join(['%04d' % v + filler for v in range(16)])
752
self.build_tree_contents([('a', content)],)
754
def test_few_ranges(self):
755
t = self.get_transport()
756
l = list(t.readv('a', ((0, 4), (1024, 4), )))
757
self.assertEqual(l[0], (0, '0000'))
758
self.assertEqual(l[1], (1024, '0001'))
759
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
761
def test_more_ranges(self):
762
t = self.get_transport()
763
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
764
self.assertEqual(l[0], (0, '0000'))
765
self.assertEqual(l[1], (1024, '0001'))
766
self.assertEqual(l[2], (4096, '0004'))
767
self.assertEqual(l[3], (8192, '0008'))
768
# The server will refuse to serve the first request (too much ranges),
769
# a second request will succeeds.
770
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
773
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
774
TestCaseWithWebserver):
775
"""Tests limited range requests server for urllib implementation"""
777
_transport = HttpTransport_urllib
780
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
781
TestLimitedRangeRequestServer,
782
TestCaseWithWebserver):
783
"""Tests limited range requests server for pycurl implementation"""
787
class TestHttpProxyWhiteBox(tests.TestCase):
788
"""Whitebox test proxy http authorization.
790
Only the urllib implementation is tested here.
794
tests.TestCase.setUp(self)
800
def _install_env(self, env):
801
for name, value in env.iteritems():
802
self._old_env[name] = osutils.set_or_unset_env(name, value)
804
def _restore_env(self):
805
for name, value in self._old_env.iteritems():
806
osutils.set_or_unset_env(name, value)
808
def _proxied_request(self):
809
handler = ProxyHandler()
810
request = Request('GET','http://baz/buzzle')
811
handler.set_proxy(request, 'http')
814
def test_empty_user(self):
815
self._install_env({'http_proxy': 'http://bar.com'})
816
request = self._proxied_request()
817
self.assertFalse(request.headers.has_key('Proxy-authorization'))
819
def test_invalid_proxy(self):
820
"""A proxy env variable without scheme"""
821
self._install_env({'http_proxy': 'host:1234'})
822
self.assertRaises(errors.InvalidURL, self._proxied_request)
825
class TestProxyHttpServer(object):
826
"""Tests proxy server.
828
This MUST be used by daughter classes that also inherit from
829
TestCaseWithTwoWebservers.
831
We can't inherit directly from TestCaseWithTwoWebservers or
832
the test framework will try to create an instance which
833
cannot run, its implementation being incomplete.
835
Be aware that we do not setup a real proxy here. Instead, we
836
check that the *connection* goes through the proxy by serving
837
different content (the faked proxy server append '-proxied'
841
# FIXME: We don't have an https server available, so we don't
842
# test https connections.
845
TestCaseWithTwoWebservers.setUp(self)
846
self.build_tree_contents([('foo', 'contents of foo\n'),
847
('foo-proxied', 'proxied contents of foo\n')])
848
# Let's setup some attributes for tests
849
self.server = self.get_readonly_server()
850
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
851
self.no_proxy_host = self.proxy_address
852
# The secondary server is the proxy
853
self.proxy = self.get_secondary_server()
854
self.proxy_url = self.proxy.get_url()
857
def create_transport_secondary_server(self):
858
"""Creates an http server that will serve files with
859
'-proxied' appended to their names.
863
def _install_env(self, env):
864
for name, value in env.iteritems():
865
self._old_env[name] = osutils.set_or_unset_env(name, value)
867
def _restore_env(self):
868
for name, value in self._old_env.iteritems():
869
osutils.set_or_unset_env(name, value)
871
def proxied_in_env(self, env):
872
self._install_env(env)
873
url = self.server.get_url()
874
t = self._transport(url)
876
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
880
def not_proxied_in_env(self, env):
881
self._install_env(env)
882
url = self.server.get_url()
883
t = self._transport(url)
885
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
889
def test_http_proxy(self):
890
self.proxied_in_env({'http_proxy': self.proxy_url})
892
def test_HTTP_PROXY(self):
893
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
895
def test_all_proxy(self):
896
self.proxied_in_env({'all_proxy': self.proxy_url})
898
def test_ALL_PROXY(self):
899
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
901
def test_http_proxy_with_no_proxy(self):
902
self.not_proxied_in_env({'http_proxy': self.proxy_url,
903
'no_proxy': self.no_proxy_host})
905
def test_HTTP_PROXY_with_NO_PROXY(self):
906
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
907
'NO_PROXY': self.no_proxy_host})
909
def test_all_proxy_with_no_proxy(self):
910
self.not_proxied_in_env({'all_proxy': self.proxy_url,
911
'no_proxy': self.no_proxy_host})
913
def test_ALL_PROXY_with_NO_PROXY(self):
914
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
915
'NO_PROXY': self.no_proxy_host})
917
def test_http_proxy_without_scheme(self):
918
self.assertRaises(errors.InvalidURL,
920
{'http_proxy': self.proxy_address})
923
class TestProxyHttpServer_urllib(TestProxyHttpServer,
924
TestCaseWithTwoWebservers):
925
"""Tests proxy server for urllib implementation"""
927
_transport = HttpTransport_urllib
930
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
932
TestCaseWithTwoWebservers):
933
"""Tests proxy server for pycurl implementation"""
936
TestProxyHttpServer.setUp(self)
937
# Oh my ! pycurl does not check for the port as part of
938
# no_proxy :-( So we just test the host part
939
self.no_proxy_host = 'localhost'
941
def test_HTTP_PROXY(self):
942
# pycurl does not check HTTP_PROXY for security reasons
943
# (for use in a CGI context that we do not care
944
# about. Should we ?)
945
raise tests.TestNotApplicable(
946
'pycurl does not check HTTP_PROXY for security reasons')
948
def test_HTTP_PROXY_with_NO_PROXY(self):
949
raise tests.TestNotApplicable(
950
'pycurl does not check HTTP_PROXY for security reasons')
952
def test_http_proxy_without_scheme(self):
953
# pycurl *ignores* invalid proxy env variables. If that
954
# ever change in the future, this test will fail
955
# indicating that pycurl do not ignore anymore such
957
self.not_proxied_in_env({'http_proxy': self.proxy_address})
960
class TestRanges(object):
961
"""Test the Range header in GET methods..
963
This MUST be used by daughter classes that also inherit from
964
TestCaseWithWebserver.
966
We can't inherit directly from TestCaseWithWebserver or the
967
test framework will try to create an instance which cannot
968
run, its implementation being incomplete.
972
TestCaseWithWebserver.setUp(self)
973
self.build_tree_contents([('a', '0123456789')],)
974
server = self.get_readonly_server()
975
self.transport = self._transport(server.get_url())
977
def _file_contents(self, relpath, ranges):
978
offsets = [ (start, end - start + 1) for start, end in ranges]
979
coalesce = self.transport._coalesce_offsets
980
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
981
code, data = self.transport._get(relpath, coalesced)
982
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
983
for start, end in ranges:
985
yield data.read(end - start + 1)
987
def _file_tail(self, relpath, tail_amount):
988
code, data = self.transport._get(relpath, [], tail_amount)
989
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
990
data.seek(-tail_amount + 1, 2)
991
return data.read(tail_amount)
993
def test_range_header(self):
995
map(self.assertEqual,['0', '234'],
996
list(self._file_contents('a', [(0,0), (2,4)])),)
998
self.assertEqual('789', self._file_tail('a', 3))
999
# Syntactically invalid range
1000
self.assertListRaises(errors.InvalidRange,
1001
self._file_contents, 'a', [(4, 3)])
1002
# Semantically invalid range
1003
self.assertListRaises(errors.InvalidRange,
1004
self._file_contents, 'a', [(42, 128)])
1007
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1008
"""Test the Range header in GET methods for urllib implementation"""
1010
_transport = HttpTransport_urllib
1013
class TestRanges_pycurl(TestWithTransport_pycurl,
1015
TestCaseWithWebserver):
1016
"""Test the Range header in GET methods for pycurl implementation"""
1019
class TestHTTPRedirections(object):
1020
"""Test redirection between http servers.
1022
This MUST be used by daughter classes that also inherit from
1023
TestCaseWithRedirectedWebserver.
1025
We can't inherit directly from TestCaseWithTwoWebservers or the
1026
test framework will try to create an instance which cannot
1027
run, its implementation being incomplete.
1030
def create_transport_secondary_server(self):
1031
"""Create the secondary server redirecting to the primary server"""
1032
new = self.get_readonly_server()
1034
redirecting = HTTPServerRedirecting()
1035
redirecting.redirect_to(new.host, new.port)
1039
super(TestHTTPRedirections, self).setUp()
1040
self.build_tree_contents([('a', '0123456789'),
1042
'# Bazaar revision bundle v0.9\n#\n')
1045
self.old_transport = self._transport(self.old_server.get_url())
1047
def test_redirected(self):
1048
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1049
t = self._transport(self.new_server.get_url())
1050
self.assertEqual('0123456789', t.get('a').read())
1052
def test_read_redirected_bundle_from_url(self):
1053
from bzrlib.bundle import read_bundle_from_url
1054
url = self.old_transport.abspath('bundle')
1055
bundle = read_bundle_from_url(url)
1056
# If read_bundle_from_url was successful we get an empty bundle
1057
self.assertEqual([], bundle.revisions)
1060
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1061
TestCaseWithRedirectedWebserver):
1062
"""Tests redirections for urllib implementation"""
1064
_transport = HttpTransport_urllib
1068
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1069
TestHTTPRedirections,
1070
TestCaseWithRedirectedWebserver):
1071
"""Tests redirections for pycurl implementation"""
1074
class RedirectedRequest(Request):
1075
"""Request following redirections"""
1077
init_orig = Request.__init__
1079
def __init__(self, method, url, *args, **kwargs):
1080
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1081
self.follow_redirections = True
1084
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1085
"""Test redirections provided by urllib.
1087
http implementations do not redirect silently anymore (they
1088
do not redirect at all in fact). The mechanism is still in
1089
place at the _urllib2_wrappers.Request level and these tests
1092
For the pycurl implementation
1093
the redirection have been deleted as we may deprecate pycurl
1094
and I have no place to keep a working implementation.
1098
_transport = HttpTransport_urllib
1101
super(TestHTTPSilentRedirections_urllib, self).setUp()
1102
self.setup_redirected_request()
1103
self.addCleanup(self.cleanup_redirected_request)
1104
self.build_tree_contents([('a','a'),
1106
('1/a', 'redirected once'),
1108
('2/a', 'redirected twice'),
1110
('3/a', 'redirected thrice'),
1112
('4/a', 'redirected 4 times'),
1114
('5/a', 'redirected 5 times'),
1117
self.old_transport = self._transport(self.old_server.get_url())
1119
def setup_redirected_request(self):
1120
self.original_class = _urllib2_wrappers.Request
1121
_urllib2_wrappers.Request = RedirectedRequest
1123
def cleanup_redirected_request(self):
1124
_urllib2_wrappers.Request = self.original_class
1126
def create_transport_secondary_server(self):
1127
"""Create the secondary server, redirections are defined in the tests"""
1128
return HTTPServerRedirecting()
1130
def test_one_redirection(self):
1131
t = self.old_transport
1133
req = RedirectedRequest('GET', t.abspath('a'))
1134
req.follow_redirections = True
1135
new_prefix = 'http://%s:%s' % (self.new_server.host,
1136
self.new_server.port)
1137
self.old_server.redirections = \
1138
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1139
self.assertEquals('redirected once',t._perform(req).read())
1141
def test_five_redirections(self):
1142
t = self.old_transport
1144
req = RedirectedRequest('GET', t.abspath('a'))
1145
req.follow_redirections = True
1146
old_prefix = 'http://%s:%s' % (self.old_server.host,
1147
self.old_server.port)
1148
new_prefix = 'http://%s:%s' % (self.new_server.host,
1149
self.new_server.port)
1150
self.old_server.redirections = \
1151
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1152
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1153
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1154
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1155
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1157
self.assertEquals('redirected 5 times',t._perform(req).read())
1160
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1161
"""Test transport.do_catching_redirections.
1163
We arbitrarily choose to use urllib transports
1166
_transport = HttpTransport_urllib
1169
super(TestDoCatchRedirections, self).setUp()
1170
self.build_tree_contents([('a', '0123456789'),],)
1172
self.old_transport = self._transport(self.old_server.get_url())
1174
def get_a(self, transport):
1175
return transport.get('a')
1177
def test_no_redirection(self):
1178
t = self._transport(self.new_server.get_url())
1180
# We use None for redirected so that we fail if redirected
1181
self.assertEquals('0123456789',
1182
do_catching_redirections(self.get_a, t, None).read())
1184
def test_one_redirection(self):
1185
self.redirections = 0
1187
def redirected(transport, exception, redirection_notice):
1188
self.redirections += 1
1189
dir, file = urlutils.split(exception.target)
1190
return self._transport(dir)
1192
self.assertEquals('0123456789',
1193
do_catching_redirections(self.get_a,
1197
self.assertEquals(1, self.redirections)
1199
def test_redirection_loop(self):
1201
def redirected(transport, exception, redirection_notice):
1202
# By using the redirected url as a base dir for the
1203
# *old* transport, we create a loop: a => a/a =>
1205
return self.old_transport.clone(exception.target)
1207
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1208
self.get_a, self.old_transport, redirected)
1211
class TestAuth(object):
1212
"""Test some authentication scheme specified by daughter class.
1214
This MUST be used by daughter classes that also inherit from
1215
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1218
_password_prompt_prefix = ''
1221
"""Set up the test environment
1223
Daughter classes should set up their own environment
1224
(including self.server) and explicitely call this
1225
method. This is needed because we want to reuse the same
1226
tests for proxy and no-proxy accesses which have
1227
different ways of setting self.server.
1229
self.build_tree_contents([('a', 'contents of a\n'),
1230
('b', 'contents of b\n'),])
1232
def get_user_url(self, user=None, password=None):
1233
"""Build an url embedding user and password"""
1234
url = '%s://' % self.server._url_protocol
1235
if user is not None:
1237
if password is not None:
1238
url += ':' + password
1240
url += '%s:%s/' % (self.server.host, self.server.port)
1243
def test_no_user(self):
1244
self.server.add_user('joe', 'foo')
1245
t = self.get_user_transport()
1246
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1247
# Only one 'Authentication Required' error should occur
1248
self.assertEqual(1, self.server.auth_required_errors)
1250
def test_empty_pass(self):
1251
self.server.add_user('joe', '')
1252
t = self.get_user_transport('joe', '')
1253
self.assertEqual('contents of a\n', t.get('a').read())
1254
# Only one 'Authentication Required' error should occur
1255
self.assertEqual(1, self.server.auth_required_errors)
1257
def test_user_pass(self):
1258
self.server.add_user('joe', 'foo')
1259
t = self.get_user_transport('joe', 'foo')
1260
self.assertEqual('contents of a\n', t.get('a').read())
1261
# Only one 'Authentication Required' error should occur
1262
self.assertEqual(1, self.server.auth_required_errors)
1264
def test_unknown_user(self):
1265
self.server.add_user('joe', 'foo')
1266
t = self.get_user_transport('bill', 'foo')
1267
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1268
# Two 'Authentication Required' errors should occur (the
1269
# initial 'who are you' and 'I don't know you, who are
1271
self.assertEqual(2, self.server.auth_required_errors)
1273
def test_wrong_pass(self):
1274
self.server.add_user('joe', 'foo')
1275
t = self.get_user_transport('joe', 'bar')
1276
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1277
# Two 'Authentication Required' errors should occur (the
1278
# initial 'who are you' and 'this is not you, who are you')
1279
self.assertEqual(2, self.server.auth_required_errors)
1281
def test_prompt_for_password(self):
1282
self.server.add_user('joe', 'foo')
1283
t = self.get_user_transport('joe', None)
1284
stdout = tests.StringIOWrapper()
1285
ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1286
self.assertEqual('contents of a\n',t.get('a').read())
1287
# stdin should be empty
1288
self.assertEqual('', ui.ui_factory.stdin.readline())
1289
self._check_password_prompt(t._unqualified_scheme, 'joe',
1291
# And we shouldn't prompt again for a different request
1292
# against the same transport.
1293
self.assertEqual('contents of b\n',t.get('b').read())
1295
# And neither against a clone
1296
self.assertEqual('contents of b\n',t2.get('b').read())
1297
# Only one 'Authentication Required' error should occur
1298
self.assertEqual(1, self.server.auth_required_errors)
1300
def _check_password_prompt(self, scheme, user, actual_prompt):
1301
expected_prompt = (self._password_prompt_prefix
1302
+ ("%s %s@%s:%d, Realm: '%s' password: "
1304
user, self.server.host, self.server.port,
1305
self.server.auth_realm)))
1306
self.assertEquals(expected_prompt, actual_prompt)
1308
def test_no_prompt_for_password_when_using_auth_config(self):
1311
stdin_content = 'bar\n' # Not the right password
1312
self.server.add_user(user, password)
1313
t = self.get_user_transport(user, None)
1314
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1315
stdout=tests.StringIOWrapper())
1316
# Create a minimal config file with the right password
1317
conf = config.AuthenticationConfig()
1318
conf._get_config().update(
1319
{'httptest': {'scheme': 'http', 'port': self.server.port,
1320
'user': user, 'password': password}})
1322
# Issue a request to the server to connect
1323
self.assertEqual('contents of a\n',t.get('a').read())
1324
# stdin should have been left untouched
1325
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1326
# Only one 'Authentication Required' error should occur
1327
self.assertEqual(1, self.server.auth_required_errors)
1331
class TestHTTPAuth(TestAuth):
1332
"""Test HTTP authentication schemes.
1334
Daughter classes MUST inherit from TestCaseWithWebserver too.
1337
_auth_header = 'Authorization'
1340
TestCaseWithWebserver.setUp(self)
1341
self.server = self.get_readonly_server()
1342
TestAuth.setUp(self)
1344
def get_user_transport(self, user=None, password=None):
1345
return self._transport(self.get_user_url(user, password))
1348
class TestProxyAuth(TestAuth):
1349
"""Test proxy authentication schemes.
1351
Daughter classes MUST also inherit from TestCaseWithWebserver.
1353
_auth_header = 'Proxy-authorization'
1354
_password_prompt_prefix = 'Proxy '
1358
TestCaseWithWebserver.setUp(self)
1359
self.server = self.get_readonly_server()
1361
self.addCleanup(self._restore_env)
1362
TestAuth.setUp(self)
1363
# Override the contents to avoid false positives
1364
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1365
('b', 'not proxied contents of b\n'),
1366
('a-proxied', 'contents of a\n'),
1367
('b-proxied', 'contents of b\n'),
1370
def get_user_transport(self, user=None, password=None):
1371
self._install_env({'all_proxy': self.get_user_url(user, password)})
1372
return self._transport(self.server.get_url())
1374
def _install_env(self, env):
1375
for name, value in env.iteritems():
1376
self._old_env[name] = osutils.set_or_unset_env(name, value)
1378
def _restore_env(self):
1379
for name, value in self._old_env.iteritems():
1380
osutils.set_or_unset_env(name, value)
1383
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1384
"""Test http basic authentication scheme"""
1386
_transport = HttpTransport_urllib
1388
def create_transport_readonly_server(self):
1389
return HTTPBasicAuthServer()
1392
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1393
"""Test proxy basic authentication scheme"""
1395
_transport = HttpTransport_urllib
1397
def create_transport_readonly_server(self):
1398
return ProxyBasicAuthServer()
1401
class TestDigestAuth(object):
1402
"""Digest Authentication specific tests"""
1404
def test_changing_nonce(self):
1405
self.server.add_user('joe', 'foo')
1406
t = self.get_user_transport('joe', 'foo')
1407
self.assertEqual('contents of a\n', t.get('a').read())
1408
self.assertEqual('contents of b\n', t.get('b').read())
1409
# Only one 'Authentication Required' error should have
1411
self.assertEqual(1, self.server.auth_required_errors)
1412
# The server invalidates the current nonce
1413
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1414
self.assertEqual('contents of a\n', t.get('a').read())
1415
# Two 'Authentication Required' errors should occur (the
1416
# initial 'who are you' and a second 'who are you' with the new nonce)
1417
self.assertEqual(2, self.server.auth_required_errors)
1420
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1421
"""Test http digest authentication scheme"""
1423
_transport = HttpTransport_urllib
1425
def create_transport_readonly_server(self):
1426
return HTTPDigestAuthServer()
1429
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1430
TestCaseWithWebserver):
1431
"""Test proxy digest authentication scheme"""
1433
_transport = HttpTransport_urllib
1435
def create_transport_readonly_server(self):
1436
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
