3
from bzrlib.tests import TestCase
4
from bzrlib.transport.http import HttpTransport, extract_auth
6
class FakeManager (object):
1
# Copyright (C) 2005, 2006 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
# TODO: What about renaming to bzrlib.tests.transport.http ?
23
from cStringIO import StringIO
38
from bzrlib.tests import (
44
from bzrlib.tests.HttpServer import (
49
from bzrlib.tests.HTTPTestUtil import (
50
BadProtocolRequestHandler,
51
BadStatusRequestHandler,
52
ForbiddenRequestHandler,
55
HTTPServerRedirecting,
56
InvalidStatusRequestHandler,
57
LimitedRangeHTTPServer,
58
NoRangeRequestHandler,
60
ProxyDigestAuthServer,
62
SingleRangeRequestHandler,
63
SingleOnlyRangeRequestHandler,
64
TestCaseWithRedirectedWebserver,
65
TestCaseWithTwoWebservers,
66
TestCaseWithWebserver,
69
from bzrlib.transport import (
71
do_catching_redirections,
75
from bzrlib.transport.http import (
80
from bzrlib.transport.http._urllib import HttpTransport_urllib
81
from bzrlib.transport.http._urllib2_wrappers import (
87
class FakeManager(object):
8
90
self.credentials = []
10
92
def add_password(self, realm, host, username, password):
11
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)
14
166
class TestHttpUrls(TestCase):
168
# TODO: This should be moved to authorization tests once they
15
171
def test_url_parsing(self):
17
173
url = extract_auth('http://example.com', f)
18
174
self.assertEquals('http://example.com', url)
19
175
self.assertEquals(0, len(f.credentials))
20
url = extract_auth('http://user:pass@www.bazaar-ng.org/bzr/bzr.dev', f)
21
self.assertEquals('http://www.bazaar-ng.org/bzr/bzr.dev', url)
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)
22
178
self.assertEquals(1, len(f.credentials))
23
self.assertEquals([None, 'www.bazaar-ng.org', 'user', 'pass'], f.credentials[0])
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.
25
194
def test_abs_url(self):
26
195
"""Construction of absolute http URLs"""
27
t = HttpTransport('http://bazaar-ng.org/bzr/bzr.dev/')
196
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
28
197
eq = self.assertEqualDiff
30
'http://bazaar-ng.org/bzr/bzr.dev')
31
eq(t.abspath('foo/bar'),
32
'http://bazaar-ng.org/bzr/bzr.dev/foo/bar')
34
'http://bazaar-ng.org/bzr/bzr.dev/.bzr')
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')
35
201
eq(t.abspath('.bzr/1//2/./3'),
36
'http://bazaar-ng.org/bzr/bzr.dev/.bzr/1/2/3')
202
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
38
204
def test_invalid_http_urls(self):
39
205
"""Trap invalid construction of urls"""
40
t = HttpTransport('http://bazaar-ng.org/bzr/bzr.dev/')
41
self.assertRaises(ValueError,
44
self.assertRaises(ValueError,
206
t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
207
self.assertRaises(errors.InvalidURL,
209
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
48
211
def test_http_root_urls(self):
49
212
"""Construction of URLs from server root"""
50
t = HttpTransport('http://bzr.ozlabs.org/')
213
t = self._transport('http://bzr.ozlabs.org/')
51
214
eq = self.assertEqualDiff
52
215
eq(t.abspath('.bzr/tree-version'),
53
216
'http://bzr.ozlabs.org/.bzr/tree-version')
218
def test_http_impl_urls(self):
219
"""There are servers which ask for particular clients to connect"""
220
server = self._server()
223
url = server.get_url()
224
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
229
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
230
"""Test http urls with urllib"""
232
_transport = HttpTransport_urllib
233
_server = HttpServer_urllib
234
_qualified_prefix = 'http+urllib'
237
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
239
"""Test http urls with pycurl"""
241
_server = HttpServer_PyCurl
242
_qualified_prefix = 'http+pycurl'
244
# TODO: This should really be moved into another pycurl
245
# specific test. When https tests will be implemented, take
246
# this one into account.
247
def test_pycurl_without_https_support(self):
248
"""Test that pycurl without SSL do not fail with a traceback.
250
For the purpose of the test, we force pycurl to ignore
251
https by supplying a fake version_info that do not
257
raise TestSkipped('pycurl not present')
258
# Now that we have pycurl imported, we can fake its version_info
259
# This was taken from a windows pycurl without SSL
261
pycurl.version_info = lambda : (2,
269
('ftp', 'gopher', 'telnet',
270
'dict', 'ldap', 'http', 'file'),
274
self.assertRaises(errors.DependencyNotPresent, self._transport,
275
'https://launchpad.net')
277
class TestHttpConnections(object):
278
"""Test the http connections.
280
This MUST be used by daughter classes that also inherit from
281
TestCaseWithWebserver.
283
We can't inherit directly from TestCaseWithWebserver or the
284
test framework will try to create an instance which cannot
285
run, its implementation being incomplete.
289
TestCaseWithWebserver.setUp(self)
290
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
291
transport=self.get_transport())
293
def test_http_has(self):
294
server = self.get_readonly_server()
295
t = self._transport(server.get_url())
296
self.assertEqual(t.has('foo/bar'), True)
297
self.assertEqual(len(server.logs), 1)
298
self.assertContainsRe(server.logs[0],
299
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
301
def test_http_has_not_found(self):
302
server = self.get_readonly_server()
303
t = self._transport(server.get_url())
304
self.assertEqual(t.has('not-found'), False)
305
self.assertContainsRe(server.logs[1],
306
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
308
def test_http_get(self):
309
server = self.get_readonly_server()
310
t = self._transport(server.get_url())
311
fp = t.get('foo/bar')
312
self.assertEqualDiff(
314
'contents of foo/bar\n')
315
self.assertEqual(len(server.logs), 1)
316
self.assertTrue(server.logs[0].find(
317
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
318
% bzrlib.__version__) > -1)
320
def test_get_smart_medium(self):
321
# For HTTP, get_smart_medium should return the transport object.
322
server = self.get_readonly_server()
323
http_transport = self._transport(server.get_url())
324
medium = http_transport.get_smart_medium()
325
self.assertIs(medium, http_transport)
327
def test_has_on_bogus_host(self):
328
# Get a free address and don't 'accept' on it, so that we
329
# can be sure there is no http handler there, but set a
330
# reasonable timeout to not slow down tests too much.
331
default_timeout = socket.getdefaulttimeout()
333
socket.setdefaulttimeout(2)
335
s.bind(('localhost', 0))
336
t = self._transport('http://%s:%s/' % s.getsockname())
337
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
339
socket.setdefaulttimeout(default_timeout)
342
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
343
"""Test http connections with urllib"""
345
_transport = HttpTransport_urllib
349
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
351
TestCaseWithWebserver):
352
"""Test http connections with pycurl"""
355
class TestHttpTransportRegistration(TestCase):
356
"""Test registrations of various http implementations"""
358
def test_http_registered(self):
359
# urlllib should always be present
360
t = get_transport('http+urllib://bzr.google.com/')
361
self.assertIsInstance(t, Transport)
362
self.assertIsInstance(t, HttpTransport_urllib)
365
class TestPost(object):
367
def _test_post_body_is_received(self, scheme):
368
server = RecordingServer(expect_body_tail='end-of-body')
370
self.addCleanup(server.tearDown)
371
url = '%s://%s:%s/' % (scheme, server.host, server.port)
373
http_transport = get_transport(url)
374
except errors.UnsupportedProtocol:
375
raise TestSkipped('%s not available' % scheme)
376
code, response = http_transport._post('abc def end-of-body')
378
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
379
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
380
# The transport should not be assuming that the server can accept
381
# chunked encoding the first time it connects, because HTTP/1.1, so we
382
# check for the literal string.
384
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
387
class TestPost_urllib(TestCase, TestPost):
388
"""TestPost for urllib implementation"""
390
_transport = HttpTransport_urllib
392
def test_post_body_is_received_urllib(self):
393
self._test_post_body_is_received('http+urllib')
396
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
397
"""TestPost for pycurl implementation"""
399
def test_post_body_is_received_pycurl(self):
400
self._test_post_body_is_received('http+pycurl')
403
class TestRangeHeader(TestCase):
404
"""Test range_header method"""
406
def check_header(self, value, ranges=[], tail=0):
407
offsets = [ (start, end - start + 1) for start, end in ranges]
408
coalesce = Transport._coalesce_offsets
409
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
410
range_header = HttpTransportBase._range_header
411
self.assertEqual(value, range_header(coalesced, tail))
413
def test_range_header_single(self):
414
self.check_header('0-9', ranges=[(0,9)])
415
self.check_header('100-109', ranges=[(100,109)])
417
def test_range_header_tail(self):
418
self.check_header('-10', tail=10)
419
self.check_header('-50', tail=50)
421
def test_range_header_multi(self):
422
self.check_header('0-9,100-200,300-5000',
423
ranges=[(0,9), (100, 200), (300,5000)])
425
def test_range_header_mixed(self):
426
self.check_header('0-9,300-5000,-50',
427
ranges=[(0,9), (300,5000)],
431
class TestWallServer(object):
432
"""Tests exceptions during the connection phase"""
434
def create_transport_readonly_server(self):
435
return HttpServer(WallRequestHandler)
437
def test_http_has(self):
438
server = self.get_readonly_server()
439
t = self._transport(server.get_url())
440
# Unfortunately httplib (see HTTPResponse._read_status
441
# for details) make no distinction between a closed
442
# socket and badly formatted status line, so we can't
443
# just test for ConnectionError, we have to test
444
# InvalidHttpResponse too.
445
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
448
def test_http_get(self):
449
server = self.get_readonly_server()
450
t = self._transport(server.get_url())
451
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
455
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
456
"""Tests "wall" server for urllib implementation"""
458
_transport = HttpTransport_urllib
461
class TestWallServer_pycurl(TestWithTransport_pycurl,
463
TestCaseWithWebserver):
464
"""Tests "wall" server for pycurl implementation"""
467
class TestBadStatusServer(object):
468
"""Tests bad status from server."""
470
def create_transport_readonly_server(self):
471
return HttpServer(BadStatusRequestHandler)
473
def test_http_has(self):
474
server = self.get_readonly_server()
475
t = self._transport(server.get_url())
476
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
478
def test_http_get(self):
479
server = self.get_readonly_server()
480
t = self._transport(server.get_url())
481
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
484
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
485
"""Tests bad status server for urllib implementation"""
487
_transport = HttpTransport_urllib
490
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
492
TestCaseWithWebserver):
493
"""Tests bad status server for pycurl implementation"""
496
class TestInvalidStatusServer(TestBadStatusServer):
497
"""Tests invalid status from server.
499
Both implementations raises the same error as for a bad status.
502
def create_transport_readonly_server(self):
503
return HttpServer(InvalidStatusRequestHandler)
506
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
507
TestCaseWithWebserver):
508
"""Tests invalid status server for urllib implementation"""
510
_transport = HttpTransport_urllib
513
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
514
TestInvalidStatusServer,
515
TestCaseWithWebserver):
516
"""Tests invalid status server for pycurl implementation"""
519
class TestBadProtocolServer(object):
520
"""Tests bad protocol from server."""
522
def create_transport_readonly_server(self):
523
return HttpServer(BadProtocolRequestHandler)
525
def test_http_has(self):
526
server = self.get_readonly_server()
527
t = self._transport(server.get_url())
528
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
530
def test_http_get(self):
531
server = self.get_readonly_server()
532
t = self._transport(server.get_url())
533
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
536
class TestBadProtocolServer_urllib(TestBadProtocolServer,
537
TestCaseWithWebserver):
538
"""Tests bad protocol server for urllib implementation"""
540
_transport = HttpTransport_urllib
542
# curl don't check the protocol version
543
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
544
# TestBadProtocolServer,
545
# TestCaseWithWebserver):
546
# """Tests bad protocol server for pycurl implementation"""
549
class TestForbiddenServer(object):
550
"""Tests forbidden server"""
552
def create_transport_readonly_server(self):
553
return HttpServer(ForbiddenRequestHandler)
555
def test_http_has(self):
556
server = self.get_readonly_server()
557
t = self._transport(server.get_url())
558
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
560
def test_http_get(self):
561
server = self.get_readonly_server()
562
t = self._transport(server.get_url())
563
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
566
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
567
"""Tests forbidden server for urllib implementation"""
569
_transport = HttpTransport_urllib
572
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
574
TestCaseWithWebserver):
575
"""Tests forbidden server for pycurl implementation"""
578
class TestRecordingServer(TestCase):
580
def test_create(self):
581
server = RecordingServer(expect_body_tail=None)
582
self.assertEqual('', server.received_bytes)
583
self.assertEqual(None, server.host)
584
self.assertEqual(None, server.port)
586
def test_setUp_and_tearDown(self):
587
server = RecordingServer(expect_body_tail=None)
590
self.assertNotEqual(None, server.host)
591
self.assertNotEqual(None, server.port)
594
self.assertEqual(None, server.host)
595
self.assertEqual(None, server.port)
597
def test_send_receive_bytes(self):
598
server = RecordingServer(expect_body_tail='c')
600
self.addCleanup(server.tearDown)
601
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
602
sock.connect((server.host, server.port))
604
self.assertEqual('HTTP/1.1 200 OK\r\n',
605
osutils.recv_all(sock, 4096))
606
self.assertEqual('abc', server.received_bytes)
609
class TestRangeRequestServer(object):
610
"""Tests readv requests against server.
612
This MUST be used by daughter classes that also inherit from
613
TestCaseWithWebserver.
615
We can't inherit directly from TestCaseWithWebserver or the
616
test framework will try to create an instance which cannot
617
run, its implementation being incomplete.
621
TestCaseWithWebserver.setUp(self)
622
self.build_tree_contents([('a', '0123456789')],)
624
def test_readv(self):
625
server = self.get_readonly_server()
626
t = self._transport(server.get_url())
627
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
628
self.assertEqual(l[0], (0, '0'))
629
self.assertEqual(l[1], (1, '1'))
630
self.assertEqual(l[2], (3, '34'))
631
self.assertEqual(l[3], (9, '9'))
633
def test_readv_out_of_order(self):
634
server = self.get_readonly_server()
635
t = self._transport(server.get_url())
636
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
637
self.assertEqual(l[0], (1, '1'))
638
self.assertEqual(l[1], (9, '9'))
639
self.assertEqual(l[2], (0, '0'))
640
self.assertEqual(l[3], (3, '34'))
642
def test_readv_invalid_ranges(self):
643
server = self.get_readonly_server()
644
t = self._transport(server.get_url())
646
# This is intentionally reading off the end of the file
647
# since we are sure that it cannot get there
648
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
649
t.readv, 'a', [(1,1), (8,10)])
651
# This is trying to seek past the end of the file, it should
652
# also raise a special error
653
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
654
t.readv, 'a', [(12,2)])
657
class TestSingleRangeRequestServer(TestRangeRequestServer):
658
"""Test readv against a server which accept only single range requests"""
660
def create_transport_readonly_server(self):
661
return HttpServer(SingleRangeRequestHandler)
664
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
665
TestCaseWithWebserver):
666
"""Tests single range requests accepting server for urllib implementation"""
668
_transport = HttpTransport_urllib
671
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
672
TestSingleRangeRequestServer,
673
TestCaseWithWebserver):
674
"""Tests single range requests accepting server for pycurl implementation"""
677
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
678
"""Test readv against a server which only accept single range requests"""
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleOnlyRangeRequestHandler)
684
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
688
_transport = HttpTransport_urllib
691
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleOnlyRangeRequestServer,
693
TestCaseWithWebserver):
694
"""Tests single range requests accepting server for pycurl implementation"""
697
class TestNoRangeRequestServer(TestRangeRequestServer):
698
"""Test readv against a server which do not accept range requests"""
700
def create_transport_readonly_server(self):
701
return HttpServer(NoRangeRequestHandler)
704
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests range requests refusing server for urllib implementation"""
708
_transport = HttpTransport_urllib
711
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestNoRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests range requests refusing server for pycurl implementation"""
717
class TestLimitedRangeRequestServer(object):
718
"""Tests readv requests against server that errors out on too much ranges.
720
This MUST be used by daughter classes that also inherit from
721
TestCaseWithWebserver.
723
We can't inherit directly from TestCaseWithWebserver or the
724
test framework will try to create an instance which cannot
725
run, its implementation being incomplete.
730
def create_transport_readonly_server(self):
731
# Requests with more range specifiers will error out
732
return LimitedRangeHTTPServer(range_limit=self.range_limit)
734
def get_transport(self):
735
return self._transport(self.get_readonly_server().get_url())
738
TestCaseWithWebserver.setUp(self)
739
# We need to manipulate ranges that correspond to real chunks in the
740
# response, so we build a content appropriately.
741
filler = ''.join(['abcdefghij' for _ in range(102)])
742
content = ''.join(['%04d' % v + filler for v in range(16)])
743
self.build_tree_contents([('a', content)],)
745
def test_few_ranges(self):
746
t = self.get_transport()
747
l = list(t.readv('a', ((0, 4), (1024, 4), )))
748
self.assertEqual(l[0], (0, '0000'))
749
self.assertEqual(l[1], (1024, '0001'))
750
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
752
def test_a_lot_of_ranges(self):
753
t = self.get_transport()
754
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
755
self.assertEqual(l[0], (0, '0000'))
756
self.assertEqual(l[1], (1024, '0001'))
757
self.assertEqual(l[2], (4096, '0004'))
758
self.assertEqual(l[3], (8192, '0008'))
759
# The server will refuse to serve the first request (too much ranges),
760
# a second request will succeeds.
761
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
764
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
765
TestCaseWithWebserver):
766
"""Tests limited range requests server for urllib implementation"""
768
_transport = HttpTransport_urllib
771
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
772
TestLimitedRangeRequestServer,
773
TestCaseWithWebserver):
774
"""Tests limited range requests server for pycurl implementation"""
778
class TestHttpProxyWhiteBox(TestCase):
779
"""Whitebox test proxy http authorization.
781
Only the urllib implementation is tested here.
791
def _install_env(self, env):
792
for name, value in env.iteritems():
793
self._old_env[name] = osutils.set_or_unset_env(name, value)
795
def _restore_env(self):
796
for name, value in self._old_env.iteritems():
797
osutils.set_or_unset_env(name, value)
799
def _proxied_request(self):
800
handler = ProxyHandler()
801
request = Request('GET','http://baz/buzzle')
802
handler.set_proxy(request, 'http')
805
def test_empty_user(self):
806
self._install_env({'http_proxy': 'http://bar.com'})
807
request = self._proxied_request()
808
self.assertFalse(request.headers.has_key('Proxy-authorization'))
810
def test_invalid_proxy(self):
811
"""A proxy env variable without scheme"""
812
self._install_env({'http_proxy': 'host:1234'})
813
self.assertRaises(errors.InvalidURL, self._proxied_request)
816
class TestProxyHttpServer(object):
817
"""Tests proxy server.
819
This MUST be used by daughter classes that also inherit from
820
TestCaseWithTwoWebservers.
822
We can't inherit directly from TestCaseWithTwoWebservers or
823
the test framework will try to create an instance which
824
cannot run, its implementation being incomplete.
826
Be aware that we do not setup a real proxy here. Instead, we
827
check that the *connection* goes through the proxy by serving
828
different content (the faked proxy server append '-proxied'
832
# FIXME: We don't have an https server available, so we don't
833
# test https connections.
836
TestCaseWithTwoWebservers.setUp(self)
837
self.build_tree_contents([('foo', 'contents of foo\n'),
838
('foo-proxied', 'proxied contents of foo\n')])
839
# Let's setup some attributes for tests
840
self.server = self.get_readonly_server()
841
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
842
self.no_proxy_host = self.proxy_address
843
# The secondary server is the proxy
844
self.proxy = self.get_secondary_server()
845
self.proxy_url = self.proxy.get_url()
848
def create_transport_secondary_server(self):
849
"""Creates an http server that will serve files with
850
'-proxied' appended to their names.
854
def _install_env(self, env):
855
for name, value in env.iteritems():
856
self._old_env[name] = osutils.set_or_unset_env(name, value)
858
def _restore_env(self):
859
for name, value in self._old_env.iteritems():
860
osutils.set_or_unset_env(name, value)
862
def proxied_in_env(self, env):
863
self._install_env(env)
864
url = self.server.get_url()
865
t = self._transport(url)
867
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
871
def not_proxied_in_env(self, env):
872
self._install_env(env)
873
url = self.server.get_url()
874
t = self._transport(url)
876
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
880
def test_http_proxy(self):
881
self.proxied_in_env({'http_proxy': self.proxy_url})
883
def test_HTTP_PROXY(self):
884
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
886
def test_all_proxy(self):
887
self.proxied_in_env({'all_proxy': self.proxy_url})
889
def test_ALL_PROXY(self):
890
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
892
def test_http_proxy_with_no_proxy(self):
893
self.not_proxied_in_env({'http_proxy': self.proxy_url,
894
'no_proxy': self.no_proxy_host})
896
def test_HTTP_PROXY_with_NO_PROXY(self):
897
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
898
'NO_PROXY': self.no_proxy_host})
900
def test_all_proxy_with_no_proxy(self):
901
self.not_proxied_in_env({'all_proxy': self.proxy_url,
902
'no_proxy': self.no_proxy_host})
904
def test_ALL_PROXY_with_NO_PROXY(self):
905
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
906
'NO_PROXY': self.no_proxy_host})
908
def test_http_proxy_without_scheme(self):
909
self.assertRaises(errors.InvalidURL,
911
{'http_proxy': self.proxy_address})
914
class TestProxyHttpServer_urllib(TestProxyHttpServer,
915
TestCaseWithTwoWebservers):
916
"""Tests proxy server for urllib implementation"""
918
_transport = HttpTransport_urllib
921
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
923
TestCaseWithTwoWebservers):
924
"""Tests proxy server for pycurl implementation"""
927
TestProxyHttpServer.setUp(self)
928
# Oh my ! pycurl does not check for the port as part of
929
# no_proxy :-( So we just test the host part
930
self.no_proxy_host = 'localhost'
932
def test_HTTP_PROXY(self):
933
# pycurl does not check HTTP_PROXY for security reasons
934
# (for use in a CGI context that we do not care
935
# about. Should we ?)
936
raise TestSkipped('pycurl does not check HTTP_PROXY '
937
'for security reasons')
939
def test_HTTP_PROXY_with_NO_PROXY(self):
940
raise TestSkipped('pycurl does not check HTTP_PROXY '
941
'for security reasons')
943
def test_http_proxy_without_scheme(self):
944
# pycurl *ignores* invalid proxy env variables. If that
945
# ever change in the future, this test will fail
946
# indicating that pycurl do not ignore anymore such
948
self.not_proxied_in_env({'http_proxy': self.proxy_address})
951
class TestRanges(object):
952
"""Test the Range header in GET methods..
954
This MUST be used by daughter classes that also inherit from
955
TestCaseWithWebserver.
957
We can't inherit directly from TestCaseWithWebserver or the
958
test framework will try to create an instance which cannot
959
run, its implementation being incomplete.
963
TestCaseWithWebserver.setUp(self)
964
self.build_tree_contents([('a', '0123456789')],)
965
server = self.get_readonly_server()
966
self.transport = self._transport(server.get_url())
968
def _file_contents(self, relpath, ranges):
969
offsets = [ (start, end - start + 1) for start, end in ranges]
970
coalesce = self.transport._coalesce_offsets
971
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
972
code, data = self.transport._get(relpath, coalesced)
973
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
974
for start, end in ranges:
976
yield data.read(end - start + 1)
978
def _file_tail(self, relpath, tail_amount):
979
code, data = self.transport._get(relpath, [], tail_amount)
980
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
981
data.seek(-tail_amount + 1, 2)
982
return data.read(tail_amount)
984
def test_range_header(self):
986
map(self.assertEqual,['0', '234'],
987
list(self._file_contents('a', [(0,0), (2,4)])),)
989
self.assertEqual('789', self._file_tail('a', 3))
990
# Syntactically invalid range
991
self.assertListRaises(errors.InvalidRange,
992
self._file_contents, 'a', [(4, 3)])
993
# Semantically invalid range
994
self.assertListRaises(errors.InvalidRange,
995
self._file_contents, 'a', [(42, 128)])
998
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
999
"""Test the Range header in GET methods for urllib implementation"""
1001
_transport = HttpTransport_urllib
1004
class TestRanges_pycurl(TestWithTransport_pycurl,
1006
TestCaseWithWebserver):
1007
"""Test the Range header in GET methods for pycurl implementation"""
1010
class TestHTTPRedirections(object):
1011
"""Test redirection between http servers.
1013
This MUST be used by daughter classes that also inherit from
1014
TestCaseWithRedirectedWebserver.
1016
We can't inherit directly from TestCaseWithTwoWebservers or the
1017
test framework will try to create an instance which cannot
1018
run, its implementation being incomplete.
1021
def create_transport_secondary_server(self):
1022
"""Create the secondary server redirecting to the primary server"""
1023
new = self.get_readonly_server()
1025
redirecting = HTTPServerRedirecting()
1026
redirecting.redirect_to(new.host, new.port)
1030
super(TestHTTPRedirections, self).setUp()
1031
self.build_tree_contents([('a', '0123456789'),
1033
'# Bazaar revision bundle v0.9\n#\n')
1036
self.old_transport = self._transport(self.old_server.get_url())
1038
def test_redirected(self):
1039
self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
1040
t = self._transport(self.new_server.get_url())
1041
self.assertEqual('0123456789', t.get('a').read())
1043
def test_read_redirected_bundle_from_url(self):
1044
from bzrlib.bundle import read_bundle_from_url
1045
url = self.old_transport.abspath('bundle')
1046
bundle = read_bundle_from_url(url)
1047
# If read_bundle_from_url was successful we get an empty bundle
1048
self.assertEqual([], bundle.revisions)
1051
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1052
TestCaseWithRedirectedWebserver):
1053
"""Tests redirections for urllib implementation"""
1055
_transport = HttpTransport_urllib
1059
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1060
TestHTTPRedirections,
1061
TestCaseWithRedirectedWebserver):
1062
"""Tests redirections for pycurl implementation"""
1065
class RedirectedRequest(Request):
1066
"""Request following redirections"""
1068
init_orig = Request.__init__
1070
def __init__(self, method, url, *args, **kwargs):
1071
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1072
self.follow_redirections = True
1075
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1076
"""Test redirections provided by urllib.
1078
http implementations do not redirect silently anymore (they
1079
do not redirect at all in fact). The mechanism is still in
1080
place at the _urllib2_wrappers.Request level and these tests
1083
For the pycurl implementation
1084
the redirection have been deleted as we may deprecate pycurl
1085
and I have no place to keep a working implementation.
1089
_transport = HttpTransport_urllib
1092
super(TestHTTPSilentRedirections_urllib, self).setUp()
1093
self.setup_redirected_request()
1094
self.addCleanup(self.cleanup_redirected_request)
1095
self.build_tree_contents([('a','a'),
1097
('1/a', 'redirected once'),
1099
('2/a', 'redirected twice'),
1101
('3/a', 'redirected thrice'),
1103
('4/a', 'redirected 4 times'),
1105
('5/a', 'redirected 5 times'),
1108
self.old_transport = self._transport(self.old_server.get_url())
1110
def setup_redirected_request(self):
1111
self.original_class = _urllib2_wrappers.Request
1112
_urllib2_wrappers.Request = RedirectedRequest
1114
def cleanup_redirected_request(self):
1115
_urllib2_wrappers.Request = self.original_class
1117
def create_transport_secondary_server(self):
1118
"""Create the secondary server, redirections are defined in the tests"""
1119
return HTTPServerRedirecting()
1121
def test_one_redirection(self):
1122
t = self.old_transport
1124
req = RedirectedRequest('GET', t.abspath('a'))
1125
req.follow_redirections = True
1126
new_prefix = 'http://%s:%s' % (self.new_server.host,
1127
self.new_server.port)
1128
self.old_server.redirections = \
1129
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1130
self.assertEquals('redirected once',t._perform(req).read())
1132
def test_five_redirections(self):
1133
t = self.old_transport
1135
req = RedirectedRequest('GET', t.abspath('a'))
1136
req.follow_redirections = True
1137
old_prefix = 'http://%s:%s' % (self.old_server.host,
1138
self.old_server.port)
1139
new_prefix = 'http://%s:%s' % (self.new_server.host,
1140
self.new_server.port)
1141
self.old_server.redirections = \
1142
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1143
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1144
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1145
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1146
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1148
self.assertEquals('redirected 5 times',t._perform(req).read())
1151
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1152
"""Test transport.do_catching_redirections.
1154
We arbitrarily choose to use urllib transports
1157
_transport = HttpTransport_urllib
1160
super(TestDoCatchRedirections, self).setUp()
1161
self.build_tree_contents([('a', '0123456789'),],)
1163
self.old_transport = self._transport(self.old_server.get_url())
1165
def get_a(self, transport):
1166
return transport.get('a')
1168
def test_no_redirection(self):
1169
t = self._transport(self.new_server.get_url())
1171
# We use None for redirected so that we fail if redirected
1172
self.assertEquals('0123456789',
1173
do_catching_redirections(self.get_a, t, None).read())
1175
def test_one_redirection(self):
1176
self.redirections = 0
1178
def redirected(transport, exception, redirection_notice):
1179
self.redirections += 1
1180
dir, file = urlutils.split(exception.target)
1181
return self._transport(dir)
1183
self.assertEquals('0123456789',
1184
do_catching_redirections(self.get_a,
1188
self.assertEquals(1, self.redirections)
1190
def test_redirection_loop(self):
1192
def redirected(transport, exception, redirection_notice):
1193
# By using the redirected url as a base dir for the
1194
# *old* transport, we create a loop: a => a/a =>
1196
return self.old_transport.clone(exception.target)
1198
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1199
self.get_a, self.old_transport, redirected)
1202
class TestAuth(object):
1203
"""Test some authentication scheme specified by daughter class.
1205
This MUST be used by daughter classes that also inherit from
1206
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1209
_password_prompt_prefix = ''
1212
"""Set up the test environment
1214
Daughter classes should set up their own environment
1215
(including self.server) and explicitely call this
1216
method. This is needed because we want to reuse the same
1217
tests for proxy and no-proxy accesses which have
1218
different ways of setting self.server.
1220
self.build_tree_contents([('a', 'contents of a\n'),
1221
('b', 'contents of b\n'),])
1223
def get_user_url(self, user=None, password=None):
1224
"""Build an url embedding user and password"""
1225
url = '%s://' % self.server._url_protocol
1226
if user is not None:
1228
if password is not None:
1229
url += ':' + password
1231
url += '%s:%s/' % (self.server.host, self.server.port)
1234
def test_no_user(self):
1235
self.server.add_user('joe', 'foo')
1236
t = self.get_user_transport()
1237
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1238
# Only one 'Authentication Required' error should occur
1239
self.assertEqual(1, self.server.auth_required_errors)
1241
def test_empty_pass(self):
1242
self.server.add_user('joe', '')
1243
t = self.get_user_transport('joe', '')
1244
self.assertEqual('contents of a\n', t.get('a').read())
1245
# Only one 'Authentication Required' error should occur
1246
self.assertEqual(1, self.server.auth_required_errors)
1248
def test_user_pass(self):
1249
self.server.add_user('joe', 'foo')
1250
t = self.get_user_transport('joe', 'foo')
1251
self.assertEqual('contents of a\n', t.get('a').read())
1252
# Only one 'Authentication Required' error should occur
1253
self.assertEqual(1, self.server.auth_required_errors)
1255
def test_unknown_user(self):
1256
self.server.add_user('joe', 'foo')
1257
t = self.get_user_transport('bill', 'foo')
1258
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1259
# Two 'Authentication Required' errors should occur (the
1260
# initial 'who are you' and 'I don't know you, who are
1262
self.assertEqual(2, self.server.auth_required_errors)
1264
def test_wrong_pass(self):
1265
self.server.add_user('joe', 'foo')
1266
t = self.get_user_transport('joe', 'bar')
1267
self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1268
# Two 'Authentication Required' errors should occur (the
1269
# initial 'who are you' and 'this is not you, who are you')
1270
self.assertEqual(2, self.server.auth_required_errors)
1272
def test_prompt_for_password(self):
1273
self.server.add_user('joe', 'foo')
1274
t = self.get_user_transport('joe', None)
1275
stdout = StringIOWrapper()
1276
ui.ui_factory = TestUIFactory(stdin='foo\n', stdout=stdout)
1277
self.assertEqual('contents of a\n',t.get('a').read())
1278
# stdin should be empty
1279
self.assertEqual('', ui.ui_factory.stdin.readline())
1280
self._check_password_prompt(t._unqualified_scheme, 'joe',
1282
# And we shouldn't prompt again for a different request
1283
# against the same transport.
1284
self.assertEqual('contents of b\n',t.get('b').read())
1286
# And neither against a clone
1287
self.assertEqual('contents of b\n',t2.get('b').read())
1288
# Only one 'Authentication Required' error should occur
1289
self.assertEqual(1, self.server.auth_required_errors)
1291
def _check_password_prompt(self, scheme, user, actual_prompt):
1292
expected_prompt = (self._password_prompt_prefix
1293
+ ("%s %s@%s:%d, Realm: '%s' password: "
1295
user, self.server.host, self.server.port,
1296
self.server.auth_realm)))
1297
self.assertEquals(expected_prompt, actual_prompt)
1299
def test_no_prompt_for_password_when_using_auth_config(self):
1302
stdin_content = 'bar\n' # Not the right password
1303
self.server.add_user(user, password)
1304
t = self.get_user_transport(user, None)
1305
ui.ui_factory = TestUIFactory(stdin=stdin_content,
1306
stdout=StringIOWrapper())
1307
# Create a minimal config file with the right password
1308
conf = config.AuthenticationConfig()
1309
conf._get_config().update(
1310
{'httptest': {'scheme': 'http', 'port': self.server.port,
1311
'user': user, 'password': password}})
1313
# Issue a request to the server to connect
1314
self.assertEqual('contents of a\n',t.get('a').read())
1315
# stdin should have been left untouched
1316
self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
1317
# Only one 'Authentication Required' error should occur
1318
self.assertEqual(1, self.server.auth_required_errors)
1322
class TestHTTPAuth(TestAuth):
1323
"""Test HTTP authentication schemes.
1325
Daughter classes MUST inherit from TestCaseWithWebserver too.
1328
_auth_header = 'Authorization'
1331
TestCaseWithWebserver.setUp(self)
1332
self.server = self.get_readonly_server()
1333
TestAuth.setUp(self)
1335
def get_user_transport(self, user=None, password=None):
1336
return self._transport(self.get_user_url(user, password))
1339
class TestProxyAuth(TestAuth):
1340
"""Test proxy authentication schemes.
1342
Daughter classes MUST also inherit from TestCaseWithWebserver.
1344
_auth_header = 'Proxy-authorization'
1345
_password_prompt_prefix = 'Proxy '
1349
TestCaseWithWebserver.setUp(self)
1350
self.server = self.get_readonly_server()
1352
self.addCleanup(self._restore_env)
1353
TestAuth.setUp(self)
1354
# Override the contents to avoid false positives
1355
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1356
('b', 'not proxied contents of b\n'),
1357
('a-proxied', 'contents of a\n'),
1358
('b-proxied', 'contents of b\n'),
1361
def get_user_transport(self, user=None, password=None):
1362
self._install_env({'all_proxy': self.get_user_url(user, password)})
1363
return self._transport(self.server.get_url())
1365
def _install_env(self, env):
1366
for name, value in env.iteritems():
1367
self._old_env[name] = osutils.set_or_unset_env(name, value)
1369
def _restore_env(self):
1370
for name, value in self._old_env.iteritems():
1371
osutils.set_or_unset_env(name, value)
1374
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1375
"""Test http basic authentication scheme"""
1377
_transport = HttpTransport_urllib
1379
def create_transport_readonly_server(self):
1380
return HTTPBasicAuthServer()
1383
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1384
"""Test proxy basic authentication scheme"""
1386
_transport = HttpTransport_urllib
1388
def create_transport_readonly_server(self):
1389
return ProxyBasicAuthServer()
1392
class TestDigestAuth(object):
1393
"""Digest Authentication specific tests"""
1395
def test_changing_nonce(self):
1396
self.server.add_user('joe', 'foo')
1397
t = self.get_user_transport('joe', 'foo')
1398
self.assertEqual('contents of a\n', t.get('a').read())
1399
self.assertEqual('contents of b\n', t.get('b').read())
1400
# Only one 'Authentication Required' error should have
1402
self.assertEqual(1, self.server.auth_required_errors)
1403
# The server invalidates the current nonce
1404
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1405
self.assertEqual('contents of a\n', t.get('a').read())
1406
# Two 'Authentication Required' errors should occur (the
1407
# initial 'who are you' and a second 'who are you' with the new nonce)
1408
self.assertEqual(2, self.server.auth_required_errors)
1411
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1412
"""Test http digest authentication scheme"""
1414
_transport = HttpTransport_urllib
1416
def create_transport_readonly_server(self):
1417
return HTTPDigestAuthServer()
1420
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1421
TestCaseWithWebserver):
1422
"""Test proxy digest authentication scheme"""
1424
_transport = HttpTransport_urllib
1426
def create_transport_readonly_server(self):
1427
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
