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
NoRangeRequestHandler,
58
ProxyDigestAuthServer,
60
SingleRangeRequestHandler,
61
TestCaseWithRedirectedWebserver,
62
TestCaseWithTwoWebservers,
63
TestCaseWithWebserver,
66
from bzrlib.transport import (
67
do_catching_redirections,
71
from bzrlib.transport.http import (
76
from bzrlib.transport.http._urllib import HttpTransport_urllib
77
from bzrlib.transport.http._urllib2_wrappers import (
84
class FakeManager(object):
89
def add_password(self, realm, host, username, password):
90
self.credentials.append([realm, host, username, password])
93
class RecordingServer(object):
94
"""A fake HTTP server.
96
It records the bytes sent to it, and replies with a 200.
99
def __init__(self, expect_body_tail=None):
102
:type expect_body_tail: str
103
:param expect_body_tail: a reply won't be sent until this string is
106
self._expect_body_tail = expect_body_tail
109
self.received_bytes = ''
112
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
113
self._sock.bind(('127.0.0.1', 0))
114
self.host, self.port = self._sock.getsockname()
115
self._ready = threading.Event()
116
self._thread = threading.Thread(target=self._accept_read_and_reply)
117
self._thread.setDaemon(True)
121
def _accept_read_and_reply(self):
124
self._sock.settimeout(5)
126
conn, address = self._sock.accept()
127
# On win32, the accepted connection will be non-blocking to start
128
# with because we're using settimeout.
129
conn.setblocking(True)
130
while not self.received_bytes.endswith(self._expect_body_tail):
131
self.received_bytes += conn.recv(4096)
132
conn.sendall('HTTP/1.1 200 OK\r\n')
133
except socket.timeout:
134
# Make sure the client isn't stuck waiting for us to e.g. accept.
137
# The client may have already closed the socket.
144
# We might have already closed it. We don't care.
150
class TestWithTransport_pycurl(object):
151
"""Test case to inherit from if pycurl is present"""
153
def _get_pycurl_maybe(self):
155
from bzrlib.transport.http._pycurl import PyCurlTransport
156
return PyCurlTransport
157
except errors.DependencyNotPresent:
158
raise TestSkipped('pycurl not present')
160
_transport = property(_get_pycurl_maybe)
3
from bzrlib.selftest import TestCase
4
from bzrlib.transport.http import HttpTransport
163
6
class TestHttpUrls(TestCase):
165
# TODO: This should be moved to authorization tests once they
168
def test_url_parsing(self):
170
url = extract_auth('http://example.com', f)
171
self.assertEquals('http://example.com', url)
172
self.assertEquals(0, len(f.credentials))
173
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
174
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
175
self.assertEquals(1, len(f.credentials))
176
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
180
class TestHttpTransportUrls(object):
181
"""Test the http urls.
183
This MUST be used by daughter classes that also inherit from
186
We can't inherit directly from TestCase or the
187
test framework will try to create an instance which cannot
188
run, its implementation being incomplete.
191
7
def test_abs_url(self):
192
8
"""Construction of absolute http URLs"""
193
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
9
t = HttpTransport('http://bazaar-ng.org/bzr/bzr.dev/')
194
10
eq = self.assertEqualDiff
195
eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
196
eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
197
eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
12
'http://bazaar-ng.org/bzr/bzr.dev')
13
eq(t.abspath('foo/bar'),
14
'http://bazaar-ng.org/bzr/bzr.dev/foo/bar')
16
'http://bazaar-ng.org/bzr/bzr.dev/.bzr')
198
17
eq(t.abspath('.bzr/1//2/./3'),
199
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
18
'http://bazaar-ng.org/bzr/bzr.dev/.bzr/1/2/3')
201
20
def test_invalid_http_urls(self):
202
21
"""Trap invalid construction of urls"""
203
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
204
self.assertRaises(ValueError, t.abspath, '.bzr/')
205
t = self._transport('http://http://bazaar-vcs.org/bzr/bzr.dev/')
206
self.assertRaises((errors.InvalidURL, errors.ConnectionError),
22
t = HttpTransport('http://bazaar-ng.org/bzr/bzr.dev/')
23
self.assertRaises(ValueError,
26
self.assertRaises(ValueError,
209
30
def test_http_root_urls(self):
210
31
"""Construction of URLs from server root"""
211
t = self._transport('http://bzr.ozlabs.org/')
32
t = HttpTransport('http://bzr.ozlabs.org/')
212
33
eq = self.assertEqualDiff
213
34
eq(t.abspath('.bzr/tree-version'),
214
35
'http://bzr.ozlabs.org/.bzr/tree-version')
216
def test_http_impl_urls(self):
217
"""There are servers which ask for particular clients to connect"""
218
server = self._server()
221
url = server.get_url()
222
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
227
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
228
"""Test http urls with urllib"""
230
_transport = HttpTransport_urllib
231
_server = HttpServer_urllib
232
_qualified_prefix = 'http+urllib'
235
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
237
"""Test http urls with pycurl"""
239
_server = HttpServer_PyCurl
240
_qualified_prefix = 'http+pycurl'
242
# TODO: This should really be moved into another pycurl
243
# specific test. When https tests will be implemented, take
244
# this one into account.
245
def test_pycurl_without_https_support(self):
246
"""Test that pycurl without SSL do not fail with a traceback.
248
For the purpose of the test, we force pycurl to ignore
249
https by supplying a fake version_info that do not
255
raise TestSkipped('pycurl not present')
256
# Now that we have pycurl imported, we can fake its version_info
257
# This was taken from a windows pycurl without SSL
259
pycurl.version_info = lambda : (2,
267
('ftp', 'gopher', 'telnet',
268
'dict', 'ldap', 'http', 'file'),
272
self.assertRaises(errors.DependencyNotPresent, self._transport,
273
'https://launchpad.net')
275
class TestHttpConnections(object):
276
"""Test the http connections.
278
This MUST be used by daughter classes that also inherit from
279
TestCaseWithWebserver.
281
We can't inherit directly from TestCaseWithWebserver or the
282
test framework will try to create an instance which cannot
283
run, its implementation being incomplete.
287
TestCaseWithWebserver.setUp(self)
288
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
289
transport=self.get_transport())
291
def test_http_has(self):
292
server = self.get_readonly_server()
293
t = self._transport(server.get_url())
294
self.assertEqual(t.has('foo/bar'), True)
295
self.assertEqual(len(server.logs), 1)
296
self.assertContainsRe(server.logs[0],
297
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
299
def test_http_has_not_found(self):
300
server = self.get_readonly_server()
301
t = self._transport(server.get_url())
302
self.assertEqual(t.has('not-found'), False)
303
self.assertContainsRe(server.logs[1],
304
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
306
def test_http_get(self):
307
server = self.get_readonly_server()
308
t = self._transport(server.get_url())
309
fp = t.get('foo/bar')
310
self.assertEqualDiff(
312
'contents of foo/bar\n')
313
self.assertEqual(len(server.logs), 1)
314
self.assertTrue(server.logs[0].find(
315
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
316
% bzrlib.__version__) > -1)
318
def test_get_smart_medium(self):
319
# For HTTP, get_smart_medium should return the transport object.
320
server = self.get_readonly_server()
321
http_transport = self._transport(server.get_url())
322
medium = http_transport.get_smart_medium()
323
self.assertIs(medium, http_transport)
325
def test_has_on_bogus_host(self):
326
# Get a free address and don't 'accept' on it, so that we
327
# can be sure there is no http handler there, but set a
328
# reasonable timeout to not slow down tests too much.
329
default_timeout = socket.getdefaulttimeout()
331
socket.setdefaulttimeout(2)
333
s.bind(('localhost', 0))
334
t = self._transport('http://%s:%s/' % s.getsockname())
335
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
337
socket.setdefaulttimeout(default_timeout)
340
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
341
"""Test http connections with urllib"""
343
_transport = HttpTransport_urllib
347
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
349
TestCaseWithWebserver):
350
"""Test http connections with pycurl"""
353
class TestHttpTransportRegistration(TestCase):
354
"""Test registrations of various http implementations"""
356
def test_http_registered(self):
357
# urlllib should always be present
358
t = get_transport('http+urllib://bzr.google.com/')
359
self.assertIsInstance(t, Transport)
360
self.assertIsInstance(t, HttpTransport_urllib)
363
class TestOffsets(TestCase):
364
"""Test offsets_to_ranges method"""
366
def test_offsets_to_ranges_simple(self):
367
to_range = HttpTransportBase.offsets_to_ranges
368
ranges = to_range([(10, 1)])
369
self.assertEqual([[10, 10]], ranges)
371
ranges = to_range([(0, 1), (1, 1)])
372
self.assertEqual([[0, 1]], ranges)
374
ranges = to_range([(1, 1), (0, 1)])
375
self.assertEqual([[0, 1]], ranges)
377
def test_offset_to_ranges_overlapped(self):
378
to_range = HttpTransportBase.offsets_to_ranges
380
ranges = to_range([(10, 1), (20, 2), (22, 5)])
381
self.assertEqual([[10, 10], [20, 26]], ranges)
383
ranges = to_range([(10, 1), (11, 2), (22, 5)])
384
self.assertEqual([[10, 12], [22, 26]], ranges)
387
class TestPost(object):
389
def _test_post_body_is_received(self, scheme):
390
server = RecordingServer(expect_body_tail='end-of-body')
392
self.addCleanup(server.tearDown)
393
url = '%s://%s:%s/' % (scheme, server.host, server.port)
395
http_transport = get_transport(url)
396
except errors.UnsupportedProtocol:
397
raise TestSkipped('%s not available' % scheme)
398
code, response = http_transport._post('abc def end-of-body')
400
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
401
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
402
# The transport should not be assuming that the server can accept
403
# chunked encoding the first time it connects, because HTTP/1.1, so we
404
# check for the literal string.
406
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
409
class TestPost_urllib(TestCase, TestPost):
410
"""TestPost for urllib implementation"""
412
_transport = HttpTransport_urllib
414
def test_post_body_is_received_urllib(self):
415
self._test_post_body_is_received('http+urllib')
418
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
419
"""TestPost for pycurl implementation"""
421
def test_post_body_is_received_pycurl(self):
422
self._test_post_body_is_received('http+pycurl')
425
class TestRangeHeader(TestCase):
426
"""Test range_header method"""
428
def check_header(self, value, ranges=[], tail=0):
429
range_header = HttpTransportBase.range_header
430
self.assertEqual(value, range_header(ranges, tail))
432
def test_range_header_single(self):
433
self.check_header('0-9', ranges=[[0,9]])
434
self.check_header('100-109', ranges=[[100,109]])
436
def test_range_header_tail(self):
437
self.check_header('-10', tail=10)
438
self.check_header('-50', tail=50)
440
def test_range_header_multi(self):
441
self.check_header('0-9,100-200,300-5000',
442
ranges=[(0,9), (100, 200), (300,5000)])
444
def test_range_header_mixed(self):
445
self.check_header('0-9,300-5000,-50',
446
ranges=[(0,9), (300,5000)],
450
class TestWallServer(object):
451
"""Tests exceptions during the connection phase"""
453
def create_transport_readonly_server(self):
454
return HttpServer(WallRequestHandler)
456
def test_http_has(self):
457
server = self.get_readonly_server()
458
t = self._transport(server.get_url())
459
# Unfortunately httplib (see HTTPResponse._read_status
460
# for details) make no distinction between a closed
461
# socket and badly formatted status line, so we can't
462
# just test for ConnectionError, we have to test
463
# InvalidHttpResponse too.
464
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
467
def test_http_get(self):
468
server = self.get_readonly_server()
469
t = self._transport(server.get_url())
470
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
474
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
475
"""Tests "wall" server for urllib implementation"""
477
_transport = HttpTransport_urllib
480
class TestWallServer_pycurl(TestWithTransport_pycurl,
482
TestCaseWithWebserver):
483
"""Tests "wall" server for pycurl implementation"""
486
class TestBadStatusServer(object):
487
"""Tests bad status from server."""
489
def create_transport_readonly_server(self):
490
return HttpServer(BadStatusRequestHandler)
492
def test_http_has(self):
493
server = self.get_readonly_server()
494
t = self._transport(server.get_url())
495
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
497
def test_http_get(self):
498
server = self.get_readonly_server()
499
t = self._transport(server.get_url())
500
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
503
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
504
"""Tests bad status server for urllib implementation"""
506
_transport = HttpTransport_urllib
509
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
511
TestCaseWithWebserver):
512
"""Tests bad status server for pycurl implementation"""
515
class TestInvalidStatusServer(TestBadStatusServer):
516
"""Tests invalid status from server.
518
Both implementations raises the same error as for a bad status.
521
def create_transport_readonly_server(self):
522
return HttpServer(InvalidStatusRequestHandler)
525
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
526
TestCaseWithWebserver):
527
"""Tests invalid status server for urllib implementation"""
529
_transport = HttpTransport_urllib
532
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
533
TestInvalidStatusServer,
534
TestCaseWithWebserver):
535
"""Tests invalid status server for pycurl implementation"""
538
class TestBadProtocolServer(object):
539
"""Tests bad protocol from server."""
541
def create_transport_readonly_server(self):
542
return HttpServer(BadProtocolRequestHandler)
544
def test_http_has(self):
545
server = self.get_readonly_server()
546
t = self._transport(server.get_url())
547
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
549
def test_http_get(self):
550
server = self.get_readonly_server()
551
t = self._transport(server.get_url())
552
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
555
class TestBadProtocolServer_urllib(TestBadProtocolServer,
556
TestCaseWithWebserver):
557
"""Tests bad protocol server for urllib implementation"""
559
_transport = HttpTransport_urllib
561
# curl don't check the protocol version
562
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
563
# TestBadProtocolServer,
564
# TestCaseWithWebserver):
565
# """Tests bad protocol server for pycurl implementation"""
568
class TestForbiddenServer(object):
569
"""Tests forbidden server"""
571
def create_transport_readonly_server(self):
572
return HttpServer(ForbiddenRequestHandler)
574
def test_http_has(self):
575
server = self.get_readonly_server()
576
t = self._transport(server.get_url())
577
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
579
def test_http_get(self):
580
server = self.get_readonly_server()
581
t = self._transport(server.get_url())
582
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
585
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
586
"""Tests forbidden server for urllib implementation"""
588
_transport = HttpTransport_urllib
591
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
593
TestCaseWithWebserver):
594
"""Tests forbidden server for pycurl implementation"""
597
class TestRecordingServer(TestCase):
599
def test_create(self):
600
server = RecordingServer(expect_body_tail=None)
601
self.assertEqual('', server.received_bytes)
602
self.assertEqual(None, server.host)
603
self.assertEqual(None, server.port)
605
def test_setUp_and_tearDown(self):
606
server = RecordingServer(expect_body_tail=None)
609
self.assertNotEqual(None, server.host)
610
self.assertNotEqual(None, server.port)
613
self.assertEqual(None, server.host)
614
self.assertEqual(None, server.port)
616
def test_send_receive_bytes(self):
617
server = RecordingServer(expect_body_tail='c')
619
self.addCleanup(server.tearDown)
620
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
621
sock.connect((server.host, server.port))
623
self.assertEqual('HTTP/1.1 200 OK\r\n',
624
osutils.recv_all(sock, 4096))
625
self.assertEqual('abc', server.received_bytes)
628
class TestRangeRequestServer(object):
629
"""Tests readv requests against server.
631
This MUST be used by daughter classes that also inherit from
632
TestCaseWithWebserver.
634
We can't inherit directly from TestCaseWithWebserver or the
635
test framework will try to create an instance which cannot
636
run, its implementation being incomplete.
640
TestCaseWithWebserver.setUp(self)
641
self.build_tree_contents([('a', '0123456789')],)
643
def test_readv(self):
644
server = self.get_readonly_server()
645
t = self._transport(server.get_url())
646
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
647
self.assertEqual(l[0], (0, '0'))
648
self.assertEqual(l[1], (1, '1'))
649
self.assertEqual(l[2], (3, '34'))
650
self.assertEqual(l[3], (9, '9'))
652
def test_readv_out_of_order(self):
653
server = self.get_readonly_server()
654
t = self._transport(server.get_url())
655
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
656
self.assertEqual(l[0], (1, '1'))
657
self.assertEqual(l[1], (9, '9'))
658
self.assertEqual(l[2], (0, '0'))
659
self.assertEqual(l[3], (3, '34'))
661
def test_readv_invalid_ranges(self):
662
server = self.get_readonly_server()
663
t = self._transport(server.get_url())
665
# This is intentionally reading off the end of the file
666
# since we are sure that it cannot get there
667
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
668
t.readv, 'a', [(1,1), (8,10)])
670
# This is trying to seek past the end of the file, it should
671
# also raise a special error
672
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
673
t.readv, 'a', [(12,2)])
676
class TestSingleRangeRequestServer(TestRangeRequestServer):
677
"""Test readv against a server which accept only single range requests"""
679
def create_transport_readonly_server(self):
680
return HttpServer(SingleRangeRequestHandler)
683
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
684
TestCaseWithWebserver):
685
"""Tests single range requests accepting server for urllib implementation"""
687
_transport = HttpTransport_urllib
690
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
691
TestSingleRangeRequestServer,
692
TestCaseWithWebserver):
693
"""Tests single range requests accepting server for pycurl implementation"""
696
class TestNoRangeRequestServer(TestRangeRequestServer):
697
"""Test readv against a server which do not accept range requests"""
699
def create_transport_readonly_server(self):
700
return HttpServer(NoRangeRequestHandler)
703
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
704
TestCaseWithWebserver):
705
"""Tests range requests refusing server for urllib implementation"""
707
_transport = HttpTransport_urllib
710
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
711
TestNoRangeRequestServer,
712
TestCaseWithWebserver):
713
"""Tests range requests refusing server for pycurl implementation"""
716
class TestHttpProxyWhiteBox(TestCase):
717
"""Whitebox test proxy http authorization.
719
Only the urllib implementation is tested here.
729
def _install_env(self, env):
730
for name, value in env.iteritems():
731
self._old_env[name] = osutils.set_or_unset_env(name, value)
733
def _restore_env(self):
734
for name, value in self._old_env.iteritems():
735
osutils.set_or_unset_env(name, value)
737
def _proxied_request(self):
738
handler = ProxyHandler(PasswordManager())
739
request = Request('GET','http://baz/buzzle')
740
handler.set_proxy(request, 'http')
743
def test_empty_user(self):
744
self._install_env({'http_proxy': 'http://bar.com'})
745
request = self._proxied_request()
746
self.assertFalse(request.headers.has_key('Proxy-authorization'))
748
def test_invalid_proxy(self):
749
"""A proxy env variable without scheme"""
750
self._install_env({'http_proxy': 'host:1234'})
751
self.assertRaises(errors.InvalidURL, self._proxied_request)
754
class TestProxyHttpServer(object):
755
"""Tests proxy server.
757
This MUST be used by daughter classes that also inherit from
758
TestCaseWithTwoWebservers.
760
We can't inherit directly from TestCaseWithTwoWebservers or
761
the test framework will try to create an instance which
762
cannot run, its implementation being incomplete.
764
Be aware that we do not setup a real proxy here. Instead, we
765
check that the *connection* goes through the proxy by serving
766
different content (the faked proxy server append '-proxied'
770
# FIXME: We don't have an https server available, so we don't
771
# test https connections.
773
# FIXME: Once the test suite is better fitted to test
774
# authorization schemes, test proxy authorizations too (see
778
TestCaseWithTwoWebservers.setUp(self)
779
self.build_tree_contents([('foo', 'contents of foo\n'),
780
('foo-proxied', 'proxied contents of foo\n')])
781
# Let's setup some attributes for tests
782
self.server = self.get_readonly_server()
783
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
784
self.no_proxy_host = self.proxy_address
785
# The secondary server is the proxy
786
self.proxy = self.get_secondary_server()
787
self.proxy_url = self.proxy.get_url()
790
def create_transport_secondary_server(self):
791
"""Creates an http server that will serve files with
792
'-proxied' appended to their names.
796
def _install_env(self, env):
797
for name, value in env.iteritems():
798
self._old_env[name] = osutils.set_or_unset_env(name, value)
800
def _restore_env(self):
801
for name, value in self._old_env.iteritems():
802
osutils.set_or_unset_env(name, value)
804
def proxied_in_env(self, env):
805
self._install_env(env)
806
url = self.server.get_url()
807
t = self._transport(url)
809
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
813
def not_proxied_in_env(self, env):
814
self._install_env(env)
815
url = self.server.get_url()
816
t = self._transport(url)
818
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
822
def test_http_proxy(self):
823
self.proxied_in_env({'http_proxy': self.proxy_url})
825
def test_HTTP_PROXY(self):
826
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
828
def test_all_proxy(self):
829
self.proxied_in_env({'all_proxy': self.proxy_url})
831
def test_ALL_PROXY(self):
832
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
834
def test_http_proxy_with_no_proxy(self):
835
self.not_proxied_in_env({'http_proxy': self.proxy_url,
836
'no_proxy': self.no_proxy_host})
838
def test_HTTP_PROXY_with_NO_PROXY(self):
839
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
840
'NO_PROXY': self.no_proxy_host})
842
def test_all_proxy_with_no_proxy(self):
843
self.not_proxied_in_env({'all_proxy': self.proxy_url,
844
'no_proxy': self.no_proxy_host})
846
def test_ALL_PROXY_with_NO_PROXY(self):
847
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
848
'NO_PROXY': self.no_proxy_host})
850
def test_http_proxy_without_scheme(self):
851
self.assertRaises(errors.InvalidURL,
853
{'http_proxy': self.proxy_address})
856
class TestProxyHttpServer_urllib(TestProxyHttpServer,
857
TestCaseWithTwoWebservers):
858
"""Tests proxy server for urllib implementation"""
860
_transport = HttpTransport_urllib
863
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
865
TestCaseWithTwoWebservers):
866
"""Tests proxy server for pycurl implementation"""
869
TestProxyHttpServer.setUp(self)
870
# Oh my ! pycurl does not check for the port as part of
871
# no_proxy :-( So we just test the host part
872
self.no_proxy_host = 'localhost'
874
def test_HTTP_PROXY(self):
875
# pycurl do not check HTTP_PROXY for security reasons
876
# (for use in a CGI context that we do not care
877
# about. Should we ?)
880
def test_HTTP_PROXY_with_NO_PROXY(self):
883
def test_http_proxy_without_scheme(self):
884
# pycurl *ignores* invalid proxy env variables. If that
885
# ever change in the future, this test will fail
886
# indicating that pycurl do not ignore anymore such
888
self.not_proxied_in_env({'http_proxy': self.proxy_address})
891
class TestRanges(object):
892
"""Test the Range header in GET methods..
894
This MUST be used by daughter classes that also inherit from
895
TestCaseWithWebserver.
897
We can't inherit directly from TestCaseWithWebserver or the
898
test framework will try to create an instance which cannot
899
run, its implementation being incomplete.
903
TestCaseWithWebserver.setUp(self)
904
self.build_tree_contents([('a', '0123456789')],)
905
server = self.get_readonly_server()
906
self.transport = self._transport(server.get_url())
908
def _file_contents(self, relpath, ranges, tail_amount=0):
909
code, data = self.transport._get(relpath, ranges)
910
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
911
for start, end in ranges:
913
yield data.read(end - start + 1)
915
def _file_tail(self, relpath, tail_amount):
916
code, data = self.transport._get(relpath, [], tail_amount)
917
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
918
data.seek(-tail_amount + 1, 2)
919
return data.read(tail_amount)
921
def test_range_header(self):
923
map(self.assertEqual,['0', '234'],
924
list(self._file_contents('a', [(0,0), (2,4)])),)
926
self.assertEqual('789', self._file_tail('a', 3))
927
# Syntactically invalid range
928
self.assertRaises(errors.InvalidRange,
929
self.transport._get, 'a', [(4, 3)])
930
# Semantically invalid range
931
self.assertRaises(errors.InvalidRange,
932
self.transport._get, 'a', [(42, 128)])
935
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
936
"""Test the Range header in GET methods for urllib implementation"""
938
_transport = HttpTransport_urllib
941
class TestRanges_pycurl(TestWithTransport_pycurl,
943
TestCaseWithWebserver):
944
"""Test the Range header in GET methods for pycurl implementation"""
947
class TestHTTPRedirections(object):
948
"""Test redirection between http servers.
950
This MUST be used by daughter classes that also inherit from
951
TestCaseWithRedirectedWebserver.
953
We can't inherit directly from TestCaseWithTwoWebservers or the
954
test framework will try to create an instance which cannot
955
run, its implementation being incomplete.
958
def create_transport_secondary_server(self):
959
"""Create the secondary server redirecting to the primary server"""
960
new = self.get_readonly_server()
962
redirecting = HTTPServerRedirecting()
963
redirecting.redirect_to(new.host, new.port)
967
super(TestHTTPRedirections, self).setUp()
968
self.build_tree_contents([('a', '0123456789'),
970
'# Bazaar revision bundle v0.9\n#\n')
973
self.old_transport = self._transport(self.old_server.get_url())
975
def test_redirected(self):
976
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
977
t = self._transport(self.new_server.get_url())
978
self.assertEqual('0123456789', t.get('a').read())
980
def test_read_redirected_bundle_from_url(self):
981
from bzrlib.bundle import read_bundle_from_url
982
url = self.old_transport.abspath('bundle')
983
bundle = read_bundle_from_url(url)
984
# If read_bundle_from_url was successful we get an empty bundle
985
self.assertEqual([], bundle.revisions)
988
class TestHTTPRedirections_urllib(TestHTTPRedirections,
989
TestCaseWithRedirectedWebserver):
990
"""Tests redirections for urllib implementation"""
992
_transport = HttpTransport_urllib
996
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
997
TestHTTPRedirections,
998
TestCaseWithRedirectedWebserver):
999
"""Tests redirections for pycurl implementation"""
1002
class RedirectedRequest(Request):
1003
"""Request following redirections"""
1005
init_orig = Request.__init__
1007
def __init__(self, method, url, *args, **kwargs):
1008
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1009
self.follow_redirections = True
1012
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1013
"""Test redirections provided by urllib.
1015
http implementations do not redirect silently anymore (they
1016
do not redirect at all in fact). The mechanism is still in
1017
place at the _urllib2_wrappers.Request level and these tests
1020
For the pycurl implementation
1021
the redirection have been deleted as we may deprecate pycurl
1022
and I have no place to keep a working implementation.
1026
_transport = HttpTransport_urllib
1029
super(TestHTTPSilentRedirections_urllib, self).setUp()
1030
self.setup_redirected_request()
1031
self.addCleanup(self.cleanup_redirected_request)
1032
self.build_tree_contents([('a','a'),
1034
('1/a', 'redirected once'),
1036
('2/a', 'redirected twice'),
1038
('3/a', 'redirected thrice'),
1040
('4/a', 'redirected 4 times'),
1042
('5/a', 'redirected 5 times'),
1045
self.old_transport = self._transport(self.old_server.get_url())
1047
def setup_redirected_request(self):
1048
self.original_class = _urllib2_wrappers.Request
1049
_urllib2_wrappers.Request = RedirectedRequest
1051
def cleanup_redirected_request(self):
1052
_urllib2_wrappers.Request = self.original_class
1054
def create_transport_secondary_server(self):
1055
"""Create the secondary server, redirections are defined in the tests"""
1056
return HTTPServerRedirecting()
1058
def test_one_redirection(self):
1059
t = self.old_transport
1061
req = RedirectedRequest('GET', t.abspath('a'))
1062
req.follow_redirections = True
1063
new_prefix = 'http://%s:%s' % (self.new_server.host,
1064
self.new_server.port)
1065
self.old_server.redirections = \
1066
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1067
self.assertEquals('redirected once',t._perform(req).read())
1069
def test_five_redirections(self):
1070
t = self.old_transport
1072
req = RedirectedRequest('GET', t.abspath('a'))
1073
req.follow_redirections = True
1074
old_prefix = 'http://%s:%s' % (self.old_server.host,
1075
self.old_server.port)
1076
new_prefix = 'http://%s:%s' % (self.new_server.host,
1077
self.new_server.port)
1078
self.old_server.redirections = \
1079
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1080
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1081
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1082
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1083
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1085
self.assertEquals('redirected 5 times',t._perform(req).read())
1088
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1089
"""Test transport.do_catching_redirections.
1091
We arbitrarily choose to use urllib transports
1094
_transport = HttpTransport_urllib
1097
super(TestDoCatchRedirections, self).setUp()
1098
self.build_tree_contents([('a', '0123456789'),],)
1100
self.old_transport = self._transport(self.old_server.get_url())
1102
def get_a(self, transport):
1103
return transport.get('a')
1105
def test_no_redirection(self):
1106
t = self._transport(self.new_server.get_url())
1108
# We use None for redirected so that we fail if redirected
1109
self.assertEquals('0123456789',
1110
do_catching_redirections(self.get_a, t, None).read())
1112
def test_one_redirection(self):
1113
self.redirections = 0
1115
def redirected(transport, exception, redirection_notice):
1116
self.redirections += 1
1117
dir, file = urlutils.split(exception.target)
1118
return self._transport(dir)
1120
self.assertEquals('0123456789',
1121
do_catching_redirections(self.get_a,
1125
self.assertEquals(1, self.redirections)
1127
def test_redirection_loop(self):
1129
def redirected(transport, exception, redirection_notice):
1130
# By using the redirected url as a base dir for the
1131
# *old* transport, we create a loop: a => a/a =>
1133
return self.old_transport.clone(exception.target)
1135
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1136
self.get_a, self.old_transport, redirected)
1139
class TestAuth(object):
1140
"""Test some authentication scheme specified by daughter class.
1142
This MUST be used by daughter classes that also inherit from
1143
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1147
"""Set up the test environment
1149
Daughter classes should set up their own environment
1150
(including self.server) and explicitely call this
1151
method. This is needed because we want to reuse the same
1152
tests for proxy and no-proxy accesses which have
1153
different ways of setting self.server.
1155
self.build_tree_contents([('a', 'contents of a\n'),
1156
('b', 'contents of b\n'),])
1157
self.old_factory = ui.ui_factory
1158
self.old_stdout = sys.stdout
1159
sys.stdout = StringIOWrapper()
1160
self.addCleanup(self.restoreUIFactory)
1162
def restoreUIFactory(self):
1163
ui.ui_factory = self.old_factory
1164
sys.stdout = self.old_stdout
1166
def get_user_url(self, user=None, password=None):
1167
"""Build an url embedding user and password"""
1168
url = '%s://' % self.server._url_protocol
1169
if user is not None:
1171
if password is not None:
1172
url += ':' + password
1174
url += '%s:%s/' % (self.server.host, self.server.port)
1177
def test_no_user(self):
1178
self.server.add_user('joe', 'foo')
1179
t = self.get_user_transport()
1180
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1181
# Only one 'Authentication Required' error should occur
1182
self.assertEqual(1, self.server.auth_required_errors)
1184
def test_empty_pass(self):
1185
self.server.add_user('joe', '')
1186
t = self.get_user_transport('joe', '')
1187
self.assertEqual('contents of a\n', t.get('a').read())
1188
# Only one 'Authentication Required' error should occur
1189
self.assertEqual(1, self.server.auth_required_errors)
1191
def test_user_pass(self):
1192
self.server.add_user('joe', 'foo')
1193
t = self.get_user_transport('joe', 'foo')
1194
self.assertEqual('contents of a\n', t.get('a').read())
1195
# Only one 'Authentication Required' error should occur
1196
self.assertEqual(1, self.server.auth_required_errors)
1198
def test_unknown_user(self):
1199
self.server.add_user('joe', 'foo')
1200
t = self.get_user_transport('bill', 'foo')
1201
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1202
# Two 'Authentication Required' errors should occur (the
1203
# initial 'who are you' and 'I don't know you, who are
1205
self.assertEqual(2, self.server.auth_required_errors)
1207
def test_wrong_pass(self):
1208
self.server.add_user('joe', 'foo')
1209
t = self.get_user_transport('joe', 'bar')
1210
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1211
# Two 'Authentication Required' errors should occur (the
1212
# initial 'who are you' and 'this is not you, who are you')
1213
self.assertEqual(2, self.server.auth_required_errors)
1215
def test_prompt_for_password(self):
1216
self.server.add_user('joe', 'foo')
1217
t = self.get_user_transport('joe', None)
1218
ui.ui_factory = TestUIFactory(stdin='foo\n')
1219
self.assertEqual('contents of a\n',t.get('a').read())
1220
# stdin should be empty
1221
self.assertEqual('', ui.ui_factory.stdin.readline())
1222
# And we shouldn't prompt again for a different request
1223
# against the same transport.
1224
self.assertEqual('contents of b\n',t.get('b').read())
1226
# And neither against a clone
1227
self.assertEqual('contents of b\n',t2.get('b').read())
1228
# Only one 'Authentication Required' error should occur
1229
self.assertEqual(1, self.server.auth_required_errors)
1232
class TestHTTPAuth(TestAuth):
1233
"""Test HTTP authentication schemes.
1235
Daughter classes MUST inherit from TestCaseWithWebserver too.
1238
_auth_header = 'Authorization'
1241
TestCaseWithWebserver.setUp(self)
1242
self.server = self.get_readonly_server()
1243
TestAuth.setUp(self)
1245
def get_user_transport(self, user=None, password=None):
1246
return self._transport(self.get_user_url(user, password))
1249
class TestProxyAuth(TestAuth):
1250
"""Test proxy authentication schemes.
1252
Daughter classes MUST also inherit from TestCaseWithWebserver.
1254
_auth_header = 'Proxy-authorization'
1257
TestCaseWithWebserver.setUp(self)
1258
self.server = self.get_readonly_server()
1260
self.addCleanup(self._restore_env)
1261
TestAuth.setUp(self)
1262
# Override the contents to avoid false positives
1263
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1264
('b', 'not proxied contents of b\n'),
1265
('a-proxied', 'contents of a\n'),
1266
('b-proxied', 'contents of b\n'),
1269
def get_user_transport(self, user=None, password=None):
1270
self._install_env({'all_proxy': self.get_user_url(user, password)})
1271
return self._transport(self.server.get_url())
1273
def _install_env(self, env):
1274
for name, value in env.iteritems():
1275
self._old_env[name] = osutils.set_or_unset_env(name, value)
1277
def _restore_env(self):
1278
for name, value in self._old_env.iteritems():
1279
osutils.set_or_unset_env(name, value)
1282
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1283
"""Test http basic authentication scheme"""
1285
_transport = HttpTransport_urllib
1287
def create_transport_readonly_server(self):
1288
return HTTPBasicAuthServer()
1291
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1292
"""Test proxy basic authentication scheme"""
1294
_transport = HttpTransport_urllib
1296
def create_transport_readonly_server(self):
1297
return ProxyBasicAuthServer()
1300
class TestDigestAuth(object):
1301
"""Digest Authentication specific tests"""
1303
def test_changing_nonce(self):
1304
self.server.add_user('joe', 'foo')
1305
t = self.get_user_transport('joe', 'foo')
1306
self.assertEqual('contents of a\n', t.get('a').read())
1307
self.assertEqual('contents of b\n', t.get('b').read())
1308
# Only one 'Authentication Required' error should have
1310
self.assertEqual(1, self.server.auth_required_errors)
1311
# The server invalidates the current nonce
1312
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1313
self.assertEqual('contents of a\n', t.get('a').read())
1314
# Two 'Authentication Required' errors should occur (the
1315
# initial 'who are you' and a second 'who are you' with the new nonce)
1316
self.assertEqual(2, self.server.auth_required_errors)
1319
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1320
"""Test http digest authentication scheme"""
1322
_transport = HttpTransport_urllib
1324
def create_transport_readonly_server(self):
1325
return HTTPDigestAuthServer()
1328
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1329
TestCaseWithWebserver):
1330
"""Test proxy digest authentication scheme"""
1332
_transport = HttpTransport_urllib
1334
def create_transport_readonly_server(self):
1335
return ProxyDigestAuthServer()
added
removed
bzrlib/selftest/testhttp.py
