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
40
from bzrlib.tests import (
48
from bzrlib.tests.http_utils import (
49
BadProtocolRequestHandler,
50
BadStatusRequestHandler,
51
ForbiddenRequestHandler,
54
HTTPServerRedirecting,
55
InvalidStatusRequestHandler,
56
LimitedRangeHTTPServer,
57
NoRangeRequestHandler,
59
ProxyDigestAuthServer,
61
SingleRangeRequestHandler,
62
SingleOnlyRangeRequestHandler,
63
TestCaseWithRedirectedWebserver,
64
TestCaseWithTwoWebservers,
65
TestCaseWithWebserver,
68
from bzrlib.transport import (
70
do_catching_redirections,
74
from bzrlib.transport.http import (
79
from bzrlib.transport.http._urllib import HttpTransport_urllib
80
from bzrlib.transport.http._urllib2_wrappers import (
86
class FakeManager(object):
91
def add_password(self, realm, host, username, password):
92
self.credentials.append([realm, host, username, password])
95
class TestHTTPServer(tests.TestCase):
96
"""Test the HTTP servers implementations."""
98
def test_invalid_protocol(self):
99
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
101
protocol_version = 'HTTP/0.1'
103
server = http_server.HttpServer(BogusRequestHandler)
105
self.assertRaises(httplib.UnknownProtocol,server.setUp)
108
self.fail('HTTP Server creation did not raise UnknownProtocol')
110
def test_server_start_and_stop(self):
111
server = http_server.HttpServer()
113
self.assertTrue(server._http_running)
115
self.assertFalse(server._http_running)
117
def test_create_http_server_one_zero(self):
118
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
120
protocol_version = 'HTTP/1.0'
122
server = http_server.HttpServer(RequestHandlerOneZero)
125
self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
129
def test_create_http_server_one_one(self):
130
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
132
protocol_version = 'HTTP/1.1'
134
server = http_server.HttpServer(RequestHandlerOneOne)
137
self.assertIsInstance(server._httpd,
138
http_server.TestingThreadingHTTPServer)
143
class TestWithTransport_pycurl(object):
144
"""Test case to inherit from if pycurl is present"""
146
def _get_pycurl_maybe(self):
148
from bzrlib.transport.http._pycurl import PyCurlTransport
149
return PyCurlTransport
150
except errors.DependencyNotPresent:
151
raise tests.TestSkipped('pycurl not present')
153
_transport = property(_get_pycurl_maybe)
156
class TestHttpUrls(tests.TestCase):
158
# TODO: This should be moved to authorization tests once they
161
def test_url_parsing(self):
163
url = extract_auth('http://example.com', f)
164
self.assertEquals('http://example.com', url)
165
self.assertEquals(0, len(f.credentials))
166
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
167
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
168
self.assertEquals(1, len(f.credentials))
169
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
173
class TestHttpUrls_pycurl(TestWithTransport_pycurl, tests.TestCase):
174
"""Test http urls with pycurl"""
176
_server = http_server.HttpServer_PyCurl
177
_qualified_prefix = 'http+pycurl'
179
# TODO: This should really be moved into another pycurl
180
# specific test. When https tests will be implemented, take
181
# this one into account.
182
def test_pycurl_without_https_support(self):
183
"""Test that pycurl without SSL do not fail with a traceback.
185
For the purpose of the test, we force pycurl to ignore
186
https by supplying a fake version_info that do not
192
raise tests.TestSkipped('pycurl not present')
193
# Now that we have pycurl imported, we can fake its version_info
194
# This was taken from a windows pycurl without SSL
196
pycurl.version_info = lambda : (2,
204
('ftp', 'gopher', 'telnet',
205
'dict', 'ldap', 'http', 'file'),
209
self.assertRaises(errors.DependencyNotPresent, self._transport,
210
'https://launchpad.net')
212
class TestHttpTransportRegistration(tests.TestCase):
213
"""Test registrations of various http implementations"""
215
def test_http_registered(self):
216
# urlllib should always be present
217
t = get_transport('http+urllib://bzr.google.com/')
218
self.assertIsInstance(t, Transport)
219
self.assertIsInstance(t, HttpTransport_urllib)
222
class TestRangeHeader(tests.TestCase):
223
"""Test range_header method"""
225
def check_header(self, value, ranges=[], tail=0):
226
offsets = [ (start, end - start + 1) for start, end in ranges]
227
coalesce = Transport._coalesce_offsets
228
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
229
range_header = HttpTransportBase._range_header
230
self.assertEqual(value, range_header(coalesced, tail))
232
def test_range_header_single(self):
233
self.check_header('0-9', ranges=[(0,9)])
234
self.check_header('100-109', ranges=[(100,109)])
236
def test_range_header_tail(self):
237
self.check_header('-10', tail=10)
238
self.check_header('-50', tail=50)
240
def test_range_header_multi(self):
241
self.check_header('0-9,100-200,300-5000',
242
ranges=[(0,9), (100, 200), (300,5000)])
244
def test_range_header_mixed(self):
245
self.check_header('0-9,300-5000,-50',
246
ranges=[(0,9), (300,5000)],
250
class TestWallServer(object):
251
"""Tests exceptions during the connection phase"""
253
def create_transport_readonly_server(self):
254
return http_server.HttpServer(WallRequestHandler)
256
def test_http_has(self):
257
server = self.get_readonly_server()
258
t = self._transport(server.get_url())
259
# Unfortunately httplib (see HTTPResponse._read_status
260
# for details) make no distinction between a closed
261
# socket and badly formatted status line, so we can't
262
# just test for ConnectionError, we have to test
263
# InvalidHttpResponse too.
264
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
267
def test_http_get(self):
268
server = self.get_readonly_server()
269
t = self._transport(server.get_url())
270
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
274
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
275
"""Tests "wall" server for urllib implementation"""
277
_transport = HttpTransport_urllib
280
class TestWallServer_pycurl(TestWithTransport_pycurl,
282
TestCaseWithWebserver):
283
"""Tests "wall" server for pycurl implementation"""
286
class TestBadStatusServer(object):
287
"""Tests bad status from server."""
289
def create_transport_readonly_server(self):
290
return http_server.HttpServer(BadStatusRequestHandler)
292
def test_http_has(self):
293
server = self.get_readonly_server()
294
t = self._transport(server.get_url())
295
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
297
def test_http_get(self):
298
server = self.get_readonly_server()
299
t = self._transport(server.get_url())
300
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
303
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
304
"""Tests bad status server for urllib implementation"""
306
_transport = HttpTransport_urllib
309
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
311
TestCaseWithWebserver):
312
"""Tests bad status server for pycurl implementation"""
315
class TestInvalidStatusServer(TestBadStatusServer):
316
"""Tests invalid status from server.
318
Both implementations raises the same error as for a bad status.
321
def create_transport_readonly_server(self):
322
return http_server.HttpServer(InvalidStatusRequestHandler)
325
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
326
TestCaseWithWebserver):
327
"""Tests invalid status server for urllib implementation"""
329
_transport = HttpTransport_urllib
332
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
333
TestInvalidStatusServer,
334
TestCaseWithWebserver):
335
"""Tests invalid status server for pycurl implementation"""
338
class TestBadProtocolServer(object):
339
"""Tests bad protocol from server."""
341
def create_transport_readonly_server(self):
342
return http_server.HttpServer(BadProtocolRequestHandler)
344
def test_http_has(self):
345
server = self.get_readonly_server()
346
t = self._transport(server.get_url())
347
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
349
def test_http_get(self):
350
server = self.get_readonly_server()
351
t = self._transport(server.get_url())
352
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
355
class TestBadProtocolServer_urllib(TestBadProtocolServer,
356
TestCaseWithWebserver):
357
"""Tests bad protocol server for urllib implementation"""
359
_transport = HttpTransport_urllib
361
# curl don't check the protocol version
362
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
363
# TestBadProtocolServer,
364
# TestCaseWithWebserver):
365
# """Tests bad protocol server for pycurl implementation"""
368
class TestForbiddenServer(object):
369
"""Tests forbidden server"""
371
def create_transport_readonly_server(self):
372
return http_server.HttpServer(ForbiddenRequestHandler)
374
def test_http_has(self):
375
server = self.get_readonly_server()
376
t = self._transport(server.get_url())
377
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
379
def test_http_get(self):
380
server = self.get_readonly_server()
381
t = self._transport(server.get_url())
382
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
385
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
386
"""Tests forbidden server for urllib implementation"""
388
_transport = HttpTransport_urllib
391
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
393
TestCaseWithWebserver):
394
"""Tests forbidden server for pycurl implementation"""
397
class TestRecordingServer(tests.TestCase):
399
def test_create(self):
400
server = http_utils.RecordingServer(expect_body_tail=None)
401
self.assertEqual('', server.received_bytes)
402
self.assertEqual(None, server.host)
403
self.assertEqual(None, server.port)
405
def test_setUp_and_tearDown(self):
406
server = http_utils.RecordingServer(expect_body_tail=None)
409
self.assertNotEqual(None, server.host)
410
self.assertNotEqual(None, server.port)
413
self.assertEqual(None, server.host)
414
self.assertEqual(None, server.port)
416
def test_send_receive_bytes(self):
417
server = http_utils.RecordingServer(expect_body_tail='c')
419
self.addCleanup(server.tearDown)
420
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
421
sock.connect((server.host, server.port))
423
self.assertEqual('HTTP/1.1 200 OK\r\n',
424
osutils.recv_all(sock, 4096))
425
self.assertEqual('abc', server.received_bytes)
428
class TestRangeRequestServer(object):
429
"""Tests readv requests against server.
431
This MUST be used by daughter classes that also inherit from
432
TestCaseWithWebserver.
434
We can't inherit directly from TestCaseWithWebserver or the
435
test framework will try to create an instance which cannot
436
run, its implementation being incomplete.
440
TestCaseWithWebserver.setUp(self)
441
self.build_tree_contents([('a', '0123456789')],)
443
def test_readv(self):
444
server = self.get_readonly_server()
445
t = self._transport(server.get_url())
446
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
447
self.assertEqual(l[0], (0, '0'))
448
self.assertEqual(l[1], (1, '1'))
449
self.assertEqual(l[2], (3, '34'))
450
self.assertEqual(l[3], (9, '9'))
452
def test_readv_out_of_order(self):
453
server = self.get_readonly_server()
454
t = self._transport(server.get_url())
455
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
456
self.assertEqual(l[0], (1, '1'))
457
self.assertEqual(l[1], (9, '9'))
458
self.assertEqual(l[2], (0, '0'))
459
self.assertEqual(l[3], (3, '34'))
461
def test_readv_invalid_ranges(self):
462
server = self.get_readonly_server()
463
t = self._transport(server.get_url())
465
# This is intentionally reading off the end of the file
466
# since we are sure that it cannot get there
467
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
468
t.readv, 'a', [(1,1), (8,10)])
470
# This is trying to seek past the end of the file, it should
471
# also raise a special error
472
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
473
t.readv, 'a', [(12,2)])
475
def test_readv_multiple_get_requests(self):
476
server = self.get_readonly_server()
477
t = self._transport(server.get_url())
478
# force transport to issue multiple requests
479
t._max_readv_combine = 1
480
t._max_get_ranges = 1
481
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
482
self.assertEqual(l[0], (0, '0'))
483
self.assertEqual(l[1], (1, '1'))
484
self.assertEqual(l[2], (3, '34'))
485
self.assertEqual(l[3], (9, '9'))
486
# The server should have issued 4 requests
487
self.assertEqual(4, server.GET_request_nb)
489
def test_readv_get_max_size(self):
490
server = self.get_readonly_server()
491
t = self._transport(server.get_url())
492
# force transport to issue multiple requests by limiting the number of
493
# bytes by request. Note that this apply to coalesced offsets only, a
494
# single range ill keep its size even if bigger than the limit.
496
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
497
self.assertEqual(l[0], (0, '0'))
498
self.assertEqual(l[1], (1, '1'))
499
self.assertEqual(l[2], (2, '2345'))
500
self.assertEqual(l[3], (6, '6789'))
501
# The server should have issued 3 requests
502
self.assertEqual(3, server.GET_request_nb)
505
class TestSingleRangeRequestServer(TestRangeRequestServer):
506
"""Test readv against a server which accept only single range requests"""
508
def create_transport_readonly_server(self):
509
return http_server.HttpServer(SingleRangeRequestHandler)
512
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
513
TestCaseWithWebserver):
514
"""Tests single range requests accepting server for urllib implementation"""
516
_transport = HttpTransport_urllib
519
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
520
TestSingleRangeRequestServer,
521
TestCaseWithWebserver):
522
"""Tests single range requests accepting server for pycurl implementation"""
525
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
526
"""Test readv against a server which only accept single range requests"""
528
def create_transport_readonly_server(self):
529
return http_server.HttpServer(SingleOnlyRangeRequestHandler)
532
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
533
TestCaseWithWebserver):
534
"""Tests single range requests accepting server for urllib implementation"""
536
_transport = HttpTransport_urllib
539
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
540
TestSingleOnlyRangeRequestServer,
541
TestCaseWithWebserver):
542
"""Tests single range requests accepting server for pycurl implementation"""
545
class TestNoRangeRequestServer(TestRangeRequestServer):
546
"""Test readv against a server which do not accept range requests"""
548
def create_transport_readonly_server(self):
549
return http_server.HttpServer(NoRangeRequestHandler)
552
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
553
TestCaseWithWebserver):
554
"""Tests range requests refusing server for urllib implementation"""
556
_transport = HttpTransport_urllib
559
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
560
TestNoRangeRequestServer,
561
TestCaseWithWebserver):
562
"""Tests range requests refusing server for pycurl implementation"""
565
class TestLimitedRangeRequestServer(object):
566
"""Tests readv requests against server that errors out on too much ranges.
568
This MUST be used by daughter classes that also inherit from
569
TestCaseWithWebserver.
571
We can't inherit directly from TestCaseWithWebserver or the
572
test framework will try to create an instance which cannot
573
run, its implementation being incomplete.
578
def create_transport_readonly_server(self):
579
# Requests with more range specifiers will error out
580
return LimitedRangeHTTPServer(range_limit=self.range_limit)
582
def get_transport(self):
583
return self._transport(self.get_readonly_server().get_url())
586
TestCaseWithWebserver.setUp(self)
587
# We need to manipulate ranges that correspond to real chunks in the
588
# response, so we build a content appropriately.
589
filler = ''.join(['abcdefghij' for x in range(102)])
590
content = ''.join(['%04d' % v + filler for v in range(16)])
591
self.build_tree_contents([('a', content)],)
593
def test_few_ranges(self):
594
t = self.get_transport()
595
l = list(t.readv('a', ((0, 4), (1024, 4), )))
596
self.assertEqual(l[0], (0, '0000'))
597
self.assertEqual(l[1], (1024, '0001'))
598
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
600
def test_more_ranges(self):
601
t = self.get_transport()
602
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
603
self.assertEqual(l[0], (0, '0000'))
604
self.assertEqual(l[1], (1024, '0001'))
605
self.assertEqual(l[2], (4096, '0004'))
606
self.assertEqual(l[3], (8192, '0008'))
607
# The server will refuse to serve the first request (too much ranges),
608
# a second request will succeeds.
609
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
612
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
613
TestCaseWithWebserver):
614
"""Tests limited range requests server for urllib implementation"""
616
_transport = HttpTransport_urllib
619
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
620
TestLimitedRangeRequestServer,
621
TestCaseWithWebserver):
622
"""Tests limited range requests server for pycurl implementation"""
626
class TestHttpProxyWhiteBox(tests.TestCase):
627
"""Whitebox test proxy http authorization.
629
Only the urllib implementation is tested here.
633
tests.TestCase.setUp(self)
639
def _install_env(self, env):
640
for name, value in env.iteritems():
641
self._old_env[name] = osutils.set_or_unset_env(name, value)
643
def _restore_env(self):
644
for name, value in self._old_env.iteritems():
645
osutils.set_or_unset_env(name, value)
647
def _proxied_request(self):
648
handler = ProxyHandler()
649
request = Request('GET','http://baz/buzzle')
650
handler.set_proxy(request, 'http')
653
def test_empty_user(self):
654
self._install_env({'http_proxy': 'http://bar.com'})
655
request = self._proxied_request()
656
self.assertFalse(request.headers.has_key('Proxy-authorization'))
658
def test_invalid_proxy(self):
659
"""A proxy env variable without scheme"""
660
self._install_env({'http_proxy': 'host:1234'})
661
self.assertRaises(errors.InvalidURL, self._proxied_request)
664
class TestProxyHttpServer(object):
665
"""Tests proxy server.
667
This MUST be used by daughter classes that also inherit from
668
TestCaseWithTwoWebservers.
670
We can't inherit directly from TestCaseWithTwoWebservers or
671
the test framework will try to create an instance which
672
cannot run, its implementation being incomplete.
674
Be aware that we do not setup a real proxy here. Instead, we
675
check that the *connection* goes through the proxy by serving
676
different content (the faked proxy server append '-proxied'
680
# FIXME: We don't have an https server available, so we don't
681
# test https connections.
684
TestCaseWithTwoWebservers.setUp(self)
685
self.build_tree_contents([('foo', 'contents of foo\n'),
686
('foo-proxied', 'proxied contents of foo\n')])
687
# Let's setup some attributes for tests
688
self.server = self.get_readonly_server()
689
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
690
self.no_proxy_host = self.proxy_address
691
# The secondary server is the proxy
692
self.proxy = self.get_secondary_server()
693
self.proxy_url = self.proxy.get_url()
696
def create_transport_secondary_server(self):
697
"""Creates an http server that will serve files with
698
'-proxied' appended to their names.
702
def _install_env(self, env):
703
for name, value in env.iteritems():
704
self._old_env[name] = osutils.set_or_unset_env(name, value)
706
def _restore_env(self):
707
for name, value in self._old_env.iteritems():
708
osutils.set_or_unset_env(name, value)
710
def proxied_in_env(self, env):
711
self._install_env(env)
712
url = self.server.get_url()
713
t = self._transport(url)
715
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
719
def not_proxied_in_env(self, env):
720
self._install_env(env)
721
url = self.server.get_url()
722
t = self._transport(url)
724
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
728
def test_http_proxy(self):
729
self.proxied_in_env({'http_proxy': self.proxy_url})
731
def test_HTTP_PROXY(self):
732
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
734
def test_all_proxy(self):
735
self.proxied_in_env({'all_proxy': self.proxy_url})
737
def test_ALL_PROXY(self):
738
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
740
def test_http_proxy_with_no_proxy(self):
741
self.not_proxied_in_env({'http_proxy': self.proxy_url,
742
'no_proxy': self.no_proxy_host})
744
def test_HTTP_PROXY_with_NO_PROXY(self):
745
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
746
'NO_PROXY': self.no_proxy_host})
748
def test_all_proxy_with_no_proxy(self):
749
self.not_proxied_in_env({'all_proxy': self.proxy_url,
750
'no_proxy': self.no_proxy_host})
752
def test_ALL_PROXY_with_NO_PROXY(self):
753
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
754
'NO_PROXY': self.no_proxy_host})
756
def test_http_proxy_without_scheme(self):
757
self.assertRaises(errors.InvalidURL,
759
{'http_proxy': self.proxy_address})
762
class TestProxyHttpServer_urllib(TestProxyHttpServer,
763
TestCaseWithTwoWebservers):
764
"""Tests proxy server for urllib implementation"""
766
_transport = HttpTransport_urllib
769
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
771
TestCaseWithTwoWebservers):
772
"""Tests proxy server for pycurl implementation"""
775
TestProxyHttpServer.setUp(self)
776
# Oh my ! pycurl does not check for the port as part of
777
# no_proxy :-( So we just test the host part
778
self.no_proxy_host = 'localhost'
780
def test_HTTP_PROXY(self):
781
# pycurl does not check HTTP_PROXY for security reasons
782
# (for use in a CGI context that we do not care
783
# about. Should we ?)
784
raise tests.TestNotApplicable(
785
'pycurl does not check HTTP_PROXY for security reasons')
787
def test_HTTP_PROXY_with_NO_PROXY(self):
788
raise tests.TestNotApplicable(
789
'pycurl does not check HTTP_PROXY for security reasons')
791
def test_http_proxy_without_scheme(self):
792
# pycurl *ignores* invalid proxy env variables. If that
793
# ever change in the future, this test will fail
794
# indicating that pycurl do not ignore anymore such
796
self.not_proxied_in_env({'http_proxy': self.proxy_address})
799
class TestRanges(object):
800
"""Test the Range header in GET methods..
802
This MUST be used by daughter classes that also inherit from
803
TestCaseWithWebserver.
805
We can't inherit directly from TestCaseWithWebserver or the
806
test framework will try to create an instance which cannot
807
run, its implementation being incomplete.
811
TestCaseWithWebserver.setUp(self)
812
self.build_tree_contents([('a', '0123456789')],)
813
server = self.get_readonly_server()
814
self.transport = self._transport(server.get_url())
816
def _file_contents(self, relpath, ranges):
817
offsets = [ (start, end - start + 1) for start, end in ranges]
818
coalesce = self.transport._coalesce_offsets
819
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
820
code, data = self.transport._get(relpath, coalesced)
821
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
822
for start, end in ranges:
824
yield data.read(end - start + 1)
826
def _file_tail(self, relpath, tail_amount):
827
code, data = self.transport._get(relpath, [], tail_amount)
828
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
829
data.seek(-tail_amount, 2)
830
return data.read(tail_amount)
832
def test_range_header(self):
834
map(self.assertEqual,['0', '234'],
835
list(self._file_contents('a', [(0,0), (2,4)])),)
837
self.assertEqual('789', self._file_tail('a', 3))
838
# Syntactically invalid range
839
self.assertListRaises(errors.InvalidHttpRange,
840
self._file_contents, 'a', [(4, 3)])
841
# Semantically invalid range
842
self.assertListRaises(errors.InvalidHttpRange,
843
self._file_contents, 'a', [(42, 128)])
846
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
847
"""Test the Range header in GET methods for urllib implementation"""
849
_transport = HttpTransport_urllib
852
class TestRanges_pycurl(TestWithTransport_pycurl,
854
TestCaseWithWebserver):
855
"""Test the Range header in GET methods for pycurl implementation"""
858
class TestHTTPRedirections(object):
859
"""Test redirection between http servers.
861
This MUST be used by daughter classes that also inherit from
862
TestCaseWithRedirectedWebserver.
864
We can't inherit directly from TestCaseWithTwoWebservers or the
865
test framework will try to create an instance which cannot
866
run, its implementation being incomplete.
869
def create_transport_secondary_server(self):
870
"""Create the secondary server redirecting to the primary server"""
871
new = self.get_readonly_server()
873
redirecting = HTTPServerRedirecting()
874
redirecting.redirect_to(new.host, new.port)
878
super(TestHTTPRedirections, self).setUp()
879
self.build_tree_contents([('a', '0123456789'),
881
'# Bazaar revision bundle v0.9\n#\n')
884
self.old_transport = self._transport(self.old_server.get_url())
886
def test_redirected(self):
887
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
888
t = self._transport(self.new_server.get_url())
889
self.assertEqual('0123456789', t.get('a').read())
891
def test_read_redirected_bundle_from_url(self):
892
from bzrlib.bundle import read_bundle_from_url
893
url = self.old_transport.abspath('bundle')
894
bundle = read_bundle_from_url(url)
895
# If read_bundle_from_url was successful we get an empty bundle
896
self.assertEqual([], bundle.revisions)
899
class TestHTTPRedirections_urllib(TestHTTPRedirections,
900
TestCaseWithRedirectedWebserver):
901
"""Tests redirections for urllib implementation"""
903
_transport = HttpTransport_urllib
907
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
908
TestHTTPRedirections,
909
TestCaseWithRedirectedWebserver):
910
"""Tests redirections for pycurl implementation"""
913
class RedirectedRequest(Request):
914
"""Request following redirections"""
916
init_orig = Request.__init__
918
def __init__(self, method, url, *args, **kwargs):
919
RedirectedRequest.init_orig(self, method, url, args, kwargs)
920
self.follow_redirections = True
923
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
924
"""Test redirections provided by urllib.
926
http implementations do not redirect silently anymore (they
927
do not redirect at all in fact). The mechanism is still in
928
place at the _urllib2_wrappers.Request level and these tests
931
For the pycurl implementation
932
the redirection have been deleted as we may deprecate pycurl
933
and I have no place to keep a working implementation.
937
_transport = HttpTransport_urllib
940
super(TestHTTPSilentRedirections_urllib, self).setUp()
941
self.setup_redirected_request()
942
self.addCleanup(self.cleanup_redirected_request)
943
self.build_tree_contents([('a','a'),
945
('1/a', 'redirected once'),
947
('2/a', 'redirected twice'),
949
('3/a', 'redirected thrice'),
951
('4/a', 'redirected 4 times'),
953
('5/a', 'redirected 5 times'),
956
self.old_transport = self._transport(self.old_server.get_url())
958
def setup_redirected_request(self):
959
self.original_class = _urllib2_wrappers.Request
960
_urllib2_wrappers.Request = RedirectedRequest
962
def cleanup_redirected_request(self):
963
_urllib2_wrappers.Request = self.original_class
965
def create_transport_secondary_server(self):
966
"""Create the secondary server, redirections are defined in the tests"""
967
return HTTPServerRedirecting()
969
def test_one_redirection(self):
970
t = self.old_transport
972
req = RedirectedRequest('GET', t.abspath('a'))
973
req.follow_redirections = True
974
new_prefix = 'http://%s:%s' % (self.new_server.host,
975
self.new_server.port)
976
self.old_server.redirections = \
977
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
978
self.assertEquals('redirected once',t._perform(req).read())
980
def test_five_redirections(self):
981
t = self.old_transport
983
req = RedirectedRequest('GET', t.abspath('a'))
984
req.follow_redirections = True
985
old_prefix = 'http://%s:%s' % (self.old_server.host,
986
self.old_server.port)
987
new_prefix = 'http://%s:%s' % (self.new_server.host,
988
self.new_server.port)
989
self.old_server.redirections = \
990
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
991
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
992
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
993
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
994
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
996
self.assertEquals('redirected 5 times',t._perform(req).read())
999
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1000
"""Test transport.do_catching_redirections.
1002
We arbitrarily choose to use urllib transports
1005
_transport = HttpTransport_urllib
1008
super(TestDoCatchRedirections, self).setUp()
1009
self.build_tree_contents([('a', '0123456789'),],)
1011
self.old_transport = self._transport(self.old_server.get_url())
1013
def get_a(self, transport):
1014
return transport.get('a')
1016
def test_no_redirection(self):
1017
t = self._transport(self.new_server.get_url())
1019
# We use None for redirected so that we fail if redirected
1020
self.assertEquals('0123456789',
1021
do_catching_redirections(self.get_a, t, None).read())
1023
def test_one_redirection(self):
1024
self.redirections = 0
1026
def redirected(transport, exception, redirection_notice):
1027
self.redirections += 1
1028
dir, file = urlutils.split(exception.target)
1029
return self._transport(dir)
1031
self.assertEquals('0123456789',
1032
do_catching_redirections(self.get_a,
1036
self.assertEquals(1, self.redirections)
1038
def test_redirection_loop(self):
1040
def redirected(transport, exception, redirection_notice):
1041
# By using the redirected url as a base dir for the
1042
# *old* transport, we create a loop: a => a/a =>
1044
return self.old_transport.clone(exception.target)
1046
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1047
self.get_a, self.old_transport, redirected)
1050
class TestAuth(object):
1051
"""Test some authentication scheme specified by daughter class.
1053
This MUST be used by daughter classes that also inherit from
1054
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1057
_password_prompt_prefix = ''
1060
"""Set up the test environment
1062
Daughter classes should set up their own environment
1063
(including self.server) and explicitely call this
1064
method. This is needed because we want to reuse the same
1065
tests for proxy and no-proxy accesses which have
1066
different ways of setting self.server.
1068
self.build_tree_contents([('a', 'contents of a\n'),
1069
('b', 'contents of b\n'),])
1071
def get_user_url(self, user=None, password=None):
1072
"""Build an url embedding user and password"""
1073
url = '%s://' % self.server._url_protocol
1074
if user is not None:
1076
if password is not None:
1077
url += ':' + password
1079
url += '%s:%s/' % (self.server.host, self.server.port)
1082
def test_no_user(self):
1083
self.server.add_user('joe', 'foo')
1084
t = self.get_user_transport()
1085
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1086
# Only one 'Authentication Required' error should occur
1087
self.assertEqual(1, self.server.auth_required_errors)
1089
def test_empty_pass(self):
1090
self.server.add_user('joe', '')
1091
t = self.get_user_transport('joe', '')
1092
self.assertEqual('contents of a\n', t.get('a').read())
1093
# Only one 'Authentication Required' error should occur
1094
self.assertEqual(1, self.server.auth_required_errors)
1096
def test_user_pass(self):
1097
self.server.add_user('joe', 'foo')
1098
t = self.get_user_transport('joe', 'foo')
1099
self.assertEqual('contents of a\n', t.get('a').read())
1100
# Only one 'Authentication Required' error should occur
1101
self.assertEqual(1, self.server.auth_required_errors)
1103
def test_unknown_user(self):
1104
self.server.add_user('joe', 'foo')
1105
t = self.get_user_transport('bill', 'foo')
1106
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1107
# Two 'Authentication Required' errors should occur (the
1108
# initial 'who are you' and 'I don't know you, who are
1110
self.assertEqual(2, self.server.auth_required_errors)
1112
def test_wrong_pass(self):
1113
self.server.add_user('joe', 'foo')
1114
t = self.get_user_transport('joe', 'bar')
1115
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1116
# Two 'Authentication Required' errors should occur (the
1117
# initial 'who are you' and 'this is not you, who are you')
1118
self.assertEqual(2, self.server.auth_required_errors)
1120
def test_prompt_for_password(self):
1121
self.server.add_user('joe', 'foo')
1122
t = self.get_user_transport('joe', None)
1123
stdout = tests.StringIOWrapper()
1124
ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1125
self.assertEqual('contents of a\n',t.get('a').read())
1126
# stdin should be empty
1127
self.assertEqual('', ui.ui_factory.stdin.readline())
1128
self._check_password_prompt(t._unqualified_scheme, 'joe',
1130
# And we shouldn't prompt again for a different request
1131
# against the same transport.
1132
self.assertEqual('contents of b\n',t.get('b').read())
1134
# And neither against a clone
1135
self.assertEqual('contents of b\n',t2.get('b').read())
1136
# Only one 'Authentication Required' error should occur
1137
self.assertEqual(1, self.server.auth_required_errors)
1139
def _check_password_prompt(self, scheme, user, actual_prompt):
1140
expected_prompt = (self._password_prompt_prefix
1141
+ ("%s %s@%s:%d, Realm: '%s' password: "
1143
user, self.server.host, self.server.port,
1144
self.server.auth_realm)))
1145
self.assertEquals(expected_prompt, actual_prompt)
1147
def test_no_prompt_for_password_when_using_auth_config(self):
1150
stdin_content = 'bar\n' # Not the right password
1151
self.server.add_user(user, password)
1152
t = self.get_user_transport(user, None)
1153
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1154
stdout=tests.StringIOWrapper())
1155
# Create a minimal config file with the right password
1156
conf = config.AuthenticationConfig()
1157
conf._get_config().update(
1158
{'httptest': {'scheme': 'http', 'port': self.server.port,
1159
'user': user, 'password': password}})
1161
# Issue a request to the server to connect
1162
self.assertEqual('contents of a\n',t.get('a').read())
1163
# stdin should have been left untouched
1164
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1165
# Only one 'Authentication Required' error should occur
1166
self.assertEqual(1, self.server.auth_required_errors)
1170
class TestHTTPAuth(TestAuth):
1171
"""Test HTTP authentication schemes.
1173
Daughter classes MUST inherit from TestCaseWithWebserver too.
1176
_auth_header = 'Authorization'
1179
TestCaseWithWebserver.setUp(self)
1180
self.server = self.get_readonly_server()
1181
TestAuth.setUp(self)
1183
def get_user_transport(self, user=None, password=None):
1184
return self._transport(self.get_user_url(user, password))
1187
class TestProxyAuth(TestAuth):
1188
"""Test proxy authentication schemes.
1190
Daughter classes MUST also inherit from TestCaseWithWebserver.
1192
_auth_header = 'Proxy-authorization'
1193
_password_prompt_prefix = 'Proxy '
1197
TestCaseWithWebserver.setUp(self)
1198
self.server = self.get_readonly_server()
1200
self.addCleanup(self._restore_env)
1201
TestAuth.setUp(self)
1202
# Override the contents to avoid false positives
1203
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1204
('b', 'not proxied contents of b\n'),
1205
('a-proxied', 'contents of a\n'),
1206
('b-proxied', 'contents of b\n'),
1209
def get_user_transport(self, user=None, password=None):
1210
self._install_env({'all_proxy': self.get_user_url(user, password)})
1211
return self._transport(self.server.get_url())
1213
def _install_env(self, env):
1214
for name, value in env.iteritems():
1215
self._old_env[name] = osutils.set_or_unset_env(name, value)
1217
def _restore_env(self):
1218
for name, value in self._old_env.iteritems():
1219
osutils.set_or_unset_env(name, value)
1222
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1223
"""Test http basic authentication scheme"""
1225
_transport = HttpTransport_urllib
1227
def create_transport_readonly_server(self):
1228
return HTTPBasicAuthServer()
1231
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1232
"""Test proxy basic authentication scheme"""
1234
_transport = HttpTransport_urllib
1236
def create_transport_readonly_server(self):
1237
return ProxyBasicAuthServer()
1240
class TestDigestAuth(object):
1241
"""Digest Authentication specific tests"""
1243
def test_changing_nonce(self):
1244
self.server.add_user('joe', 'foo')
1245
t = self.get_user_transport('joe', 'foo')
1246
self.assertEqual('contents of a\n', t.get('a').read())
1247
self.assertEqual('contents of b\n', t.get('b').read())
1248
# Only one 'Authentication Required' error should have
1250
self.assertEqual(1, self.server.auth_required_errors)
1251
# The server invalidates the current nonce
1252
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1253
self.assertEqual('contents of a\n', t.get('a').read())
1254
# Two 'Authentication Required' errors should occur (the
1255
# initial 'who are you' and a second 'who are you' with the new nonce)
1256
self.assertEqual(2, self.server.auth_required_errors)
1259
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1260
"""Test http digest authentication scheme"""
1262
_transport = HttpTransport_urllib
1264
def create_transport_readonly_server(self):
1265
return HTTPDigestAuthServer()
1268
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1269
TestCaseWithWebserver):
1270
"""Test proxy digest authentication scheme"""
1272
_transport = HttpTransport_urllib
1274
def create_transport_readonly_server(self):
1275
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
