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 ?
29
from bzrlib import errors
30
from bzrlib import osutils
31
from bzrlib.tests import (
35
from bzrlib.tests.HttpServer import (
40
from bzrlib.tests.HTTPTestUtil import (
41
BadProtocolRequestHandler,
42
BadStatusRequestHandler,
43
FakeProxyRequestHandler,
44
ForbiddenRequestHandler,
45
HTTPServerRedirecting,
46
InvalidStatusRequestHandler,
47
NoRangeRequestHandler,
48
SingleRangeRequestHandler,
49
TestCaseWithTwoWebservers,
50
TestCaseWithWebserver,
53
from bzrlib.transport import (
57
from bzrlib.transport.http import (
61
from bzrlib.transport.http._urllib import HttpTransport_urllib
64
class FakeManager(object):
69
def add_password(self, realm, host, username, password):
70
self.credentials.append([realm, host, username, password])
73
class RecordingServer(object):
74
"""A fake HTTP server.
76
It records the bytes sent to it, and replies with a 200.
79
def __init__(self, expect_body_tail=None):
82
:type expect_body_tail: str
83
:param expect_body_tail: a reply won't be sent until this string is
86
self._expect_body_tail = expect_body_tail
89
self.received_bytes = ''
92
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
93
self._sock.bind(('127.0.0.1', 0))
94
self.host, self.port = self._sock.getsockname()
95
self._ready = threading.Event()
96
self._thread = threading.Thread(target=self._accept_read_and_reply)
97
self._thread.setDaemon(True)
101
def _accept_read_and_reply(self):
104
self._sock.settimeout(5)
106
conn, address = self._sock.accept()
107
# On win32, the accepted connection will be non-blocking to start
108
# with because we're using settimeout.
109
conn.setblocking(True)
110
while not self.received_bytes.endswith(self._expect_body_tail):
111
self.received_bytes += conn.recv(4096)
112
conn.sendall('HTTP/1.1 200 OK\r\n')
113
except socket.timeout:
114
# Make sure the client isn't stuck waiting for us to e.g. accept.
117
# The client may have already closed the socket.
124
# We might have already closed it. We don't care.
130
class TestHttpUrls(TestCase):
132
# FIXME: Some of these tests should be done for both
135
def test_url_parsing(self):
137
url = extract_auth('http://example.com', f)
138
self.assertEquals('http://example.com', url)
139
self.assertEquals(0, len(f.credentials))
140
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
141
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
142
self.assertEquals(1, len(f.credentials))
143
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
146
def test_abs_url(self):
147
"""Construction of absolute http URLs"""
148
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
149
eq = self.assertEqualDiff
151
'http://bazaar-vcs.org/bzr/bzr.dev')
152
eq(t.abspath('foo/bar'),
153
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
154
eq(t.abspath('.bzr'),
155
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
156
eq(t.abspath('.bzr/1//2/./3'),
157
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
159
def test_invalid_http_urls(self):
160
"""Trap invalid construction of urls"""
161
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
162
self.assertRaises(ValueError,
165
t = HttpTransport_urllib('http://http://bazaar-vcs.org/bzr/bzr.dev/')
166
self.assertRaises(errors.InvalidURL, t.has, 'foo/bar')
168
def test_http_root_urls(self):
169
"""Construction of URLs from server root"""
170
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
171
eq = self.assertEqualDiff
172
eq(t.abspath('.bzr/tree-version'),
173
'http://bzr.ozlabs.org/.bzr/tree-version')
175
def test_http_impl_urls(self):
176
"""There are servers which ask for particular clients to connect"""
177
server = HttpServer_PyCurl()
180
url = server.get_url()
181
self.assertTrue(url.startswith('http+pycurl://'))
186
class TestHttpConnections(object):
187
"""Test the http connections.
189
This MUST be used by daughter classes that also inherit from
190
TestCaseWithWebserver.
192
We can't inherit directly from TestCaseWithWebserver or the
193
test framework will try to create an instance which cannot
194
run, its implementation being incomplete.
198
TestCaseWithWebserver.setUp(self)
199
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
200
transport=self.get_transport())
202
def test_http_has(self):
203
server = self.get_readonly_server()
204
t = self._transport(server.get_url())
205
self.assertEqual(t.has('foo/bar'), True)
206
self.assertEqual(len(server.logs), 1)
207
self.assertContainsRe(server.logs[0],
208
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
210
def test_http_has_not_found(self):
211
server = self.get_readonly_server()
212
t = self._transport(server.get_url())
213
self.assertEqual(t.has('not-found'), False)
214
self.assertContainsRe(server.logs[1],
215
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
217
def test_http_get(self):
218
server = self.get_readonly_server()
219
t = self._transport(server.get_url())
220
fp = t.get('foo/bar')
221
self.assertEqualDiff(
223
'contents of foo/bar\n')
224
self.assertEqual(len(server.logs), 1)
225
self.assertTrue(server.logs[0].find(
226
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
227
% bzrlib.__version__) > -1)
229
def test_get_smart_medium(self):
230
# For HTTP, get_smart_medium should return the transport object.
231
server = self.get_readonly_server()
232
http_transport = self._transport(server.get_url())
233
medium = http_transport.get_smart_medium()
234
self.assertIs(medium, http_transport)
236
def test_has_on_bogus_host(self):
237
# Get a free address and don't 'accept' on it, so that we
238
# can be sure there is no http handler there, but set a
239
# reasonable timeout to not slow down tests too much.
240
default_timeout = socket.getdefaulttimeout()
242
socket.setdefaulttimeout(2)
244
s.bind(('localhost', 0))
245
t = self._transport('http://%s:%s/' % s.getsockname())
246
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
248
socket.setdefaulttimeout(default_timeout)
251
class TestWithTransport_pycurl(object):
252
"""Test case to inherit from if pycurl is present"""
254
def _get_pycurl_maybe(self):
256
from bzrlib.transport.http._pycurl import PyCurlTransport
257
return PyCurlTransport
258
except errors.DependencyNotPresent:
259
raise TestSkipped('pycurl not present')
261
_transport = property(_get_pycurl_maybe)
264
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
265
"""Test http connections with urllib"""
267
_transport = HttpTransport_urllib
271
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
273
TestCaseWithWebserver):
274
"""Test http connections with pycurl"""
277
class TestHttpTransportRegistration(TestCase):
278
"""Test registrations of various http implementations"""
280
def test_http_registered(self):
281
# urlllib should always be present
282
t = get_transport('http+urllib://bzr.google.com/')
283
self.assertIsInstance(t, Transport)
284
self.assertIsInstance(t, HttpTransport_urllib)
287
class TestOffsets(TestCase):
288
"""Test offsets_to_ranges method"""
290
def test_offsets_to_ranges_simple(self):
291
to_range = HttpTransportBase.offsets_to_ranges
292
ranges = to_range([(10, 1)])
293
self.assertEqual([[10, 10]], ranges)
295
ranges = to_range([(0, 1), (1, 1)])
296
self.assertEqual([[0, 1]], ranges)
298
ranges = to_range([(1, 1), (0, 1)])
299
self.assertEqual([[0, 1]], ranges)
301
def test_offset_to_ranges_overlapped(self):
302
to_range = HttpTransportBase.offsets_to_ranges
304
ranges = to_range([(10, 1), (20, 2), (22, 5)])
305
self.assertEqual([[10, 10], [20, 26]], ranges)
307
ranges = to_range([(10, 1), (11, 2), (22, 5)])
308
self.assertEqual([[10, 12], [22, 26]], ranges)
311
class TestPost(object):
313
def _test_post_body_is_received(self, scheme):
314
server = RecordingServer(expect_body_tail='end-of-body')
316
self.addCleanup(server.tearDown)
317
url = '%s://%s:%s/' % (scheme, server.host, server.port)
319
http_transport = get_transport(url)
320
except errors.UnsupportedProtocol:
321
raise TestSkipped('%s not available' % scheme)
322
code, response = http_transport._post('abc def end-of-body')
324
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
325
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
326
# The transport should not be assuming that the server can accept
327
# chunked encoding the first time it connects, because HTTP/1.1, so we
328
# check for the literal string.
330
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
333
class TestPost_urllib(TestCase, TestPost):
334
"""TestPost for urllib implementation"""
336
_transport = HttpTransport_urllib
338
def test_post_body_is_received_urllib(self):
339
self._test_post_body_is_received('http+urllib')
342
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
343
"""TestPost for pycurl implementation"""
345
def test_post_body_is_received_pycurl(self):
346
self._test_post_body_is_received('http+pycurl')
349
class TestRangeHeader(TestCase):
350
"""Test range_header method"""
352
def check_header(self, value, ranges=[], tail=0):
353
range_header = HttpTransportBase.range_header
354
self.assertEqual(value, range_header(ranges, tail))
356
def test_range_header_single(self):
357
self.check_header('0-9', ranges=[[0,9]])
358
self.check_header('100-109', ranges=[[100,109]])
360
def test_range_header_tail(self):
361
self.check_header('-10', tail=10)
362
self.check_header('-50', tail=50)
364
def test_range_header_multi(self):
365
self.check_header('0-9,100-200,300-5000',
366
ranges=[(0,9), (100, 200), (300,5000)])
368
def test_range_header_mixed(self):
369
self.check_header('0-9,300-5000,-50',
370
ranges=[(0,9), (300,5000)],
374
class TestWallServer(object):
375
"""Tests exceptions during the connection phase"""
377
def create_transport_readonly_server(self):
378
return HttpServer(WallRequestHandler)
380
def test_http_has(self):
381
server = self.get_readonly_server()
382
t = self._transport(server.get_url())
383
# Unfortunately httplib (see HTTPResponse._read_status
384
# for details) make no distinction between a closed
385
# socket and badly formatted status line, so we can't
386
# just test for ConnectionError, we have to test
387
# InvalidHttpResponse too.
388
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
391
def test_http_get(self):
392
server = self.get_readonly_server()
393
t = self._transport(server.get_url())
394
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
398
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
399
"""Tests "wall" server for urllib implementation"""
401
_transport = HttpTransport_urllib
404
class TestWallServer_pycurl(TestWithTransport_pycurl,
406
TestCaseWithWebserver):
407
"""Tests "wall" server for pycurl implementation"""
410
class TestBadStatusServer(object):
411
"""Tests bad status from server."""
413
def create_transport_readonly_server(self):
414
return HttpServer(BadStatusRequestHandler)
416
def test_http_has(self):
417
server = self.get_readonly_server()
418
t = self._transport(server.get_url())
419
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
421
def test_http_get(self):
422
server = self.get_readonly_server()
423
t = self._transport(server.get_url())
424
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
427
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
428
"""Tests bad status server for urllib implementation"""
430
_transport = HttpTransport_urllib
433
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
435
TestCaseWithWebserver):
436
"""Tests bad status server for pycurl implementation"""
439
class TestInvalidStatusServer(TestBadStatusServer):
440
"""Tests invalid status from server.
442
Both implementations raises the same error as for a bad status.
445
def create_transport_readonly_server(self):
446
return HttpServer(InvalidStatusRequestHandler)
449
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
450
TestCaseWithWebserver):
451
"""Tests invalid status server for urllib implementation"""
453
_transport = HttpTransport_urllib
456
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
457
TestInvalidStatusServer,
458
TestCaseWithWebserver):
459
"""Tests invalid status server for pycurl implementation"""
462
class TestBadProtocolServer(object):
463
"""Tests bad protocol from server."""
465
def create_transport_readonly_server(self):
466
return HttpServer(BadProtocolRequestHandler)
468
def test_http_has(self):
469
server = self.get_readonly_server()
470
t = self._transport(server.get_url())
471
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
473
def test_http_get(self):
474
server = self.get_readonly_server()
475
t = self._transport(server.get_url())
476
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
479
class TestBadProtocolServer_urllib(TestBadProtocolServer,
480
TestCaseWithWebserver):
481
"""Tests bad protocol server for urllib implementation"""
483
_transport = HttpTransport_urllib
485
# curl don't check the protocol version
486
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
487
# TestBadProtocolServer,
488
# TestCaseWithWebserver):
489
# """Tests bad protocol server for pycurl implementation"""
492
class TestForbiddenServer(object):
493
"""Tests forbidden server"""
495
def create_transport_readonly_server(self):
496
return HttpServer(ForbiddenRequestHandler)
498
def test_http_has(self):
499
server = self.get_readonly_server()
500
t = self._transport(server.get_url())
501
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
503
def test_http_get(self):
504
server = self.get_readonly_server()
505
t = self._transport(server.get_url())
506
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
509
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
510
"""Tests forbidden server for urllib implementation"""
512
_transport = HttpTransport_urllib
515
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
517
TestCaseWithWebserver):
518
"""Tests forbidden server for pycurl implementation"""
521
class TestRecordingServer(TestCase):
523
def test_create(self):
524
server = RecordingServer(expect_body_tail=None)
525
self.assertEqual('', server.received_bytes)
526
self.assertEqual(None, server.host)
527
self.assertEqual(None, server.port)
529
def test_setUp_and_tearDown(self):
530
server = RecordingServer(expect_body_tail=None)
533
self.assertNotEqual(None, server.host)
534
self.assertNotEqual(None, server.port)
537
self.assertEqual(None, server.host)
538
self.assertEqual(None, server.port)
540
def test_send_receive_bytes(self):
541
server = RecordingServer(expect_body_tail='c')
543
self.addCleanup(server.tearDown)
544
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
545
sock.connect((server.host, server.port))
547
self.assertEqual('HTTP/1.1 200 OK\r\n',
548
osutils.recv_all(sock, 4096))
549
self.assertEqual('abc', server.received_bytes)
552
class TestRangeRequestServer(object):
553
"""Tests readv requests against server.
555
This MUST be used by daughter classes that also inherit from
556
TestCaseWithWebserver.
558
We can't inherit directly from TestCaseWithWebserver or the
559
test framework will try to create an instance which cannot
560
run, its implementation being incomplete.
564
TestCaseWithWebserver.setUp(self)
565
self.build_tree_contents([('a', '0123456789')],)
567
def test_readv(self):
568
server = self.get_readonly_server()
569
t = self._transport(server.get_url())
570
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
571
self.assertEqual(l[0], (0, '0'))
572
self.assertEqual(l[1], (1, '1'))
573
self.assertEqual(l[2], (3, '34'))
574
self.assertEqual(l[3], (9, '9'))
576
def test_readv_out_of_order(self):
577
server = self.get_readonly_server()
578
t = self._transport(server.get_url())
579
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
580
self.assertEqual(l[0], (1, '1'))
581
self.assertEqual(l[1], (9, '9'))
582
self.assertEqual(l[2], (0, '0'))
583
self.assertEqual(l[3], (3, '34'))
585
def test_readv_invalid_ranges(self):
586
server = self.get_readonly_server()
587
t = self._transport(server.get_url())
589
# This is intentionally reading off the end of the file
590
# since we are sure that it cannot get there
591
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
592
t.readv, 'a', [(1,1), (8,10)])
594
# This is trying to seek past the end of the file, it should
595
# also raise a special error
596
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
597
t.readv, 'a', [(12,2)])
600
class TestSingleRangeRequestServer(TestRangeRequestServer):
601
"""Test readv against a server which accept only single range requests"""
603
def create_transport_readonly_server(self):
604
return HttpServer(SingleRangeRequestHandler)
607
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
608
TestCaseWithWebserver):
609
"""Tests single range requests accepting server for urllib implementation"""
611
_transport = HttpTransport_urllib
614
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
615
TestSingleRangeRequestServer,
616
TestCaseWithWebserver):
617
"""Tests single range requests accepting server for pycurl implementation"""
620
class TestNoRangeRequestServer(TestRangeRequestServer):
621
"""Test readv against a server which do not accept range requests"""
623
def create_transport_readonly_server(self):
624
return HttpServer(NoRangeRequestHandler)
627
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
628
TestCaseWithWebserver):
629
"""Tests range requests refusing server for urllib implementation"""
631
_transport = HttpTransport_urllib
634
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
635
TestNoRangeRequestServer,
636
TestCaseWithWebserver):
637
"""Tests range requests refusing server for pycurl implementation"""
640
class TestProxyHttpServer(object):
641
"""Tests proxy server.
643
This MUST be used by daughter classes that also inherit from
644
TestCaseWithTwoWebservers.
646
We can't inherit directly from TestCaseWithTwoWebservers or
647
the test framework will try to create an instance which
648
cannot run, its implementation being incomplete.
650
Be aware that we do not setup a real proxy here. Instead, we
651
check that the *connection* goes through the proxy by serving
652
different content (the faked proxy server append '-proxied'
656
# FIXME: We don't have an https server available, so we don't
657
# test https connections.
660
TestCaseWithTwoWebservers.setUp(self)
661
self.build_tree_contents([('foo', 'contents of foo\n'),
662
('foo-proxied', 'proxied contents of foo\n')])
663
# Let's setup some attributes for tests
664
self.server = self.get_readonly_server()
665
self.no_proxy_host = 'localhost:%d' % self.server.port
666
# The secondary server is the proxy
667
self.proxy = self.get_secondary_server()
668
self.proxy_url = self.proxy.get_url()
671
def create_transport_secondary_server(self):
672
"""Creates an http server that will serve files with
673
'-proxied' appended to their names.
675
return HttpServer(FakeProxyRequestHandler)
677
def _set_and_capture_env_var(self, name, new_value):
678
"""Set an environment variable, and reset it when finished."""
679
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
681
def _install_env(self, env):
682
for name, value in env.iteritems():
683
self._set_and_capture_env_var(name, value)
685
def _restore_env(self):
686
for name, value in self._old_env.iteritems():
687
osutils.set_or_unset_env(name, value)
689
def proxied_in_env(self, env):
690
self._install_env(env)
691
url = self.server.get_url()
692
t = self._transport(url)
694
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
698
def not_proxied_in_env(self, env):
699
self._install_env(env)
700
url = self.server.get_url()
701
t = self._transport(url)
703
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
707
def test_http_proxy(self):
708
self.proxied_in_env({'http_proxy': self.proxy_url})
710
def test_HTTP_PROXY(self):
711
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
713
def test_all_proxy(self):
714
self.proxied_in_env({'all_proxy': self.proxy_url})
716
def test_ALL_PROXY(self):
717
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
719
def test_http_proxy_with_no_proxy(self):
720
self.not_proxied_in_env({'http_proxy': self.proxy_url,
721
'no_proxy': self.no_proxy_host})
723
def test_HTTP_PROXY_with_NO_PROXY(self):
724
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
725
'NO_PROXY': self.no_proxy_host})
727
def test_all_proxy_with_no_proxy(self):
728
self.not_proxied_in_env({'all_proxy': self.proxy_url,
729
'no_proxy': self.no_proxy_host})
731
def test_ALL_PROXY_with_NO_PROXY(self):
732
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
733
'NO_PROXY': self.no_proxy_host})
736
class TestProxyHttpServer_urllib(TestProxyHttpServer,
737
TestCaseWithTwoWebservers):
738
"""Tests proxy server for urllib implementation"""
740
_transport = HttpTransport_urllib
743
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
745
TestCaseWithTwoWebservers):
746
"""Tests proxy server for pycurl implementation"""
749
TestProxyHttpServer.setUp(self)
750
# Oh my ! pycurl does not check for the port as part of
751
# no_proxy :-( So we just test the host part
752
self.no_proxy_host = 'localhost'
754
def test_HTTP_PROXY(self):
755
# pycurl do not check HTTP_PROXY for security reasons
756
# (for use in a CGI context that we do not care
757
# about. Should we ?)
760
def test_HTTP_PROXY_with_NO_PROXY(self):
764
class TestRanges(object):
765
"""Test the Range header in GET methods..
767
This MUST be used by daughter classes that also inherit from
768
TestCaseWithWebserver.
770
We can't inherit directly from TestCaseWithWebserver or the
771
test framework will try to create an instance which cannot
772
run, its implementation being incomplete.
776
TestCaseWithWebserver.setUp(self)
777
self.build_tree_contents([('a', '0123456789')],)
778
server = self.get_readonly_server()
779
self.transport = self._transport(server.get_url())
781
def _file_contents(self, relpath, ranges, tail_amount=0):
782
code, data = self.transport._get(relpath, ranges)
783
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
784
for start, end in ranges:
786
yield data.read(end - start + 1)
788
def _file_tail(self, relpath, tail_amount):
789
code, data = self.transport._get(relpath, [], tail_amount)
790
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
791
data.seek(-tail_amount + 1, 2)
792
return data.read(tail_amount)
794
def test_range_header(self):
796
map(self.assertEqual,['0', '234'],
797
list(self._file_contents('a', [(0,0), (2,4)])),)
799
self.assertEqual('789', self._file_tail('a', 3))
800
# Syntactically invalid range
801
self.assertRaises(errors.InvalidRange,
802
self.transport._get, 'a', [(4, 3)])
803
# Semantically invalid range
804
self.assertRaises(errors.InvalidRange,
805
self.transport._get, 'a', [(42, 128)])
808
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
809
"""Test the Range header in GET methods for urllib implementation"""
811
_transport = HttpTransport_urllib
814
class TestRanges_pycurl(TestWithTransport_pycurl,
816
TestCaseWithWebserver):
817
"""Test the Range header in GET methods for pycurl implementation"""
820
class TestRedirections(object):
821
"""Test redirection from the 'old' server to the 'new' server.
823
This MUST be used by daughter classes that also inherit from
824
TestCaseWithTwoWebservers.
826
We can't inherit directly from TestCaseWithTwoWebservers or the
827
test framework will try to create an instance which cannot
828
run, its implementation being incomplete.
831
def create_transport_secondary_server(self):
832
"""Create the secondary server redirecting to the primary server"""
833
new = self.get_new_server()
835
return HTTPServerRedirecting(new.host, new.port)
837
def get_old_server(self):
838
"""The redirected server"""
839
return self.get_secondary_server()
841
def get_new_server(self):
842
"""The redirected to server"""
843
return self.get_readonly_server()
846
super(TestRedirections, self).setUp()
847
new = self.get_new_server()
848
old = self.get_old_server()
850
self.build_tree_contents([('a', '0123456789')],)
852
self.new_transport = self._transport(new.get_url())
853
self.old_transport = self._transport(old.get_url())
855
def test_redirected(self):
856
old = self.old_transport
857
self.assertRaises(errors.RedirectRequested, old.get, 'a')
859
def test_redirection_loop(self):
863
class TestRedirections_urllib(TestRedirections,
864
TestCaseWithTwoWebservers):
865
"""Tests redirections for urllib implementation"""
867
_transport = HttpTransport_urllib
added
removed
bzrlib/tests/test_http.py
