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)
3
from bzrlib.selftest import TestCase
4
from bzrlib.transport.http import HttpTransport
166
6
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
7
def test_abs_url(self):
195
8
"""Construction of absolute http URLs"""
196
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
9
t = HttpTransport('http://bazaar-ng.org/bzr/bzr.dev/')
197
10
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')
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')
201
17
eq(t.abspath('.bzr/1//2/./3'),
202
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
18
'http://bazaar-ng.org/bzr/bzr.dev/.bzr/1/2/3')
204
20
def test_invalid_http_urls(self):
205
21
"""Trap invalid construction of urls"""
206
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
207
self.assertRaises(ValueError, t.abspath, '.bzr/')
208
t = self._transport('http://http://bazaar-vcs.org/bzr/bzr.dev/')
209
self.assertRaises((errors.InvalidURL, errors.ConnectionError),
22
t = HttpTransport('http://bazaar-ng.org/bzr/bzr.dev/')
23
self.assertRaises(ValueError,
26
self.assertRaises(ValueError,
212
30
def test_http_root_urls(self):
213
31
"""Construction of URLs from server root"""
214
t = self._transport('http://bzr.ozlabs.org/')
32
t = HttpTransport('http://bzr.ozlabs.org/')
215
33
eq = self.assertEqualDiff
216
34
eq(t.abspath('.bzr/tree-version'),
217
35
'http://bzr.ozlabs.org/.bzr/tree-version')
219
def test_http_impl_urls(self):
220
"""There are servers which ask for particular clients to connect"""
221
server = self._server()
224
url = server.get_url()
225
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
230
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
231
"""Test http urls with urllib"""
233
_transport = HttpTransport_urllib
234
_server = HttpServer_urllib
235
_qualified_prefix = 'http+urllib'
238
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
240
"""Test http urls with pycurl"""
242
_server = HttpServer_PyCurl
243
_qualified_prefix = 'http+pycurl'
245
# TODO: This should really be moved into another pycurl
246
# specific test. When https tests will be implemented, take
247
# this one into account.
248
def test_pycurl_without_https_support(self):
249
"""Test that pycurl without SSL do not fail with a traceback.
251
For the purpose of the test, we force pycurl to ignore
252
https by supplying a fake version_info that do not
258
raise TestSkipped('pycurl not present')
259
# Now that we have pycurl imported, we can fake its version_info
260
# This was taken from a windows pycurl without SSL
262
pycurl.version_info = lambda : (2,
270
('ftp', 'gopher', 'telnet',
271
'dict', 'ldap', 'http', 'file'),
275
self.assertRaises(errors.DependencyNotPresent, self._transport,
276
'https://launchpad.net')
278
class TestHttpConnections(object):
279
"""Test the http connections.
281
This MUST be used by daughter classes that also inherit from
282
TestCaseWithWebserver.
284
We can't inherit directly from TestCaseWithWebserver or the
285
test framework will try to create an instance which cannot
286
run, its implementation being incomplete.
290
TestCaseWithWebserver.setUp(self)
291
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
292
transport=self.get_transport())
294
def test_http_has(self):
295
server = self.get_readonly_server()
296
t = self._transport(server.get_url())
297
self.assertEqual(t.has('foo/bar'), True)
298
self.assertEqual(len(server.logs), 1)
299
self.assertContainsRe(server.logs[0],
300
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
302
def test_http_has_not_found(self):
303
server = self.get_readonly_server()
304
t = self._transport(server.get_url())
305
self.assertEqual(t.has('not-found'), False)
306
self.assertContainsRe(server.logs[1],
307
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
309
def test_http_get(self):
310
server = self.get_readonly_server()
311
t = self._transport(server.get_url())
312
fp = t.get('foo/bar')
313
self.assertEqualDiff(
315
'contents of foo/bar\n')
316
self.assertEqual(len(server.logs), 1)
317
self.assertTrue(server.logs[0].find(
318
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
319
% bzrlib.__version__) > -1)
321
def test_get_smart_medium(self):
322
# For HTTP, get_smart_medium should return the transport object.
323
server = self.get_readonly_server()
324
http_transport = self._transport(server.get_url())
325
medium = http_transport.get_smart_medium()
326
self.assertIs(medium, http_transport)
328
def test_has_on_bogus_host(self):
329
# Get a free address and don't 'accept' on it, so that we
330
# can be sure there is no http handler there, but set a
331
# reasonable timeout to not slow down tests too much.
332
default_timeout = socket.getdefaulttimeout()
334
socket.setdefaulttimeout(2)
336
s.bind(('localhost', 0))
337
t = self._transport('http://%s:%s/' % s.getsockname())
338
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
340
socket.setdefaulttimeout(default_timeout)
343
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
344
"""Test http connections with urllib"""
346
_transport = HttpTransport_urllib
350
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
352
TestCaseWithWebserver):
353
"""Test http connections with pycurl"""
356
class TestHttpTransportRegistration(TestCase):
357
"""Test registrations of various http implementations"""
359
def test_http_registered(self):
360
# urlllib should always be present
361
t = get_transport('http+urllib://bzr.google.com/')
362
self.assertIsInstance(t, Transport)
363
self.assertIsInstance(t, HttpTransport_urllib)
366
class TestPost(object):
368
def _test_post_body_is_received(self, scheme):
369
server = RecordingServer(expect_body_tail='end-of-body')
371
self.addCleanup(server.tearDown)
372
url = '%s://%s:%s/' % (scheme, server.host, server.port)
374
http_transport = get_transport(url)
375
except errors.UnsupportedProtocol:
376
raise TestSkipped('%s not available' % scheme)
377
code, response = http_transport._post('abc def end-of-body')
379
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
380
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
381
# The transport should not be assuming that the server can accept
382
# chunked encoding the first time it connects, because HTTP/1.1, so we
383
# check for the literal string.
385
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
388
class TestPost_urllib(TestCase, TestPost):
389
"""TestPost for urllib implementation"""
391
_transport = HttpTransport_urllib
393
def test_post_body_is_received_urllib(self):
394
self._test_post_body_is_received('http+urllib')
397
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
398
"""TestPost for pycurl implementation"""
400
def test_post_body_is_received_pycurl(self):
401
self._test_post_body_is_received('http+pycurl')
404
class TestRangeHeader(TestCase):
405
"""Test range_header method"""
407
def check_header(self, value, ranges=[], tail=0):
408
offsets = [ (start, end - start + 1) for start, end in ranges]
409
coalesce = Transport._coalesce_offsets
410
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
411
range_header = HttpTransportBase._range_header
412
self.assertEqual(value, range_header(coalesced, tail))
414
def test_range_header_single(self):
415
self.check_header('0-9', ranges=[(0,9)])
416
self.check_header('100-109', ranges=[(100,109)])
418
def test_range_header_tail(self):
419
self.check_header('-10', tail=10)
420
self.check_header('-50', tail=50)
422
def test_range_header_multi(self):
423
self.check_header('0-9,100-200,300-5000',
424
ranges=[(0,9), (100, 200), (300,5000)])
426
def test_range_header_mixed(self):
427
self.check_header('0-9,300-5000,-50',
428
ranges=[(0,9), (300,5000)],
432
class TestWallServer(object):
433
"""Tests exceptions during the connection phase"""
435
def create_transport_readonly_server(self):
436
return HttpServer(WallRequestHandler)
438
def test_http_has(self):
439
server = self.get_readonly_server()
440
t = self._transport(server.get_url())
441
# Unfortunately httplib (see HTTPResponse._read_status
442
# for details) make no distinction between a closed
443
# socket and badly formatted status line, so we can't
444
# just test for ConnectionError, we have to test
445
# InvalidHttpResponse too.
446
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
449
def test_http_get(self):
450
server = self.get_readonly_server()
451
t = self._transport(server.get_url())
452
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
456
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
457
"""Tests "wall" server for urllib implementation"""
459
_transport = HttpTransport_urllib
462
class TestWallServer_pycurl(TestWithTransport_pycurl,
464
TestCaseWithWebserver):
465
"""Tests "wall" server for pycurl implementation"""
468
class TestBadStatusServer(object):
469
"""Tests bad status from server."""
471
def create_transport_readonly_server(self):
472
return HttpServer(BadStatusRequestHandler)
474
def test_http_has(self):
475
server = self.get_readonly_server()
476
t = self._transport(server.get_url())
477
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
479
def test_http_get(self):
480
server = self.get_readonly_server()
481
t = self._transport(server.get_url())
482
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
485
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
486
"""Tests bad status server for urllib implementation"""
488
_transport = HttpTransport_urllib
491
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
493
TestCaseWithWebserver):
494
"""Tests bad status server for pycurl implementation"""
497
class TestInvalidStatusServer(TestBadStatusServer):
498
"""Tests invalid status from server.
500
Both implementations raises the same error as for a bad status.
503
def create_transport_readonly_server(self):
504
return HttpServer(InvalidStatusRequestHandler)
507
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
508
TestCaseWithWebserver):
509
"""Tests invalid status server for urllib implementation"""
511
_transport = HttpTransport_urllib
514
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
515
TestInvalidStatusServer,
516
TestCaseWithWebserver):
517
"""Tests invalid status server for pycurl implementation"""
520
class TestBadProtocolServer(object):
521
"""Tests bad protocol from server."""
523
def create_transport_readonly_server(self):
524
return HttpServer(BadProtocolRequestHandler)
526
def test_http_has(self):
527
server = self.get_readonly_server()
528
t = self._transport(server.get_url())
529
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
531
def test_http_get(self):
532
server = self.get_readonly_server()
533
t = self._transport(server.get_url())
534
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
537
class TestBadProtocolServer_urllib(TestBadProtocolServer,
538
TestCaseWithWebserver):
539
"""Tests bad protocol server for urllib implementation"""
541
_transport = HttpTransport_urllib
543
# curl don't check the protocol version
544
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
545
# TestBadProtocolServer,
546
# TestCaseWithWebserver):
547
# """Tests bad protocol server for pycurl implementation"""
550
class TestForbiddenServer(object):
551
"""Tests forbidden server"""
553
def create_transport_readonly_server(self):
554
return HttpServer(ForbiddenRequestHandler)
556
def test_http_has(self):
557
server = self.get_readonly_server()
558
t = self._transport(server.get_url())
559
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
561
def test_http_get(self):
562
server = self.get_readonly_server()
563
t = self._transport(server.get_url())
564
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
567
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
568
"""Tests forbidden server for urllib implementation"""
570
_transport = HttpTransport_urllib
573
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
575
TestCaseWithWebserver):
576
"""Tests forbidden server for pycurl implementation"""
579
class TestRecordingServer(TestCase):
581
def test_create(self):
582
server = RecordingServer(expect_body_tail=None)
583
self.assertEqual('', server.received_bytes)
584
self.assertEqual(None, server.host)
585
self.assertEqual(None, server.port)
587
def test_setUp_and_tearDown(self):
588
server = RecordingServer(expect_body_tail=None)
591
self.assertNotEqual(None, server.host)
592
self.assertNotEqual(None, server.port)
595
self.assertEqual(None, server.host)
596
self.assertEqual(None, server.port)
598
def test_send_receive_bytes(self):
599
server = RecordingServer(expect_body_tail='c')
601
self.addCleanup(server.tearDown)
602
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
603
sock.connect((server.host, server.port))
605
self.assertEqual('HTTP/1.1 200 OK\r\n',
606
osutils.recv_all(sock, 4096))
607
self.assertEqual('abc', server.received_bytes)
610
class TestRangeRequestServer(object):
611
"""Tests readv requests against server.
613
This MUST be used by daughter classes that also inherit from
614
TestCaseWithWebserver.
616
We can't inherit directly from TestCaseWithWebserver or the
617
test framework will try to create an instance which cannot
618
run, its implementation being incomplete.
622
TestCaseWithWebserver.setUp(self)
623
self.build_tree_contents([('a', '0123456789')],)
625
def test_readv(self):
626
server = self.get_readonly_server()
627
t = self._transport(server.get_url())
628
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
629
self.assertEqual(l[0], (0, '0'))
630
self.assertEqual(l[1], (1, '1'))
631
self.assertEqual(l[2], (3, '34'))
632
self.assertEqual(l[3], (9, '9'))
634
def test_readv_out_of_order(self):
635
server = self.get_readonly_server()
636
t = self._transport(server.get_url())
637
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
638
self.assertEqual(l[0], (1, '1'))
639
self.assertEqual(l[1], (9, '9'))
640
self.assertEqual(l[2], (0, '0'))
641
self.assertEqual(l[3], (3, '34'))
643
def test_readv_invalid_ranges(self):
644
server = self.get_readonly_server()
645
t = self._transport(server.get_url())
647
# This is intentionally reading off the end of the file
648
# since we are sure that it cannot get there
649
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
650
t.readv, 'a', [(1,1), (8,10)])
652
# This is trying to seek past the end of the file, it should
653
# also raise a special error
654
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
655
t.readv, 'a', [(12,2)])
658
class TestSingleRangeRequestServer(TestRangeRequestServer):
659
"""Test readv against a server which accept only single range requests"""
661
def create_transport_readonly_server(self):
662
return HttpServer(SingleRangeRequestHandler)
665
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
666
TestCaseWithWebserver):
667
"""Tests single range requests accepting server for urllib implementation"""
669
_transport = HttpTransport_urllib
672
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
673
TestSingleRangeRequestServer,
674
TestCaseWithWebserver):
675
"""Tests single range requests accepting server for pycurl implementation"""
678
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
679
"""Test readv against a server which only accept single range requests"""
681
def create_transport_readonly_server(self):
682
return HttpServer(SingleOnlyRangeRequestHandler)
685
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
686
TestCaseWithWebserver):
687
"""Tests single range requests accepting server for urllib implementation"""
689
_transport = HttpTransport_urllib
692
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
693
TestSingleOnlyRangeRequestServer,
694
TestCaseWithWebserver):
695
"""Tests single range requests accepting server for pycurl implementation"""
698
class TestNoRangeRequestServer(TestRangeRequestServer):
699
"""Test readv against a server which do not accept range requests"""
701
def create_transport_readonly_server(self):
702
return HttpServer(NoRangeRequestHandler)
705
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
706
TestCaseWithWebserver):
707
"""Tests range requests refusing server for urllib implementation"""
709
_transport = HttpTransport_urllib
712
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
713
TestNoRangeRequestServer,
714
TestCaseWithWebserver):
715
"""Tests range requests refusing server for pycurl implementation"""
718
class TestLimitedRangeRequestServer(object):
719
"""Tests readv requests against server that errors out on too much ranges.
721
This MUST be used by daughter classes that also inherit from
722
TestCaseWithWebserver.
724
We can't inherit directly from TestCaseWithWebserver or the
725
test framework will try to create an instance which cannot
726
run, its implementation being incomplete.
731
def create_transport_readonly_server(self):
732
# Requests with more range specifiers will error out
733
return LimitedRangeHTTPServer(range_limit=self.range_limit)
735
def get_transport(self):
736
return self._transport(self.get_readonly_server().get_url())
739
TestCaseWithWebserver.setUp(self)
740
# We need to manipulate ranges that correspond to real chunks in the
741
# response, so we build a content appropriately.
742
filler = ''.join(['abcdefghij' for _ in range(102)])
743
content = ''.join(['%04d' % v + filler for v in range(16)])
744
self.build_tree_contents([('a', content)],)
746
def test_few_ranges(self):
747
t = self.get_transport()
748
l = list(t.readv('a', ((0, 4), (1024, 4), )))
749
self.assertEqual(l[0], (0, '0000'))
750
self.assertEqual(l[1], (1024, '0001'))
751
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
753
def test_a_lot_of_ranges(self):
754
t = self.get_transport()
755
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
756
self.assertEqual(l[0], (0, '0000'))
757
self.assertEqual(l[1], (1024, '0001'))
758
self.assertEqual(l[2], (4096, '0004'))
759
self.assertEqual(l[3], (8192, '0008'))
760
# The server will refuse to serve the first request (too much ranges),
761
# a second request will succeeds.
762
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
765
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
766
TestCaseWithWebserver):
767
"""Tests limited range requests server for urllib implementation"""
769
_transport = HttpTransport_urllib
772
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
773
TestLimitedRangeRequestServer,
774
TestCaseWithWebserver):
775
"""Tests limited range requests server for pycurl implementation"""
779
class TestHttpProxyWhiteBox(TestCase):
780
"""Whitebox test proxy http authorization.
782
Only the urllib implementation is tested here.
792
def _install_env(self, env):
793
for name, value in env.iteritems():
794
self._old_env[name] = osutils.set_or_unset_env(name, value)
796
def _restore_env(self):
797
for name, value in self._old_env.iteritems():
798
osutils.set_or_unset_env(name, value)
800
def _proxied_request(self):
801
handler = ProxyHandler(PasswordManager())
802
request = Request('GET','http://baz/buzzle')
803
handler.set_proxy(request, 'http')
806
def test_empty_user(self):
807
self._install_env({'http_proxy': 'http://bar.com'})
808
request = self._proxied_request()
809
self.assertFalse(request.headers.has_key('Proxy-authorization'))
811
def test_invalid_proxy(self):
812
"""A proxy env variable without scheme"""
813
self._install_env({'http_proxy': 'host:1234'})
814
self.assertRaises(errors.InvalidURL, self._proxied_request)
817
class TestProxyHttpServer(object):
818
"""Tests proxy server.
820
This MUST be used by daughter classes that also inherit from
821
TestCaseWithTwoWebservers.
823
We can't inherit directly from TestCaseWithTwoWebservers or
824
the test framework will try to create an instance which
825
cannot run, its implementation being incomplete.
827
Be aware that we do not setup a real proxy here. Instead, we
828
check that the *connection* goes through the proxy by serving
829
different content (the faked proxy server append '-proxied'
833
# FIXME: We don't have an https server available, so we don't
834
# test https connections.
836
# FIXME: Once the test suite is better fitted to test
837
# authorization schemes, test proxy authorizations too (see
841
TestCaseWithTwoWebservers.setUp(self)
842
self.build_tree_contents([('foo', 'contents of foo\n'),
843
('foo-proxied', 'proxied contents of foo\n')])
844
# Let's setup some attributes for tests
845
self.server = self.get_readonly_server()
846
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
847
self.no_proxy_host = self.proxy_address
848
# The secondary server is the proxy
849
self.proxy = self.get_secondary_server()
850
self.proxy_url = self.proxy.get_url()
853
def create_transport_secondary_server(self):
854
"""Creates an http server that will serve files with
855
'-proxied' appended to their names.
859
def _install_env(self, env):
860
for name, value in env.iteritems():
861
self._old_env[name] = osutils.set_or_unset_env(name, value)
863
def _restore_env(self):
864
for name, value in self._old_env.iteritems():
865
osutils.set_or_unset_env(name, value)
867
def proxied_in_env(self, env):
868
self._install_env(env)
869
url = self.server.get_url()
870
t = self._transport(url)
872
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
876
def not_proxied_in_env(self, env):
877
self._install_env(env)
878
url = self.server.get_url()
879
t = self._transport(url)
881
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
885
def test_http_proxy(self):
886
self.proxied_in_env({'http_proxy': self.proxy_url})
888
def test_HTTP_PROXY(self):
889
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
891
def test_all_proxy(self):
892
self.proxied_in_env({'all_proxy': self.proxy_url})
894
def test_ALL_PROXY(self):
895
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
897
def test_http_proxy_with_no_proxy(self):
898
self.not_proxied_in_env({'http_proxy': self.proxy_url,
899
'no_proxy': self.no_proxy_host})
901
def test_HTTP_PROXY_with_NO_PROXY(self):
902
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
903
'NO_PROXY': self.no_proxy_host})
905
def test_all_proxy_with_no_proxy(self):
906
self.not_proxied_in_env({'all_proxy': self.proxy_url,
907
'no_proxy': self.no_proxy_host})
909
def test_ALL_PROXY_with_NO_PROXY(self):
910
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
911
'NO_PROXY': self.no_proxy_host})
913
def test_http_proxy_without_scheme(self):
914
self.assertRaises(errors.InvalidURL,
916
{'http_proxy': self.proxy_address})
919
class TestProxyHttpServer_urllib(TestProxyHttpServer,
920
TestCaseWithTwoWebservers):
921
"""Tests proxy server for urllib implementation"""
923
_transport = HttpTransport_urllib
926
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
928
TestCaseWithTwoWebservers):
929
"""Tests proxy server for pycurl implementation"""
932
TestProxyHttpServer.setUp(self)
933
# Oh my ! pycurl does not check for the port as part of
934
# no_proxy :-( So we just test the host part
935
self.no_proxy_host = 'localhost'
937
def test_HTTP_PROXY(self):
938
# pycurl do not check HTTP_PROXY for security reasons
939
# (for use in a CGI context that we do not care
940
# about. Should we ?)
943
def test_HTTP_PROXY_with_NO_PROXY(self):
946
def test_http_proxy_without_scheme(self):
947
# pycurl *ignores* invalid proxy env variables. If that
948
# ever change in the future, this test will fail
949
# indicating that pycurl do not ignore anymore such
951
self.not_proxied_in_env({'http_proxy': self.proxy_address})
954
class TestRanges(object):
955
"""Test the Range header in GET methods..
957
This MUST be used by daughter classes that also inherit from
958
TestCaseWithWebserver.
960
We can't inherit directly from TestCaseWithWebserver or the
961
test framework will try to create an instance which cannot
962
run, its implementation being incomplete.
966
TestCaseWithWebserver.setUp(self)
967
self.build_tree_contents([('a', '0123456789')],)
968
server = self.get_readonly_server()
969
self.transport = self._transport(server.get_url())
971
def _file_contents(self, relpath, ranges):
972
offsets = [ (start, end - start + 1) for start, end in ranges]
973
coalesce = self.transport._coalesce_offsets
974
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
975
code, data = self.transport._get(relpath, coalesced)
976
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
977
for start, end in ranges:
979
yield data.read(end - start + 1)
981
def _file_tail(self, relpath, tail_amount):
982
code, data = self.transport._get(relpath, [], tail_amount)
983
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
984
data.seek(-tail_amount + 1, 2)
985
return data.read(tail_amount)
987
def test_range_header(self):
989
map(self.assertEqual,['0', '234'],
990
list(self._file_contents('a', [(0,0), (2,4)])),)
992
self.assertEqual('789', self._file_tail('a', 3))
993
# Syntactically invalid range
994
self.assertListRaises(errors.InvalidRange,
995
self._file_contents, 'a', [(4, 3)])
996
# Semantically invalid range
997
self.assertListRaises(errors.InvalidRange,
998
self._file_contents, 'a', [(42, 128)])
1001
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1002
"""Test the Range header in GET methods for urllib implementation"""
1004
_transport = HttpTransport_urllib
1007
class TestRanges_pycurl(TestWithTransport_pycurl,
1009
TestCaseWithWebserver):
1010
"""Test the Range header in GET methods for pycurl implementation"""
1013
class TestHTTPRedirections(object):
1014
"""Test redirection between http servers.
1016
This MUST be used by daughter classes that also inherit from
1017
TestCaseWithRedirectedWebserver.
1019
We can't inherit directly from TestCaseWithTwoWebservers or the
1020
test framework will try to create an instance which cannot
1021
run, its implementation being incomplete.
1024
def create_transport_secondary_server(self):
1025
"""Create the secondary server redirecting to the primary server"""
1026
new = self.get_readonly_server()
1028
redirecting = HTTPServerRedirecting()
1029
redirecting.redirect_to(new.host, new.port)
1033
super(TestHTTPRedirections, self).setUp()
1034
self.build_tree_contents([('a', '0123456789'),
1036
'# Bazaar revision bundle v0.9\n#\n')
1039
self.old_transport = self._transport(self.old_server.get_url())
1041
def test_redirected(self):
1042
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1043
t = self._transport(self.new_server.get_url())
1044
self.assertEqual('0123456789', t.get('a').read())
1046
def test_read_redirected_bundle_from_url(self):
1047
from bzrlib.bundle import read_bundle_from_url
1048
url = self.old_transport.abspath('bundle')
1049
bundle = read_bundle_from_url(url)
1050
# If read_bundle_from_url was successful we get an empty bundle
1051
self.assertEqual([], bundle.revisions)
1054
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1055
TestCaseWithRedirectedWebserver):
1056
"""Tests redirections for urllib implementation"""
1058
_transport = HttpTransport_urllib
1062
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1063
TestHTTPRedirections,
1064
TestCaseWithRedirectedWebserver):
1065
"""Tests redirections for pycurl implementation"""
1068
class RedirectedRequest(Request):
1069
"""Request following redirections"""
1071
init_orig = Request.__init__
1073
def __init__(self, method, url, *args, **kwargs):
1074
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1075
self.follow_redirections = True
1078
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1079
"""Test redirections provided by urllib.
1081
http implementations do not redirect silently anymore (they
1082
do not redirect at all in fact). The mechanism is still in
1083
place at the _urllib2_wrappers.Request level and these tests
1086
For the pycurl implementation
1087
the redirection have been deleted as we may deprecate pycurl
1088
and I have no place to keep a working implementation.
1092
_transport = HttpTransport_urllib
1095
super(TestHTTPSilentRedirections_urllib, self).setUp()
1096
self.setup_redirected_request()
1097
self.addCleanup(self.cleanup_redirected_request)
1098
self.build_tree_contents([('a','a'),
1100
('1/a', 'redirected once'),
1102
('2/a', 'redirected twice'),
1104
('3/a', 'redirected thrice'),
1106
('4/a', 'redirected 4 times'),
1108
('5/a', 'redirected 5 times'),
1111
self.old_transport = self._transport(self.old_server.get_url())
1113
def setup_redirected_request(self):
1114
self.original_class = _urllib2_wrappers.Request
1115
_urllib2_wrappers.Request = RedirectedRequest
1117
def cleanup_redirected_request(self):
1118
_urllib2_wrappers.Request = self.original_class
1120
def create_transport_secondary_server(self):
1121
"""Create the secondary server, redirections are defined in the tests"""
1122
return HTTPServerRedirecting()
1124
def test_one_redirection(self):
1125
t = self.old_transport
1127
req = RedirectedRequest('GET', t.abspath('a'))
1128
req.follow_redirections = True
1129
new_prefix = 'http://%s:%s' % (self.new_server.host,
1130
self.new_server.port)
1131
self.old_server.redirections = \
1132
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1133
self.assertEquals('redirected once',t._perform(req).read())
1135
def test_five_redirections(self):
1136
t = self.old_transport
1138
req = RedirectedRequest('GET', t.abspath('a'))
1139
req.follow_redirections = True
1140
old_prefix = 'http://%s:%s' % (self.old_server.host,
1141
self.old_server.port)
1142
new_prefix = 'http://%s:%s' % (self.new_server.host,
1143
self.new_server.port)
1144
self.old_server.redirections = \
1145
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1146
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1147
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1148
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1149
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1151
self.assertEquals('redirected 5 times',t._perform(req).read())
1154
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1155
"""Test transport.do_catching_redirections.
1157
We arbitrarily choose to use urllib transports
1160
_transport = HttpTransport_urllib
1163
super(TestDoCatchRedirections, self).setUp()
1164
self.build_tree_contents([('a', '0123456789'),],)
1166
self.old_transport = self._transport(self.old_server.get_url())
1168
def get_a(self, transport):
1169
return transport.get('a')
1171
def test_no_redirection(self):
1172
t = self._transport(self.new_server.get_url())
1174
# We use None for redirected so that we fail if redirected
1175
self.assertEquals('0123456789',
1176
do_catching_redirections(self.get_a, t, None).read())
1178
def test_one_redirection(self):
1179
self.redirections = 0
1181
def redirected(transport, exception, redirection_notice):
1182
self.redirections += 1
1183
dir, file = urlutils.split(exception.target)
1184
return self._transport(dir)
1186
self.assertEquals('0123456789',
1187
do_catching_redirections(self.get_a,
1191
self.assertEquals(1, self.redirections)
1193
def test_redirection_loop(self):
1195
def redirected(transport, exception, redirection_notice):
1196
# By using the redirected url as a base dir for the
1197
# *old* transport, we create a loop: a => a/a =>
1199
return self.old_transport.clone(exception.target)
1201
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1202
self.get_a, self.old_transport, redirected)
1205
class TestAuth(object):
1206
"""Test some authentication scheme specified by daughter class.
1208
This MUST be used by daughter classes that also inherit from
1209
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1213
"""Set up the test environment
1215
Daughter classes should set up their own environment
1216
(including self.server) and explicitely call this
1217
method. This is needed because we want to reuse the same
1218
tests for proxy and no-proxy accesses which have
1219
different ways of setting self.server.
1221
self.build_tree_contents([('a', 'contents of a\n'),
1222
('b', 'contents of b\n'),])
1223
self.old_factory = ui.ui_factory
1224
# The following has the unfortunate side-effect of hiding any ouput
1225
# during the tests (including pdb prompts). Feel free to comment them
1226
# for debugging purposes but leave them in place, there are needed to
1227
# run the tests without any console
1228
self.old_stdout = sys.stdout
1229
sys.stdout = StringIOWrapper()
1230
self.addCleanup(self.restoreUIFactory)
1232
def restoreUIFactory(self):
1233
ui.ui_factory = self.old_factory
1234
sys.stdout = self.old_stdout
1236
def get_user_url(self, user=None, password=None):
1237
"""Build an url embedding user and password"""
1238
url = '%s://' % self.server._url_protocol
1239
if user is not None:
1241
if password is not None:
1242
url += ':' + password
1244
url += '%s:%s/' % (self.server.host, self.server.port)
1247
def test_no_user(self):
1248
self.server.add_user('joe', 'foo')
1249
t = self.get_user_transport()
1250
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1251
# Only one 'Authentication Required' error should occur
1252
self.assertEqual(1, self.server.auth_required_errors)
1254
def test_empty_pass(self):
1255
self.server.add_user('joe', '')
1256
t = self.get_user_transport('joe', '')
1257
self.assertEqual('contents of a\n', t.get('a').read())
1258
# Only one 'Authentication Required' error should occur
1259
self.assertEqual(1, self.server.auth_required_errors)
1261
def test_user_pass(self):
1262
self.server.add_user('joe', 'foo')
1263
t = self.get_user_transport('joe', 'foo')
1264
self.assertEqual('contents of a\n', t.get('a').read())
1265
# Only one 'Authentication Required' error should occur
1266
self.assertEqual(1, self.server.auth_required_errors)
1268
def test_unknown_user(self):
1269
self.server.add_user('joe', 'foo')
1270
t = self.get_user_transport('bill', 'foo')
1271
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1272
# Two 'Authentication Required' errors should occur (the
1273
# initial 'who are you' and 'I don't know you, who are
1275
self.assertEqual(2, self.server.auth_required_errors)
1277
def test_wrong_pass(self):
1278
self.server.add_user('joe', 'foo')
1279
t = self.get_user_transport('joe', 'bar')
1280
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1281
# Two 'Authentication Required' errors should occur (the
1282
# initial 'who are you' and 'this is not you, who are you')
1283
self.assertEqual(2, self.server.auth_required_errors)
1285
def test_prompt_for_password(self):
1286
self.server.add_user('joe', 'foo')
1287
t = self.get_user_transport('joe', None)
1288
ui.ui_factory = TestUIFactory(stdin='foo\n')
1289
self.assertEqual('contents of a\n',t.get('a').read())
1290
# stdin should be empty
1291
self.assertEqual('', ui.ui_factory.stdin.readline())
1292
# And we shouldn't prompt again for a different request
1293
# against the same transport.
1294
self.assertEqual('contents of b\n',t.get('b').read())
1296
# And neither against a clone
1297
self.assertEqual('contents of b\n',t2.get('b').read())
1298
# Only one 'Authentication Required' error should occur
1299
self.assertEqual(1, self.server.auth_required_errors)
1302
class TestHTTPAuth(TestAuth):
1303
"""Test HTTP authentication schemes.
1305
Daughter classes MUST inherit from TestCaseWithWebserver too.
1308
_auth_header = 'Authorization'
1311
TestCaseWithWebserver.setUp(self)
1312
self.server = self.get_readonly_server()
1313
TestAuth.setUp(self)
1315
def get_user_transport(self, user=None, password=None):
1316
return self._transport(self.get_user_url(user, password))
1319
class TestProxyAuth(TestAuth):
1320
"""Test proxy authentication schemes.
1322
Daughter classes MUST also inherit from TestCaseWithWebserver.
1324
_auth_header = 'Proxy-authorization'
1327
TestCaseWithWebserver.setUp(self)
1328
self.server = self.get_readonly_server()
1330
self.addCleanup(self._restore_env)
1331
TestAuth.setUp(self)
1332
# Override the contents to avoid false positives
1333
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1334
('b', 'not proxied contents of b\n'),
1335
('a-proxied', 'contents of a\n'),
1336
('b-proxied', 'contents of b\n'),
1339
def get_user_transport(self, user=None, password=None):
1340
self._install_env({'all_proxy': self.get_user_url(user, password)})
1341
return self._transport(self.server.get_url())
1343
def _install_env(self, env):
1344
for name, value in env.iteritems():
1345
self._old_env[name] = osutils.set_or_unset_env(name, value)
1347
def _restore_env(self):
1348
for name, value in self._old_env.iteritems():
1349
osutils.set_or_unset_env(name, value)
1352
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1353
"""Test http basic authentication scheme"""
1355
_transport = HttpTransport_urllib
1357
def create_transport_readonly_server(self):
1358
return HTTPBasicAuthServer()
1361
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1362
"""Test proxy basic authentication scheme"""
1364
_transport = HttpTransport_urllib
1366
def create_transport_readonly_server(self):
1367
return ProxyBasicAuthServer()
1370
class TestDigestAuth(object):
1371
"""Digest Authentication specific tests"""
1373
def test_changing_nonce(self):
1374
self.server.add_user('joe', 'foo')
1375
t = self.get_user_transport('joe', 'foo')
1376
self.assertEqual('contents of a\n', t.get('a').read())
1377
self.assertEqual('contents of b\n', t.get('b').read())
1378
# Only one 'Authentication Required' error should have
1380
self.assertEqual(1, self.server.auth_required_errors)
1381
# The server invalidates the current nonce
1382
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1383
self.assertEqual('contents of a\n', t.get('a').read())
1384
# Two 'Authentication Required' errors should occur (the
1385
# initial 'who are you' and a second 'who are you' with the new nonce)
1386
self.assertEqual(2, self.server.auth_required_errors)
1389
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1390
"""Test http digest authentication scheme"""
1392
_transport = HttpTransport_urllib
1394
def create_transport_readonly_server(self):
1395
return HTTPDigestAuthServer()
1398
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1399
TestCaseWithWebserver):
1400
"""Test proxy digest authentication scheme"""
1402
_transport = HttpTransport_urllib
1404
def create_transport_readonly_server(self):
1405
return ProxyDigestAuthServer()
added
removed
bzrlib/selftest/testhttp.py
