1
# Copyright (C) 2005, 2006 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
# TODO: What about renaming to bzrlib.tests.transport.http ?
23
from cStringIO import StringIO
37
from bzrlib.tests import (
43
from bzrlib.tests.HttpServer import (
48
from bzrlib.tests.HTTPTestUtil import (
49
BadProtocolRequestHandler,
50
BadStatusRequestHandler,
51
ForbiddenRequestHandler,
54
HTTPServerRedirecting,
55
InvalidStatusRequestHandler,
56
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 (
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, errors.ConnectionError),
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(PasswordManager())
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.
835
# FIXME: Once the test suite is better fitted to test
836
# authorization schemes, test proxy authorizations too (see
840
TestCaseWithTwoWebservers.setUp(self)
841
self.build_tree_contents([('foo', 'contents of foo\n'),
842
('foo-proxied', 'proxied contents of foo\n')])
843
# Let's setup some attributes for tests
844
self.server = self.get_readonly_server()
845
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
846
self.no_proxy_host = self.proxy_address
847
# The secondary server is the proxy
848
self.proxy = self.get_secondary_server()
849
self.proxy_url = self.proxy.get_url()
852
def create_transport_secondary_server(self):
853
"""Creates an http server that will serve files with
854
'-proxied' appended to their names.
858
def _install_env(self, env):
859
for name, value in env.iteritems():
860
self._old_env[name] = osutils.set_or_unset_env(name, value)
862
def _restore_env(self):
863
for name, value in self._old_env.iteritems():
864
osutils.set_or_unset_env(name, value)
866
def proxied_in_env(self, env):
867
self._install_env(env)
868
url = self.server.get_url()
869
t = self._transport(url)
871
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
875
def not_proxied_in_env(self, env):
876
self._install_env(env)
877
url = self.server.get_url()
878
t = self._transport(url)
880
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
884
def test_http_proxy(self):
885
self.proxied_in_env({'http_proxy': self.proxy_url})
887
def test_HTTP_PROXY(self):
888
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
890
def test_all_proxy(self):
891
self.proxied_in_env({'all_proxy': self.proxy_url})
893
def test_ALL_PROXY(self):
894
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
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_HTTP_PROXY_with_NO_PROXY(self):
901
self.not_proxied_in_env({'HTTP_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_ALL_PROXY_with_NO_PROXY(self):
909
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
910
'NO_PROXY': self.no_proxy_host})
912
def test_http_proxy_without_scheme(self):
913
self.assertRaises(errors.InvalidURL,
915
{'http_proxy': self.proxy_address})
918
class TestProxyHttpServer_urllib(TestProxyHttpServer,
919
TestCaseWithTwoWebservers):
920
"""Tests proxy server for urllib implementation"""
922
_transport = HttpTransport_urllib
925
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
927
TestCaseWithTwoWebservers):
928
"""Tests proxy server for pycurl implementation"""
931
TestProxyHttpServer.setUp(self)
932
# Oh my ! pycurl does not check for the port as part of
933
# no_proxy :-( So we just test the host part
934
self.no_proxy_host = 'localhost'
936
def test_HTTP_PROXY(self):
937
# pycurl does not check HTTP_PROXY for security reasons
938
# (for use in a CGI context that we do not care
939
# about. Should we ?)
940
raise TestSkipped('pycurl does not check HTTP_PROXY '
941
'for security reasons')
943
def test_HTTP_PROXY_with_NO_PROXY(self):
944
raise TestSkipped('pycurl does not check HTTP_PROXY '
945
'for security reasons')
947
def test_http_proxy_without_scheme(self):
948
# pycurl *ignores* invalid proxy env variables. If that
949
# ever change in the future, this test will fail
950
# indicating that pycurl do not ignore anymore such
952
self.not_proxied_in_env({'http_proxy': self.proxy_address})
955
class TestRanges(object):
956
"""Test the Range header in GET methods..
958
This MUST be used by daughter classes that also inherit from
959
TestCaseWithWebserver.
961
We can't inherit directly from TestCaseWithWebserver or the
962
test framework will try to create an instance which cannot
963
run, its implementation being incomplete.
967
TestCaseWithWebserver.setUp(self)
968
self.build_tree_contents([('a', '0123456789')],)
969
server = self.get_readonly_server()
970
self.transport = self._transport(server.get_url())
972
def _file_contents(self, relpath, ranges):
973
offsets = [ (start, end - start + 1) for start, end in ranges]
974
coalesce = self.transport._coalesce_offsets
975
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
976
code, data = self.transport._get(relpath, coalesced)
977
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
978
for start, end in ranges:
980
yield data.read(end - start + 1)
982
def _file_tail(self, relpath, tail_amount):
983
code, data = self.transport._get(relpath, [], tail_amount)
984
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
985
data.seek(-tail_amount + 1, 2)
986
return data.read(tail_amount)
988
def test_range_header(self):
990
map(self.assertEqual,['0', '234'],
991
list(self._file_contents('a', [(0,0), (2,4)])),)
993
self.assertEqual('789', self._file_tail('a', 3))
994
# Syntactically invalid range
995
self.assertListRaises(errors.InvalidRange,
996
self._file_contents, 'a', [(4, 3)])
997
# Semantically invalid range
998
self.assertListRaises(errors.InvalidRange,
999
self._file_contents, 'a', [(42, 128)])
1002
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1003
"""Test the Range header in GET methods for urllib implementation"""
1005
_transport = HttpTransport_urllib
1008
class TestRanges_pycurl(TestWithTransport_pycurl,
1010
TestCaseWithWebserver):
1011
"""Test the Range header in GET methods for pycurl implementation"""
1014
class TestHTTPRedirections(object):
1015
"""Test redirection between http servers.
1017
This MUST be used by daughter classes that also inherit from
1018
TestCaseWithRedirectedWebserver.
1020
We can't inherit directly from TestCaseWithTwoWebservers or the
1021
test framework will try to create an instance which cannot
1022
run, its implementation being incomplete.
1025
def create_transport_secondary_server(self):
1026
"""Create the secondary server redirecting to the primary server"""
1027
new = self.get_readonly_server()
1029
redirecting = HTTPServerRedirecting()
1030
redirecting.redirect_to(new.host, new.port)
1034
super(TestHTTPRedirections, self).setUp()
1035
self.build_tree_contents([('a', '0123456789'),
1037
'# Bazaar revision bundle v0.9\n#\n')
1040
self.old_transport = self._transport(self.old_server.get_url())
1042
def test_redirected(self):
1043
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1044
t = self._transport(self.new_server.get_url())
1045
self.assertEqual('0123456789', t.get('a').read())
1047
def test_read_redirected_bundle_from_url(self):
1048
from bzrlib.bundle import read_bundle_from_url
1049
url = self.old_transport.abspath('bundle')
1050
bundle = read_bundle_from_url(url)
1051
# If read_bundle_from_url was successful we get an empty bundle
1052
self.assertEqual([], bundle.revisions)
1055
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1056
TestCaseWithRedirectedWebserver):
1057
"""Tests redirections for urllib implementation"""
1059
_transport = HttpTransport_urllib
1063
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1064
TestHTTPRedirections,
1065
TestCaseWithRedirectedWebserver):
1066
"""Tests redirections for pycurl implementation"""
1069
class RedirectedRequest(Request):
1070
"""Request following redirections"""
1072
init_orig = Request.__init__
1074
def __init__(self, method, url, *args, **kwargs):
1075
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1076
self.follow_redirections = True
1079
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1080
"""Test redirections provided by urllib.
1082
http implementations do not redirect silently anymore (they
1083
do not redirect at all in fact). The mechanism is still in
1084
place at the _urllib2_wrappers.Request level and these tests
1087
For the pycurl implementation
1088
the redirection have been deleted as we may deprecate pycurl
1089
and I have no place to keep a working implementation.
1093
_transport = HttpTransport_urllib
1096
super(TestHTTPSilentRedirections_urllib, self).setUp()
1097
self.setup_redirected_request()
1098
self.addCleanup(self.cleanup_redirected_request)
1099
self.build_tree_contents([('a','a'),
1101
('1/a', 'redirected once'),
1103
('2/a', 'redirected twice'),
1105
('3/a', 'redirected thrice'),
1107
('4/a', 'redirected 4 times'),
1109
('5/a', 'redirected 5 times'),
1112
self.old_transport = self._transport(self.old_server.get_url())
1114
def setup_redirected_request(self):
1115
self.original_class = _urllib2_wrappers.Request
1116
_urllib2_wrappers.Request = RedirectedRequest
1118
def cleanup_redirected_request(self):
1119
_urllib2_wrappers.Request = self.original_class
1121
def create_transport_secondary_server(self):
1122
"""Create the secondary server, redirections are defined in the tests"""
1123
return HTTPServerRedirecting()
1125
def test_one_redirection(self):
1126
t = self.old_transport
1128
req = RedirectedRequest('GET', t.abspath('a'))
1129
req.follow_redirections = True
1130
new_prefix = 'http://%s:%s' % (self.new_server.host,
1131
self.new_server.port)
1132
self.old_server.redirections = \
1133
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1134
self.assertEquals('redirected once',t._perform(req).read())
1136
def test_five_redirections(self):
1137
t = self.old_transport
1139
req = RedirectedRequest('GET', t.abspath('a'))
1140
req.follow_redirections = True
1141
old_prefix = 'http://%s:%s' % (self.old_server.host,
1142
self.old_server.port)
1143
new_prefix = 'http://%s:%s' % (self.new_server.host,
1144
self.new_server.port)
1145
self.old_server.redirections = \
1146
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1147
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1148
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1149
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1150
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1152
self.assertEquals('redirected 5 times',t._perform(req).read())
1155
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1156
"""Test transport.do_catching_redirections.
1158
We arbitrarily choose to use urllib transports
1161
_transport = HttpTransport_urllib
1164
super(TestDoCatchRedirections, self).setUp()
1165
self.build_tree_contents([('a', '0123456789'),],)
1167
self.old_transport = self._transport(self.old_server.get_url())
1169
def get_a(self, transport):
1170
return transport.get('a')
1172
def test_no_redirection(self):
1173
t = self._transport(self.new_server.get_url())
1175
# We use None for redirected so that we fail if redirected
1176
self.assertEquals('0123456789',
1177
do_catching_redirections(self.get_a, t, None).read())
1179
def test_one_redirection(self):
1180
self.redirections = 0
1182
def redirected(transport, exception, redirection_notice):
1183
self.redirections += 1
1184
dir, file = urlutils.split(exception.target)
1185
return self._transport(dir)
1187
self.assertEquals('0123456789',
1188
do_catching_redirections(self.get_a,
1192
self.assertEquals(1, self.redirections)
1194
def test_redirection_loop(self):
1196
def redirected(transport, exception, redirection_notice):
1197
# By using the redirected url as a base dir for the
1198
# *old* transport, we create a loop: a => a/a =>
1200
return self.old_transport.clone(exception.target)
1202
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1203
self.get_a, self.old_transport, redirected)
1206
class TestAuth(object):
1207
"""Test some authentication scheme specified by daughter class.
1209
This MUST be used by daughter classes that also inherit from
1210
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1214
"""Set up the test environment
1216
Daughter classes should set up their own environment
1217
(including self.server) and explicitely call this
1218
method. This is needed because we want to reuse the same
1219
tests for proxy and no-proxy accesses which have
1220
different ways of setting self.server.
1222
self.build_tree_contents([('a', 'contents of a\n'),
1223
('b', 'contents of b\n'),])
1224
self.old_factory = ui.ui_factory
1225
# The following has the unfortunate side-effect of hiding any ouput
1226
# during the tests (including pdb prompts). Feel free to comment them
1227
# for debugging purposes but leave them in place, there are needed to
1228
# run the tests without any console
1229
self.old_stdout = sys.stdout
1230
sys.stdout = StringIOWrapper()
1231
self.addCleanup(self.restoreUIFactory)
1233
def restoreUIFactory(self):
1234
ui.ui_factory = self.old_factory
1235
sys.stdout = self.old_stdout
1237
def get_user_url(self, user=None, password=None):
1238
"""Build an url embedding user and password"""
1239
url = '%s://' % self.server._url_protocol
1240
if user is not None:
1242
if password is not None:
1243
url += ':' + password
1245
url += '%s:%s/' % (self.server.host, self.server.port)
1248
def test_no_user(self):
1249
self.server.add_user('joe', 'foo')
1250
t = self.get_user_transport()
1251
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1252
# Only one 'Authentication Required' error should occur
1253
self.assertEqual(1, self.server.auth_required_errors)
1255
def test_empty_pass(self):
1256
self.server.add_user('joe', '')
1257
t = self.get_user_transport('joe', '')
1258
self.assertEqual('contents of a\n', t.get('a').read())
1259
# Only one 'Authentication Required' error should occur
1260
self.assertEqual(1, self.server.auth_required_errors)
1262
def test_user_pass(self):
1263
self.server.add_user('joe', 'foo')
1264
t = self.get_user_transport('joe', 'foo')
1265
self.assertEqual('contents of a\n', t.get('a').read())
1266
# Only one 'Authentication Required' error should occur
1267
self.assertEqual(1, self.server.auth_required_errors)
1269
def test_unknown_user(self):
1270
self.server.add_user('joe', 'foo')
1271
t = self.get_user_transport('bill', 'foo')
1272
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1273
# Two 'Authentication Required' errors should occur (the
1274
# initial 'who are you' and 'I don't know you, who are
1276
self.assertEqual(2, self.server.auth_required_errors)
1278
def test_wrong_pass(self):
1279
self.server.add_user('joe', 'foo')
1280
t = self.get_user_transport('joe', 'bar')
1281
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1282
# Two 'Authentication Required' errors should occur (the
1283
# initial 'who are you' and 'this is not you, who are you')
1284
self.assertEqual(2, self.server.auth_required_errors)
1286
def test_prompt_for_password(self):
1287
self.server.add_user('joe', 'foo')
1288
t = self.get_user_transport('joe', None)
1289
ui.ui_factory = TestUIFactory(stdin='foo\n')
1290
self.assertEqual('contents of a\n',t.get('a').read())
1291
# stdin should be empty
1292
self.assertEqual('', ui.ui_factory.stdin.readline())
1293
# And we shouldn't prompt again for a different request
1294
# against the same transport.
1295
self.assertEqual('contents of b\n',t.get('b').read())
1297
# And neither against a clone
1298
self.assertEqual('contents of b\n',t2.get('b').read())
1299
# Only one 'Authentication Required' error should occur
1300
self.assertEqual(1, self.server.auth_required_errors)
1303
class TestHTTPAuth(TestAuth):
1304
"""Test HTTP authentication schemes.
1306
Daughter classes MUST inherit from TestCaseWithWebserver too.
1309
_auth_header = 'Authorization'
1312
TestCaseWithWebserver.setUp(self)
1313
self.server = self.get_readonly_server()
1314
TestAuth.setUp(self)
1316
def get_user_transport(self, user=None, password=None):
1317
return self._transport(self.get_user_url(user, password))
1320
class TestProxyAuth(TestAuth):
1321
"""Test proxy authentication schemes.
1323
Daughter classes MUST also inherit from TestCaseWithWebserver.
1325
_auth_header = 'Proxy-authorization'
1328
TestCaseWithWebserver.setUp(self)
1329
self.server = self.get_readonly_server()
1331
self.addCleanup(self._restore_env)
1332
TestAuth.setUp(self)
1333
# Override the contents to avoid false positives
1334
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1335
('b', 'not proxied contents of b\n'),
1336
('a-proxied', 'contents of a\n'),
1337
('b-proxied', 'contents of b\n'),
1340
def get_user_transport(self, user=None, password=None):
1341
self._install_env({'all_proxy': self.get_user_url(user, password)})
1342
return self._transport(self.server.get_url())
1344
def _install_env(self, env):
1345
for name, value in env.iteritems():
1346
self._old_env[name] = osutils.set_or_unset_env(name, value)
1348
def _restore_env(self):
1349
for name, value in self._old_env.iteritems():
1350
osutils.set_or_unset_env(name, value)
1353
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1354
"""Test http basic authentication scheme"""
1356
_transport = HttpTransport_urllib
1358
def create_transport_readonly_server(self):
1359
return HTTPBasicAuthServer()
1362
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1363
"""Test proxy basic authentication scheme"""
1365
_transport = HttpTransport_urllib
1367
def create_transport_readonly_server(self):
1368
return ProxyBasicAuthServer()
1371
class TestDigestAuth(object):
1372
"""Digest Authentication specific tests"""
1374
def test_changing_nonce(self):
1375
self.server.add_user('joe', 'foo')
1376
t = self.get_user_transport('joe', 'foo')
1377
self.assertEqual('contents of a\n', t.get('a').read())
1378
self.assertEqual('contents of b\n', t.get('b').read())
1379
# Only one 'Authentication Required' error should have
1381
self.assertEqual(1, self.server.auth_required_errors)
1382
# The server invalidates the current nonce
1383
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1384
self.assertEqual('contents of a\n', t.get('a').read())
1385
# Two 'Authentication Required' errors should occur (the
1386
# initial 'who are you' and a second 'who are you' with the new nonce)
1387
self.assertEqual(2, self.server.auth_required_errors)
1390
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1391
"""Test http digest authentication scheme"""
1393
_transport = HttpTransport_urllib
1395
def create_transport_readonly_server(self):
1396
return HTTPDigestAuthServer()
1399
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1400
TestCaseWithWebserver):
1401
"""Test proxy digest authentication scheme"""
1403
_transport = HttpTransport_urllib
1405
def create_transport_readonly_server(self):
1406
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
