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
38
from bzrlib.tests import (
44
from bzrlib.tests.HttpServer import (
49
from bzrlib.tests.HTTPTestUtil import (
50
BadProtocolRequestHandler,
51
BadStatusRequestHandler,
52
ForbiddenRequestHandler,
55
HTTPServerRedirecting,
56
InvalidStatusRequestHandler,
57
LimitedRangeHTTPServer,
58
NoRangeRequestHandler,
60
ProxyDigestAuthServer,
62
SingleRangeRequestHandler,
63
SingleOnlyRangeRequestHandler,
64
TestCaseWithRedirectedWebserver,
65
TestCaseWithTwoWebservers,
66
TestCaseWithWebserver,
69
from bzrlib.transport import (
71
do_catching_redirections,
75
from bzrlib.transport.http import (
80
from bzrlib.transport.http._urllib import HttpTransport_urllib
81
from bzrlib.transport.http._urllib2_wrappers import (
87
class FakeManager(object):
92
def add_password(self, realm, host, username, password):
93
self.credentials.append([realm, host, username, password])
96
class RecordingServer(object):
97
"""A fake HTTP server.
99
It records the bytes sent to it, and replies with a 200.
102
def __init__(self, expect_body_tail=None):
105
:type expect_body_tail: str
106
:param expect_body_tail: a reply won't be sent until this string is
109
self._expect_body_tail = expect_body_tail
112
self.received_bytes = ''
115
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
116
self._sock.bind(('127.0.0.1', 0))
117
self.host, self.port = self._sock.getsockname()
118
self._ready = threading.Event()
119
self._thread = threading.Thread(target=self._accept_read_and_reply)
120
self._thread.setDaemon(True)
124
def _accept_read_and_reply(self):
127
self._sock.settimeout(5)
129
conn, address = self._sock.accept()
130
# On win32, the accepted connection will be non-blocking to start
131
# with because we're using settimeout.
132
conn.setblocking(True)
133
while not self.received_bytes.endswith(self._expect_body_tail):
134
self.received_bytes += conn.recv(4096)
135
conn.sendall('HTTP/1.1 200 OK\r\n')
136
except socket.timeout:
137
# Make sure the client isn't stuck waiting for us to e.g. accept.
140
# The client may have already closed the socket.
147
# We might have already closed it. We don't care.
153
class TestWithTransport_pycurl(object):
154
"""Test case to inherit from if pycurl is present"""
156
def _get_pycurl_maybe(self):
158
from bzrlib.transport.http._pycurl import PyCurlTransport
159
return PyCurlTransport
160
except errors.DependencyNotPresent:
161
raise TestSkipped('pycurl not present')
163
_transport = property(_get_pycurl_maybe)
166
class TestHttpUrls(TestCase):
168
# TODO: This should be moved to authorization tests once they
171
def test_url_parsing(self):
173
url = extract_auth('http://example.com', f)
174
self.assertEquals('http://example.com', url)
175
self.assertEquals(0, len(f.credentials))
176
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
177
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
178
self.assertEquals(1, len(f.credentials))
179
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
183
class TestHttpTransportUrls(object):
184
"""Test the http urls.
186
This MUST be used by daughter classes that also inherit from
189
We can't inherit directly from TestCase or the
190
test framework will try to create an instance which cannot
191
run, its implementation being incomplete.
194
def test_abs_url(self):
195
"""Construction of absolute http URLs"""
196
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
197
eq = self.assertEqualDiff
198
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
199
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
200
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
201
eq(t.abspath('.bzr/1//2/./3'),
202
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
204
def test_invalid_http_urls(self):
205
"""Trap invalid construction of urls"""
206
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
207
self.assertRaises(errors.InvalidURL,
209
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
211
def test_http_root_urls(self):
212
"""Construction of URLs from server root"""
213
t = self._transport('http://bzr.ozlabs.org/')
214
eq = self.assertEqualDiff
215
eq(t.abspath('.bzr/tree-version'),
216
'http://bzr.ozlabs.org/.bzr/tree-version')
218
def test_http_impl_urls(self):
219
"""There are servers which ask for particular clients to connect"""
220
server = self._server()
223
url = server.get_url()
224
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
229
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
230
"""Test http urls with urllib"""
232
_transport = HttpTransport_urllib
233
_server = HttpServer_urllib
234
_qualified_prefix = 'http+urllib'
237
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
239
"""Test http urls with pycurl"""
241
_server = HttpServer_PyCurl
242
_qualified_prefix = 'http+pycurl'
244
# TODO: This should really be moved into another pycurl
245
# specific test. When https tests will be implemented, take
246
# this one into account.
247
def test_pycurl_without_https_support(self):
248
"""Test that pycurl without SSL do not fail with a traceback.
250
For the purpose of the test, we force pycurl to ignore
251
https by supplying a fake version_info that do not
257
raise TestSkipped('pycurl not present')
258
# Now that we have pycurl imported, we can fake its version_info
259
# This was taken from a windows pycurl without SSL
261
pycurl.version_info = lambda : (2,
269
('ftp', 'gopher', 'telnet',
270
'dict', 'ldap', 'http', 'file'),
274
self.assertRaises(errors.DependencyNotPresent, self._transport,
275
'https://launchpad.net')
277
class TestHttpConnections(object):
278
"""Test the http connections.
280
This MUST be used by daughter classes that also inherit from
281
TestCaseWithWebserver.
283
We can't inherit directly from TestCaseWithWebserver or the
284
test framework will try to create an instance which cannot
285
run, its implementation being incomplete.
289
TestCaseWithWebserver.setUp(self)
290
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
291
transport=self.get_transport())
293
def test_http_has(self):
294
server = self.get_readonly_server()
295
t = self._transport(server.get_url())
296
self.assertEqual(t.has('foo/bar'), True)
297
self.assertEqual(len(server.logs), 1)
298
self.assertContainsRe(server.logs[0],
299
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
301
def test_http_has_not_found(self):
302
server = self.get_readonly_server()
303
t = self._transport(server.get_url())
304
self.assertEqual(t.has('not-found'), False)
305
self.assertContainsRe(server.logs[1],
306
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
308
def test_http_get(self):
309
server = self.get_readonly_server()
310
t = self._transport(server.get_url())
311
fp = t.get('foo/bar')
312
self.assertEqualDiff(
314
'contents of foo/bar\n')
315
self.assertEqual(len(server.logs), 1)
316
self.assertTrue(server.logs[0].find(
317
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
318
% bzrlib.__version__) > -1)
320
def test_get_smart_medium(self):
321
# For HTTP, get_smart_medium should return the transport object.
322
server = self.get_readonly_server()
323
http_transport = self._transport(server.get_url())
324
medium = http_transport.get_smart_medium()
325
self.assertIs(medium, http_transport)
327
def test_has_on_bogus_host(self):
328
# Get a free address and don't 'accept' on it, so that we
329
# can be sure there is no http handler there, but set a
330
# reasonable timeout to not slow down tests too much.
331
default_timeout = socket.getdefaulttimeout()
333
socket.setdefaulttimeout(2)
335
s.bind(('localhost', 0))
336
t = self._transport('http://%s:%s/' % s.getsockname())
337
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
339
socket.setdefaulttimeout(default_timeout)
342
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
343
"""Test http connections with urllib"""
345
_transport = HttpTransport_urllib
349
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
351
TestCaseWithWebserver):
352
"""Test http connections with pycurl"""
355
class TestHttpTransportRegistration(TestCase):
356
"""Test registrations of various http implementations"""
358
def test_http_registered(self):
359
# urlllib should always be present
360
t = get_transport('http+urllib://bzr.google.com/')
361
self.assertIsInstance(t, Transport)
362
self.assertIsInstance(t, HttpTransport_urllib)
365
class TestPost(object):
367
def _test_post_body_is_received(self, scheme):
368
server = RecordingServer(expect_body_tail='end-of-body')
370
self.addCleanup(server.tearDown)
371
url = '%s://%s:%s/' % (scheme, server.host, server.port)
373
http_transport = get_transport(url)
374
except errors.UnsupportedProtocol:
375
raise TestSkipped('%s not available' % scheme)
376
code, response = http_transport._post('abc def end-of-body')
378
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
379
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
380
# The transport should not be assuming that the server can accept
381
# chunked encoding the first time it connects, because HTTP/1.1, so we
382
# check for the literal string.
384
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
387
class TestPost_urllib(TestCase, TestPost):
388
"""TestPost for urllib implementation"""
390
_transport = HttpTransport_urllib
392
def test_post_body_is_received_urllib(self):
393
self._test_post_body_is_received('http+urllib')
396
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
397
"""TestPost for pycurl implementation"""
399
def test_post_body_is_received_pycurl(self):
400
self._test_post_body_is_received('http+pycurl')
403
class TestRangeHeader(TestCase):
404
"""Test range_header method"""
406
def check_header(self, value, ranges=[], tail=0):
407
offsets = [ (start, end - start + 1) for start, end in ranges]
408
coalesce = Transport._coalesce_offsets
409
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
410
range_header = HttpTransportBase._range_header
411
self.assertEqual(value, range_header(coalesced, tail))
413
def test_range_header_single(self):
414
self.check_header('0-9', ranges=[(0,9)])
415
self.check_header('100-109', ranges=[(100,109)])
417
def test_range_header_tail(self):
418
self.check_header('-10', tail=10)
419
self.check_header('-50', tail=50)
421
def test_range_header_multi(self):
422
self.check_header('0-9,100-200,300-5000',
423
ranges=[(0,9), (100, 200), (300,5000)])
425
def test_range_header_mixed(self):
426
self.check_header('0-9,300-5000,-50',
427
ranges=[(0,9), (300,5000)],
431
class TestWallServer(object):
432
"""Tests exceptions during the connection phase"""
434
def create_transport_readonly_server(self):
435
return HttpServer(WallRequestHandler)
437
def test_http_has(self):
438
server = self.get_readonly_server()
439
t = self._transport(server.get_url())
440
# Unfortunately httplib (see HTTPResponse._read_status
441
# for details) make no distinction between a closed
442
# socket and badly formatted status line, so we can't
443
# just test for ConnectionError, we have to test
444
# InvalidHttpResponse too.
445
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
448
def test_http_get(self):
449
server = self.get_readonly_server()
450
t = self._transport(server.get_url())
451
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
455
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
456
"""Tests "wall" server for urllib implementation"""
458
_transport = HttpTransport_urllib
461
class TestWallServer_pycurl(TestWithTransport_pycurl,
463
TestCaseWithWebserver):
464
"""Tests "wall" server for pycurl implementation"""
467
class TestBadStatusServer(object):
468
"""Tests bad status from server."""
470
def create_transport_readonly_server(self):
471
return HttpServer(BadStatusRequestHandler)
473
def test_http_has(self):
474
server = self.get_readonly_server()
475
t = self._transport(server.get_url())
476
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
478
def test_http_get(self):
479
server = self.get_readonly_server()
480
t = self._transport(server.get_url())
481
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
484
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
485
"""Tests bad status server for urllib implementation"""
487
_transport = HttpTransport_urllib
490
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
492
TestCaseWithWebserver):
493
"""Tests bad status server for pycurl implementation"""
496
class TestInvalidStatusServer(TestBadStatusServer):
497
"""Tests invalid status from server.
499
Both implementations raises the same error as for a bad status.
502
def create_transport_readonly_server(self):
503
return HttpServer(InvalidStatusRequestHandler)
506
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
507
TestCaseWithWebserver):
508
"""Tests invalid status server for urllib implementation"""
510
_transport = HttpTransport_urllib
513
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
514
TestInvalidStatusServer,
515
TestCaseWithWebserver):
516
"""Tests invalid status server for pycurl implementation"""
519
class TestBadProtocolServer(object):
520
"""Tests bad protocol from server."""
522
def create_transport_readonly_server(self):
523
return HttpServer(BadProtocolRequestHandler)
525
def test_http_has(self):
526
server = self.get_readonly_server()
527
t = self._transport(server.get_url())
528
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
530
def test_http_get(self):
531
server = self.get_readonly_server()
532
t = self._transport(server.get_url())
533
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
536
class TestBadProtocolServer_urllib(TestBadProtocolServer,
537
TestCaseWithWebserver):
538
"""Tests bad protocol server for urllib implementation"""
540
_transport = HttpTransport_urllib
542
# curl don't check the protocol version
543
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
544
# TestBadProtocolServer,
545
# TestCaseWithWebserver):
546
# """Tests bad protocol server for pycurl implementation"""
549
class TestForbiddenServer(object):
550
"""Tests forbidden server"""
552
def create_transport_readonly_server(self):
553
return HttpServer(ForbiddenRequestHandler)
555
def test_http_has(self):
556
server = self.get_readonly_server()
557
t = self._transport(server.get_url())
558
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
560
def test_http_get(self):
561
server = self.get_readonly_server()
562
t = self._transport(server.get_url())
563
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
566
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
567
"""Tests forbidden server for urllib implementation"""
569
_transport = HttpTransport_urllib
572
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
574
TestCaseWithWebserver):
575
"""Tests forbidden server for pycurl implementation"""
578
class TestRecordingServer(TestCase):
580
def test_create(self):
581
server = RecordingServer(expect_body_tail=None)
582
self.assertEqual('', server.received_bytes)
583
self.assertEqual(None, server.host)
584
self.assertEqual(None, server.port)
586
def test_setUp_and_tearDown(self):
587
server = RecordingServer(expect_body_tail=None)
590
self.assertNotEqual(None, server.host)
591
self.assertNotEqual(None, server.port)
594
self.assertEqual(None, server.host)
595
self.assertEqual(None, server.port)
597
def test_send_receive_bytes(self):
598
server = RecordingServer(expect_body_tail='c')
600
self.addCleanup(server.tearDown)
601
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
602
sock.connect((server.host, server.port))
604
self.assertEqual('HTTP/1.1 200 OK\r\n',
605
osutils.recv_all(sock, 4096))
606
self.assertEqual('abc', server.received_bytes)
609
class TestRangeRequestServer(object):
610
"""Tests readv requests against server.
612
This MUST be used by daughter classes that also inherit from
613
TestCaseWithWebserver.
615
We can't inherit directly from TestCaseWithWebserver or the
616
test framework will try to create an instance which cannot
617
run, its implementation being incomplete.
621
TestCaseWithWebserver.setUp(self)
622
self.build_tree_contents([('a', '0123456789')],)
624
def test_readv(self):
625
server = self.get_readonly_server()
626
t = self._transport(server.get_url())
627
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
628
self.assertEqual(l[0], (0, '0'))
629
self.assertEqual(l[1], (1, '1'))
630
self.assertEqual(l[2], (3, '34'))
631
self.assertEqual(l[3], (9, '9'))
633
def test_readv_out_of_order(self):
634
server = self.get_readonly_server()
635
t = self._transport(server.get_url())
636
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
637
self.assertEqual(l[0], (1, '1'))
638
self.assertEqual(l[1], (9, '9'))
639
self.assertEqual(l[2], (0, '0'))
640
self.assertEqual(l[3], (3, '34'))
642
def test_readv_invalid_ranges(self):
643
server = self.get_readonly_server()
644
t = self._transport(server.get_url())
646
# This is intentionally reading off the end of the file
647
# since we are sure that it cannot get there
648
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
649
t.readv, 'a', [(1,1), (8,10)])
651
# This is trying to seek past the end of the file, it should
652
# also raise a special error
653
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
654
t.readv, 'a', [(12,2)])
657
class TestSingleRangeRequestServer(TestRangeRequestServer):
658
"""Test readv against a server which accept only single range requests"""
660
def create_transport_readonly_server(self):
661
return HttpServer(SingleRangeRequestHandler)
664
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
665
TestCaseWithWebserver):
666
"""Tests single range requests accepting server for urllib implementation"""
668
_transport = HttpTransport_urllib
671
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
672
TestSingleRangeRequestServer,
673
TestCaseWithWebserver):
674
"""Tests single range requests accepting server for pycurl implementation"""
677
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
678
"""Test readv against a server which only accept single range requests"""
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleOnlyRangeRequestHandler)
684
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
688
_transport = HttpTransport_urllib
691
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleOnlyRangeRequestServer,
693
TestCaseWithWebserver):
694
"""Tests single range requests accepting server for pycurl implementation"""
697
class TestNoRangeRequestServer(TestRangeRequestServer):
698
"""Test readv against a server which do not accept range requests"""
700
def create_transport_readonly_server(self):
701
return HttpServer(NoRangeRequestHandler)
704
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests range requests refusing server for urllib implementation"""
708
_transport = HttpTransport_urllib
711
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestNoRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests range requests refusing server for pycurl implementation"""
717
class TestLimitedRangeRequestServer(object):
718
"""Tests readv requests against server that errors out on too much ranges.
720
This MUST be used by daughter classes that also inherit from
721
TestCaseWithWebserver.
723
We can't inherit directly from TestCaseWithWebserver or the
724
test framework will try to create an instance which cannot
725
run, its implementation being incomplete.
730
def create_transport_readonly_server(self):
731
# Requests with more range specifiers will error out
732
return LimitedRangeHTTPServer(range_limit=self.range_limit)
734
def get_transport(self):
735
return self._transport(self.get_readonly_server().get_url())
738
TestCaseWithWebserver.setUp(self)
739
# We need to manipulate ranges that correspond to real chunks in the
740
# response, so we build a content appropriately.
741
filler = ''.join(['abcdefghij' for _ in range(102)])
742
content = ''.join(['%04d' % v + filler for v in range(16)])
743
self.build_tree_contents([('a', content)],)
745
def test_few_ranges(self):
746
t = self.get_transport()
747
l = list(t.readv('a', ((0, 4), (1024, 4), )))
748
self.assertEqual(l[0], (0, '0000'))
749
self.assertEqual(l[1], (1024, '0001'))
750
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
752
def test_a_lot_of_ranges(self):
753
t = self.get_transport()
754
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
755
self.assertEqual(l[0], (0, '0000'))
756
self.assertEqual(l[1], (1024, '0001'))
757
self.assertEqual(l[2], (4096, '0004'))
758
self.assertEqual(l[3], (8192, '0008'))
759
# The server will refuse to serve the first request (too much ranges),
760
# a second request will succeeds.
761
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
764
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
765
TestCaseWithWebserver):
766
"""Tests limited range requests server for urllib implementation"""
768
_transport = HttpTransport_urllib
771
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
772
TestLimitedRangeRequestServer,
773
TestCaseWithWebserver):
774
"""Tests limited range requests server for pycurl implementation"""
778
class TestHttpProxyWhiteBox(TestCase):
779
"""Whitebox test proxy http authorization.
781
Only the urllib implementation is tested here.
791
def _install_env(self, env):
792
for name, value in env.iteritems():
793
self._old_env[name] = osutils.set_or_unset_env(name, value)
795
def _restore_env(self):
796
for name, value in self._old_env.iteritems():
797
osutils.set_or_unset_env(name, value)
799
def _proxied_request(self):
800
handler = ProxyHandler()
801
request = Request('GET','http://baz/buzzle')
802
handler.set_proxy(request, 'http')
805
def test_empty_user(self):
806
self._install_env({'http_proxy': 'http://bar.com'})
807
request = self._proxied_request()
808
self.assertFalse(request.headers.has_key('Proxy-authorization'))
810
def test_invalid_proxy(self):
811
"""A proxy env variable without scheme"""
812
self._install_env({'http_proxy': 'host:1234'})
813
self.assertRaises(errors.InvalidURL, self._proxied_request)
816
class TestProxyHttpServer(object):
817
"""Tests proxy server.
819
This MUST be used by daughter classes that also inherit from
820
TestCaseWithTwoWebservers.
822
We can't inherit directly from TestCaseWithTwoWebservers or
823
the test framework will try to create an instance which
824
cannot run, its implementation being incomplete.
826
Be aware that we do not setup a real proxy here. Instead, we
827
check that the *connection* goes through the proxy by serving
828
different content (the faked proxy server append '-proxied'
832
# FIXME: We don't have an https server available, so we don't
833
# test https connections.
836
TestCaseWithTwoWebservers.setUp(self)
837
self.build_tree_contents([('foo', 'contents of foo\n'),
838
('foo-proxied', 'proxied contents of foo\n')])
839
# Let's setup some attributes for tests
840
self.server = self.get_readonly_server()
841
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
842
self.no_proxy_host = self.proxy_address
843
# The secondary server is the proxy
844
self.proxy = self.get_secondary_server()
845
self.proxy_url = self.proxy.get_url()
848
def create_transport_secondary_server(self):
849
"""Creates an http server that will serve files with
850
'-proxied' appended to their names.
854
def _install_env(self, env):
855
for name, value in env.iteritems():
856
self._old_env[name] = osutils.set_or_unset_env(name, value)
858
def _restore_env(self):
859
for name, value in self._old_env.iteritems():
860
osutils.set_or_unset_env(name, value)
862
def proxied_in_env(self, env):
863
self._install_env(env)
864
url = self.server.get_url()
865
t = self._transport(url)
867
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
871
def not_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(), 'contents of foo\n')
880
def test_http_proxy(self):
881
self.proxied_in_env({'http_proxy': self.proxy_url})
883
def test_HTTP_PROXY(self):
884
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
886
def test_all_proxy(self):
887
self.proxied_in_env({'all_proxy': self.proxy_url})
889
def test_ALL_PROXY(self):
890
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
892
def test_http_proxy_with_no_proxy(self):
893
self.not_proxied_in_env({'http_proxy': self.proxy_url,
894
'no_proxy': self.no_proxy_host})
896
def test_HTTP_PROXY_with_NO_PROXY(self):
897
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
898
'NO_PROXY': self.no_proxy_host})
900
def test_all_proxy_with_no_proxy(self):
901
self.not_proxied_in_env({'all_proxy': self.proxy_url,
902
'no_proxy': self.no_proxy_host})
904
def test_ALL_PROXY_with_NO_PROXY(self):
905
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
906
'NO_PROXY': self.no_proxy_host})
908
def test_http_proxy_without_scheme(self):
909
self.assertRaises(errors.InvalidURL,
911
{'http_proxy': self.proxy_address})
914
class TestProxyHttpServer_urllib(TestProxyHttpServer,
915
TestCaseWithTwoWebservers):
916
"""Tests proxy server for urllib implementation"""
918
_transport = HttpTransport_urllib
921
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
923
TestCaseWithTwoWebservers):
924
"""Tests proxy server for pycurl implementation"""
927
TestProxyHttpServer.setUp(self)
928
# Oh my ! pycurl does not check for the port as part of
929
# no_proxy :-( So we just test the host part
930
self.no_proxy_host = 'localhost'
932
def test_HTTP_PROXY(self):
933
# pycurl do not check HTTP_PROXY for security reasons
934
# (for use in a CGI context that we do not care
935
# about. Should we ?)
938
def test_HTTP_PROXY_with_NO_PROXY(self):
941
def test_http_proxy_without_scheme(self):
942
# pycurl *ignores* invalid proxy env variables. If that
943
# ever change in the future, this test will fail
944
# indicating that pycurl do not ignore anymore such
946
self.not_proxied_in_env({'http_proxy': self.proxy_address})
949
class TestRanges(object):
950
"""Test the Range header in GET methods..
952
This MUST be used by daughter classes that also inherit from
953
TestCaseWithWebserver.
955
We can't inherit directly from TestCaseWithWebserver or the
956
test framework will try to create an instance which cannot
957
run, its implementation being incomplete.
961
TestCaseWithWebserver.setUp(self)
962
self.build_tree_contents([('a', '0123456789')],)
963
server = self.get_readonly_server()
964
self.transport = self._transport(server.get_url())
966
def _file_contents(self, relpath, ranges):
967
offsets = [ (start, end - start + 1) for start, end in ranges]
968
coalesce = self.transport._coalesce_offsets
969
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
970
code, data = self.transport._get(relpath, coalesced)
971
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
972
for start, end in ranges:
974
yield data.read(end - start + 1)
976
def _file_tail(self, relpath, tail_amount):
977
code, data = self.transport._get(relpath, [], tail_amount)
978
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
979
data.seek(-tail_amount + 1, 2)
980
return data.read(tail_amount)
982
def test_range_header(self):
984
map(self.assertEqual,['0', '234'],
985
list(self._file_contents('a', [(0,0), (2,4)])),)
987
self.assertEqual('789', self._file_tail('a', 3))
988
# Syntactically invalid range
989
self.assertListRaises(errors.InvalidRange,
990
self._file_contents, 'a', [(4, 3)])
991
# Semantically invalid range
992
self.assertListRaises(errors.InvalidRange,
993
self._file_contents, 'a', [(42, 128)])
996
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
997
"""Test the Range header in GET methods for urllib implementation"""
999
_transport = HttpTransport_urllib
1002
class TestRanges_pycurl(TestWithTransport_pycurl,
1004
TestCaseWithWebserver):
1005
"""Test the Range header in GET methods for pycurl implementation"""
1008
class TestHTTPRedirections(object):
1009
"""Test redirection between http servers.
1011
This MUST be used by daughter classes that also inherit from
1012
TestCaseWithRedirectedWebserver.
1014
We can't inherit directly from TestCaseWithTwoWebservers or the
1015
test framework will try to create an instance which cannot
1016
run, its implementation being incomplete.
1019
def create_transport_secondary_server(self):
1020
"""Create the secondary server redirecting to the primary server"""
1021
new = self.get_readonly_server()
1023
redirecting = HTTPServerRedirecting()
1024
redirecting.redirect_to(new.host, new.port)
1028
super(TestHTTPRedirections, self).setUp()
1029
self.build_tree_contents([('a', '0123456789'),
1031
'# Bazaar revision bundle v0.9\n#\n')
1034
self.old_transport = self._transport(self.old_server.get_url())
1036
def test_redirected(self):
1037
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1038
t = self._transport(self.new_server.get_url())
1039
self.assertEqual('0123456789', t.get('a').read())
1041
def test_read_redirected_bundle_from_url(self):
1042
from bzrlib.bundle import read_bundle_from_url
1043
url = self.old_transport.abspath('bundle')
1044
bundle = read_bundle_from_url(url)
1045
# If read_bundle_from_url was successful we get an empty bundle
1046
self.assertEqual([], bundle.revisions)
1049
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1050
TestCaseWithRedirectedWebserver):
1051
"""Tests redirections for urllib implementation"""
1053
_transport = HttpTransport_urllib
1057
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1058
TestHTTPRedirections,
1059
TestCaseWithRedirectedWebserver):
1060
"""Tests redirections for pycurl implementation"""
1063
class RedirectedRequest(Request):
1064
"""Request following redirections"""
1066
init_orig = Request.__init__
1068
def __init__(self, method, url, *args, **kwargs):
1069
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1070
self.follow_redirections = True
1073
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1074
"""Test redirections provided by urllib.
1076
http implementations do not redirect silently anymore (they
1077
do not redirect at all in fact). The mechanism is still in
1078
place at the _urllib2_wrappers.Request level and these tests
1081
For the pycurl implementation
1082
the redirection have been deleted as we may deprecate pycurl
1083
and I have no place to keep a working implementation.
1087
_transport = HttpTransport_urllib
1090
super(TestHTTPSilentRedirections_urllib, self).setUp()
1091
self.setup_redirected_request()
1092
self.addCleanup(self.cleanup_redirected_request)
1093
self.build_tree_contents([('a','a'),
1095
('1/a', 'redirected once'),
1097
('2/a', 'redirected twice'),
1099
('3/a', 'redirected thrice'),
1101
('4/a', 'redirected 4 times'),
1103
('5/a', 'redirected 5 times'),
1106
self.old_transport = self._transport(self.old_server.get_url())
1108
def setup_redirected_request(self):
1109
self.original_class = _urllib2_wrappers.Request
1110
_urllib2_wrappers.Request = RedirectedRequest
1112
def cleanup_redirected_request(self):
1113
_urllib2_wrappers.Request = self.original_class
1115
def create_transport_secondary_server(self):
1116
"""Create the secondary server, redirections are defined in the tests"""
1117
return HTTPServerRedirecting()
1119
def test_one_redirection(self):
1120
t = self.old_transport
1122
req = RedirectedRequest('GET', t.abspath('a'))
1123
req.follow_redirections = True
1124
new_prefix = 'http://%s:%s' % (self.new_server.host,
1125
self.new_server.port)
1126
self.old_server.redirections = \
1127
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1128
self.assertEquals('redirected once',t._perform(req).read())
1130
def test_five_redirections(self):
1131
t = self.old_transport
1133
req = RedirectedRequest('GET', t.abspath('a'))
1134
req.follow_redirections = True
1135
old_prefix = 'http://%s:%s' % (self.old_server.host,
1136
self.old_server.port)
1137
new_prefix = 'http://%s:%s' % (self.new_server.host,
1138
self.new_server.port)
1139
self.old_server.redirections = \
1140
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1141
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1142
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1143
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1144
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1146
self.assertEquals('redirected 5 times',t._perform(req).read())
1149
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1150
"""Test transport.do_catching_redirections.
1152
We arbitrarily choose to use urllib transports
1155
_transport = HttpTransport_urllib
1158
super(TestDoCatchRedirections, self).setUp()
1159
self.build_tree_contents([('a', '0123456789'),],)
1161
self.old_transport = self._transport(self.old_server.get_url())
1163
def get_a(self, transport):
1164
return transport.get('a')
1166
def test_no_redirection(self):
1167
t = self._transport(self.new_server.get_url())
1169
# We use None for redirected so that we fail if redirected
1170
self.assertEquals('0123456789',
1171
do_catching_redirections(self.get_a, t, None).read())
1173
def test_one_redirection(self):
1174
self.redirections = 0
1176
def redirected(transport, exception, redirection_notice):
1177
self.redirections += 1
1178
dir, file = urlutils.split(exception.target)
1179
return self._transport(dir)
1181
self.assertEquals('0123456789',
1182
do_catching_redirections(self.get_a,
1186
self.assertEquals(1, self.redirections)
1188
def test_redirection_loop(self):
1190
def redirected(transport, exception, redirection_notice):
1191
# By using the redirected url as a base dir for the
1192
# *old* transport, we create a loop: a => a/a =>
1194
return self.old_transport.clone(exception.target)
1196
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1197
self.get_a, self.old_transport, redirected)
1200
class TestAuth(object):
1201
"""Test some authentication scheme specified by daughter class.
1203
This MUST be used by daughter classes that also inherit from
1204
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1207
_password_prompt_prefix = ''
1210
"""Set up the test environment
1212
Daughter classes should set up their own environment
1213
(including self.server) and explicitely call this
1214
method. This is needed because we want to reuse the same
1215
tests for proxy and no-proxy accesses which have
1216
different ways of setting self.server.
1218
self.build_tree_contents([('a', 'contents of a\n'),
1219
('b', 'contents of b\n'),])
1221
def get_user_url(self, user=None, password=None):
1222
"""Build an url embedding user and password"""
1223
url = '%s://' % self.server._url_protocol
1224
if user is not None:
1226
if password is not None:
1227
url += ':' + password
1229
url += '%s:%s/' % (self.server.host, self.server.port)
1232
def test_no_user(self):
1233
self.server.add_user('joe', 'foo')
1234
t = self.get_user_transport()
1235
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1236
# Only one 'Authentication Required' error should occur
1237
self.assertEqual(1, self.server.auth_required_errors)
1239
def test_empty_pass(self):
1240
self.server.add_user('joe', '')
1241
t = self.get_user_transport('joe', '')
1242
self.assertEqual('contents of a\n', t.get('a').read())
1243
# Only one 'Authentication Required' error should occur
1244
self.assertEqual(1, self.server.auth_required_errors)
1246
def test_user_pass(self):
1247
self.server.add_user('joe', 'foo')
1248
t = self.get_user_transport('joe', 'foo')
1249
self.assertEqual('contents of a\n', t.get('a').read())
1250
# Only one 'Authentication Required' error should occur
1251
self.assertEqual(1, self.server.auth_required_errors)
1253
def test_unknown_user(self):
1254
self.server.add_user('joe', 'foo')
1255
t = self.get_user_transport('bill', 'foo')
1256
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1257
# Two 'Authentication Required' errors should occur (the
1258
# initial 'who are you' and 'I don't know you, who are
1260
self.assertEqual(2, self.server.auth_required_errors)
1262
def test_wrong_pass(self):
1263
self.server.add_user('joe', 'foo')
1264
t = self.get_user_transport('joe', 'bar')
1265
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1266
# Two 'Authentication Required' errors should occur (the
1267
# initial 'who are you' and 'this is not you, who are you')
1268
self.assertEqual(2, self.server.auth_required_errors)
1270
def test_prompt_for_password(self):
1271
self.server.add_user('joe', 'foo')
1272
t = self.get_user_transport('joe', None)
1273
stdout = StringIOWrapper()
1274
ui.ui_factory = TestUIFactory(stdin='foo\n', stdout=stdout)
1275
self.assertEqual('contents of a\n',t.get('a').read())
1276
# stdin should be empty
1277
self.assertEqual('', ui.ui_factory.stdin.readline())
1278
self._check_password_prompt(t._unqualified_scheme, 'joe',
1280
# And we shouldn't prompt again for a different request
1281
# against the same transport.
1282
self.assertEqual('contents of b\n',t.get('b').read())
1284
# And neither against a clone
1285
self.assertEqual('contents of b\n',t2.get('b').read())
1286
# Only one 'Authentication Required' error should occur
1287
self.assertEqual(1, self.server.auth_required_errors)
1289
def _check_password_prompt(self, scheme, user, actual_prompt):
1290
expected_prompt = (self._password_prompt_prefix
1291
+ ("%s %s@%s:%d, Realm: '%s' password: "
1293
user, self.server.host, self.server.port,
1294
self.server.auth_realm)))
1295
self.assertEquals(expected_prompt, actual_prompt)
1297
def test_no_prompt_for_password_when_using_auth_config(self):
1300
stdin_content = 'bar\n' # Not the right password
1301
self.server.add_user(user, password)
1302
t = self.get_user_transport(user, None)
1303
ui.ui_factory = TestUIFactory(stdin=stdin_content,
1304
stdout=StringIOWrapper())
1305
# Create a minimal config file with the right password
1306
conf = config.AuthenticationConfig()
1307
conf._get_config().update(
1308
{'httptest': {'scheme': 'http', 'port': self.server.port,
1309
'user': user, 'password': password}})
1311
# Issue a request to the server to connect
1312
self.assertEqual('contents of a\n',t.get('a').read())
1313
# stdin should have been left untouched
1314
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1315
# Only one 'Authentication Required' error should occur
1316
self.assertEqual(1, self.server.auth_required_errors)
1320
class TestHTTPAuth(TestAuth):
1321
"""Test HTTP authentication schemes.
1323
Daughter classes MUST inherit from TestCaseWithWebserver too.
1326
_auth_header = 'Authorization'
1329
TestCaseWithWebserver.setUp(self)
1330
self.server = self.get_readonly_server()
1331
TestAuth.setUp(self)
1333
def get_user_transport(self, user=None, password=None):
1334
return self._transport(self.get_user_url(user, password))
1337
class TestProxyAuth(TestAuth):
1338
"""Test proxy authentication schemes.
1340
Daughter classes MUST also inherit from TestCaseWithWebserver.
1342
_auth_header = 'Proxy-authorization'
1343
_password_prompt_prefix = 'Proxy '
1347
TestCaseWithWebserver.setUp(self)
1348
self.server = self.get_readonly_server()
1350
self.addCleanup(self._restore_env)
1351
TestAuth.setUp(self)
1352
# Override the contents to avoid false positives
1353
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1354
('b', 'not proxied contents of b\n'),
1355
('a-proxied', 'contents of a\n'),
1356
('b-proxied', 'contents of b\n'),
1359
def get_user_transport(self, user=None, password=None):
1360
self._install_env({'all_proxy': self.get_user_url(user, password)})
1361
return self._transport(self.server.get_url())
1363
def _install_env(self, env):
1364
for name, value in env.iteritems():
1365
self._old_env[name] = osutils.set_or_unset_env(name, value)
1367
def _restore_env(self):
1368
for name, value in self._old_env.iteritems():
1369
osutils.set_or_unset_env(name, value)
1372
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1373
"""Test http basic authentication scheme"""
1375
_transport = HttpTransport_urllib
1377
def create_transport_readonly_server(self):
1378
return HTTPBasicAuthServer()
1381
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1382
"""Test proxy basic authentication scheme"""
1384
_transport = HttpTransport_urllib
1386
def create_transport_readonly_server(self):
1387
return ProxyBasicAuthServer()
1390
class TestDigestAuth(object):
1391
"""Digest Authentication specific tests"""
1393
def test_changing_nonce(self):
1394
self.server.add_user('joe', 'foo')
1395
t = self.get_user_transport('joe', 'foo')
1396
self.assertEqual('contents of a\n', t.get('a').read())
1397
self.assertEqual('contents of b\n', t.get('b').read())
1398
# Only one 'Authentication Required' error should have
1400
self.assertEqual(1, self.server.auth_required_errors)
1401
# The server invalidates the current nonce
1402
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1403
self.assertEqual('contents of a\n', t.get('a').read())
1404
# Two 'Authentication Required' errors should occur (the
1405
# initial 'who are you' and a second 'who are you' with the new nonce)
1406
self.assertEqual(2, self.server.auth_required_errors)
1409
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1410
"""Test http digest authentication scheme"""
1412
_transport = HttpTransport_urllib
1414
def create_transport_readonly_server(self):
1415
return HTTPDigestAuthServer()
1418
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1419
TestCaseWithWebserver):
1420
"""Test proxy digest authentication scheme"""
1422
_transport = HttpTransport_urllib
1424
def create_transport_readonly_server(self):
1425
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
