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
InvalidStatusRequestHandler,
46
NoRangeRequestHandler,
47
SingleRangeRequestHandler,
48
TestCaseWithTwoWebservers,
49
TestCaseWithWebserver,
52
from bzrlib.transport import (
56
from bzrlib.transport.http import (
60
from bzrlib.transport.http._urllib import HttpTransport_urllib
63
class FakeManager(object):
68
def add_password(self, realm, host, username, password):
69
self.credentials.append([realm, host, username, password])
72
class RecordingServer(object):
73
"""A fake HTTP server.
75
It records the bytes sent to it, and replies with a 200.
78
def __init__(self, expect_body_tail=None):
81
:type expect_body_tail: str
82
:param expect_body_tail: a reply won't be sent until this string is
85
self._expect_body_tail = expect_body_tail
88
self.received_bytes = ''
91
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
92
self._sock.bind(('127.0.0.1', 0))
93
self.host, self.port = self._sock.getsockname()
94
self._ready = threading.Event()
95
self._thread = threading.Thread(target=self._accept_read_and_reply)
96
self._thread.setDaemon(True)
100
def _accept_read_and_reply(self):
103
self._sock.settimeout(5)
105
conn, address = self._sock.accept()
106
# On win32, the accepted connection will be non-blocking to start
107
# with because we're using settimeout.
108
conn.setblocking(True)
109
while not self.received_bytes.endswith(self._expect_body_tail):
110
self.received_bytes += conn.recv(4096)
111
conn.sendall('HTTP/1.1 200 OK\r\n')
112
except socket.timeout:
113
# Make sure the client isn't stuck waiting for us to e.g. accept.
116
# The client may have already closed the socket.
123
# We might have already closed it. We don't care.
129
class TestHttpUrls(TestCase):
131
# FIXME: Some of these tests should be done for both
134
def test_url_parsing(self):
136
url = extract_auth('http://example.com', f)
137
self.assertEquals('http://example.com', url)
138
self.assertEquals(0, len(f.credentials))
139
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
140
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
141
self.assertEquals(1, len(f.credentials))
142
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
145
def test_abs_url(self):
146
"""Construction of absolute http URLs"""
147
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
148
eq = self.assertEqualDiff
150
'http://bazaar-vcs.org/bzr/bzr.dev')
151
eq(t.abspath('foo/bar'),
152
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
153
eq(t.abspath('.bzr'),
154
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
155
eq(t.abspath('.bzr/1//2/./3'),
156
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
158
def test_invalid_http_urls(self):
159
"""Trap invalid construction of urls"""
160
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
161
self.assertRaises(ValueError,
164
t = HttpTransport_urllib('http://http://bazaar-vcs.org/bzr/bzr.dev/')
165
self.assertRaises(errors.InvalidURL, t.has, 'foo/bar')
167
def test_http_root_urls(self):
168
"""Construction of URLs from server root"""
169
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
170
eq = self.assertEqualDiff
171
eq(t.abspath('.bzr/tree-version'),
172
'http://bzr.ozlabs.org/.bzr/tree-version')
174
def test_http_impl_urls(self):
175
"""There are servers which ask for particular clients to connect"""
176
server = HttpServer_PyCurl()
179
url = server.get_url()
180
self.assertTrue(url.startswith('http+pycurl://'))
185
class TestHttpConnections(object):
186
"""Test the http connections.
188
This MUST be used by daughter classes that also inherit from
189
TestCaseWithWebserver.
191
We can't inherit directly from TestCaseWithWebserver or the
192
test framework will try to create an instance which cannot
193
run, its implementation being incomplete.
197
TestCaseWithWebserver.setUp(self)
198
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
199
transport=self.get_transport())
201
def test_http_has(self):
202
server = self.get_readonly_server()
203
t = self._transport(server.get_url())
204
self.assertEqual(t.has('foo/bar'), True)
205
self.assertEqual(len(server.logs), 1)
206
self.assertContainsRe(server.logs[0],
207
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
209
def test_http_has_not_found(self):
210
server = self.get_readonly_server()
211
t = self._transport(server.get_url())
212
self.assertEqual(t.has('not-found'), False)
213
self.assertContainsRe(server.logs[1],
214
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
216
def test_http_get(self):
217
server = self.get_readonly_server()
218
t = self._transport(server.get_url())
219
fp = t.get('foo/bar')
220
self.assertEqualDiff(
222
'contents of foo/bar\n')
223
self.assertEqual(len(server.logs), 1)
224
self.assertTrue(server.logs[0].find(
225
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
226
% bzrlib.__version__) > -1)
228
def test_get_smart_medium(self):
229
# For HTTP, get_smart_medium should return the transport object.
230
server = self.get_readonly_server()
231
http_transport = self._transport(server.get_url())
232
medium = http_transport.get_smart_medium()
233
self.assertIs(medium, http_transport)
235
def test_has_on_bogus_host(self):
236
# Get a free address and don't 'accept' on it, so that we
237
# can be sure there is no http handler there, but set a
238
# reasonable timeout to not slow down tests too much.
239
default_timeout = socket.getdefaulttimeout()
241
socket.setdefaulttimeout(2)
243
s.bind(('localhost', 0))
244
t = self._transport('http://%s:%s/' % s.getsockname())
245
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
247
socket.setdefaulttimeout(default_timeout)
250
class TestWithTransport_pycurl(object):
251
"""Test case to inherit from if pycurl is present"""
253
def _get_pycurl_maybe(self):
255
from bzrlib.transport.http._pycurl import PyCurlTransport
256
return PyCurlTransport
257
except errors.DependencyNotPresent:
258
raise TestSkipped('pycurl not present')
260
_transport = property(_get_pycurl_maybe)
263
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
264
"""Test http connections with urllib"""
266
_transport = HttpTransport_urllib
270
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
272
TestCaseWithWebserver):
273
"""Test http connections with pycurl"""
276
class TestHttpTransportRegistration(TestCase):
277
"""Test registrations of various http implementations"""
279
def test_http_registered(self):
280
# urlllib should always be present
281
t = get_transport('http+urllib://bzr.google.com/')
282
self.assertIsInstance(t, Transport)
283
self.assertIsInstance(t, HttpTransport_urllib)
286
class TestOffsets(TestCase):
287
"""Test offsets_to_ranges method"""
289
def test_offsets_to_ranges_simple(self):
290
to_range = HttpTransportBase.offsets_to_ranges
291
ranges = to_range([(10, 1)])
292
self.assertEqual([[10, 10]], ranges)
294
ranges = to_range([(0, 1), (1, 1)])
295
self.assertEqual([[0, 1]], ranges)
297
ranges = to_range([(1, 1), (0, 1)])
298
self.assertEqual([[0, 1]], ranges)
300
def test_offset_to_ranges_overlapped(self):
301
to_range = HttpTransportBase.offsets_to_ranges
303
ranges = to_range([(10, 1), (20, 2), (22, 5)])
304
self.assertEqual([[10, 10], [20, 26]], ranges)
306
ranges = to_range([(10, 1), (11, 2), (22, 5)])
307
self.assertEqual([[10, 12], [22, 26]], ranges)
310
class TestPost(object):
312
def _test_post_body_is_received(self, scheme):
313
server = RecordingServer(expect_body_tail='end-of-body')
315
self.addCleanup(server.tearDown)
316
url = '%s://%s:%s/' % (scheme, server.host, server.port)
318
http_transport = get_transport(url)
319
except errors.UnsupportedProtocol:
320
raise TestSkipped('%s not available' % scheme)
321
code, response = http_transport._post('abc def end-of-body')
323
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
324
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
325
# The transport should not be assuming that the server can accept
326
# chunked encoding the first time it connects, because HTTP/1.1, so we
327
# check for the literal string.
329
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
332
class TestPost_urllib(TestCase, TestPost):
333
"""TestPost for urllib implementation"""
335
_transport = HttpTransport_urllib
337
def test_post_body_is_received_urllib(self):
338
self._test_post_body_is_received('http+urllib')
341
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
342
"""TestPost for pycurl implementation"""
344
def test_post_body_is_received_pycurl(self):
345
self._test_post_body_is_received('http+pycurl')
348
class TestRangeHeader(TestCase):
349
"""Test range_header method"""
351
def check_header(self, value, ranges=[], tail=0):
352
range_header = HttpTransportBase.range_header
353
self.assertEqual(value, range_header(ranges, tail))
355
def test_range_header_single(self):
356
self.check_header('0-9', ranges=[[0,9]])
357
self.check_header('100-109', ranges=[[100,109]])
359
def test_range_header_tail(self):
360
self.check_header('-10', tail=10)
361
self.check_header('-50', tail=50)
363
def test_range_header_multi(self):
364
self.check_header('0-9,100-200,300-5000',
365
ranges=[(0,9), (100, 200), (300,5000)])
367
def test_range_header_mixed(self):
368
self.check_header('0-9,300-5000,-50',
369
ranges=[(0,9), (300,5000)],
373
class TestWallServer(object):
374
"""Tests exceptions during the connection phase"""
376
def create_transport_readonly_server(self):
377
return HttpServer(WallRequestHandler)
379
def test_http_has(self):
380
server = self.get_readonly_server()
381
t = self._transport(server.get_url())
382
# Unfortunately httplib (see HTTPResponse._read_status
383
# for details) make no distinction between a closed
384
# socket and badly formatted status line, so we can't
385
# just test for ConnectionError, we have to test
386
# InvalidHttpResponse too.
387
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
390
def test_http_get(self):
391
server = self.get_readonly_server()
392
t = self._transport(server.get_url())
393
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
397
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
398
"""Tests "wall" server for urllib implementation"""
400
_transport = HttpTransport_urllib
403
class TestWallServer_pycurl(TestWithTransport_pycurl,
405
TestCaseWithWebserver):
406
"""Tests "wall" server for pycurl implementation"""
409
class TestBadStatusServer(object):
410
"""Tests bad status from server."""
412
def create_transport_readonly_server(self):
413
return HttpServer(BadStatusRequestHandler)
415
def test_http_has(self):
416
server = self.get_readonly_server()
417
t = self._transport(server.get_url())
418
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
420
def test_http_get(self):
421
server = self.get_readonly_server()
422
t = self._transport(server.get_url())
423
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
426
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
427
"""Tests bad status server for urllib implementation"""
429
_transport = HttpTransport_urllib
432
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
434
TestCaseWithWebserver):
435
"""Tests bad status server for pycurl implementation"""
438
class TestInvalidStatusServer(TestBadStatusServer):
439
"""Tests invalid status from server.
441
Both implementations raises the same error as for a bad status.
444
def create_transport_readonly_server(self):
445
return HttpServer(InvalidStatusRequestHandler)
448
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
449
TestCaseWithWebserver):
450
"""Tests invalid status server for urllib implementation"""
452
_transport = HttpTransport_urllib
455
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
456
TestInvalidStatusServer,
457
TestCaseWithWebserver):
458
"""Tests invalid status server for pycurl implementation"""
461
class TestBadProtocolServer(object):
462
"""Tests bad protocol from server."""
464
def create_transport_readonly_server(self):
465
return HttpServer(BadProtocolRequestHandler)
467
def test_http_has(self):
468
server = self.get_readonly_server()
469
t = self._transport(server.get_url())
470
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
472
def test_http_get(self):
473
server = self.get_readonly_server()
474
t = self._transport(server.get_url())
475
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
478
class TestBadProtocolServer_urllib(TestBadProtocolServer,
479
TestCaseWithWebserver):
480
"""Tests bad protocol server for urllib implementation"""
482
_transport = HttpTransport_urllib
484
# curl don't check the protocol version
485
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
486
# TestBadProtocolServer,
487
# TestCaseWithWebserver):
488
# """Tests bad protocol server for pycurl implementation"""
491
class TestForbiddenServer(object):
492
"""Tests forbidden server"""
494
def create_transport_readonly_server(self):
495
return HttpServer(ForbiddenRequestHandler)
497
def test_http_has(self):
498
server = self.get_readonly_server()
499
t = self._transport(server.get_url())
500
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
502
def test_http_get(self):
503
server = self.get_readonly_server()
504
t = self._transport(server.get_url())
505
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
508
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
509
"""Tests forbidden server for urllib implementation"""
511
_transport = HttpTransport_urllib
514
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
516
TestCaseWithWebserver):
517
"""Tests forbidden server for pycurl implementation"""
520
class TestRecordingServer(TestCase):
522
def test_create(self):
523
server = RecordingServer(expect_body_tail=None)
524
self.assertEqual('', server.received_bytes)
525
self.assertEqual(None, server.host)
526
self.assertEqual(None, server.port)
528
def test_setUp_and_tearDown(self):
529
server = RecordingServer(expect_body_tail=None)
532
self.assertNotEqual(None, server.host)
533
self.assertNotEqual(None, server.port)
536
self.assertEqual(None, server.host)
537
self.assertEqual(None, server.port)
539
def test_send_receive_bytes(self):
540
server = RecordingServer(expect_body_tail='c')
542
self.addCleanup(server.tearDown)
543
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
544
sock.connect((server.host, server.port))
546
self.assertEqual('HTTP/1.1 200 OK\r\n',
547
osutils.recv_all(sock, 4096))
548
self.assertEqual('abc', server.received_bytes)
551
class TestRangeRequestServer(object):
552
"""Tests readv requests against server.
554
This MUST be used by daughter classes that also inherit from
555
TestCaseWithWebserver.
557
We can't inherit directly from TestCaseWithWebserver or the
558
test framework will try to create an instance which cannot
559
run, its implementation being incomplete.
563
TestCaseWithWebserver.setUp(self)
564
self.build_tree_contents([('a', '0123456789')],)
566
def test_readv(self):
567
server = self.get_readonly_server()
568
t = self._transport(server.get_url())
569
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
570
self.assertEqual(l[0], (0, '0'))
571
self.assertEqual(l[1], (1, '1'))
572
self.assertEqual(l[2], (3, '34'))
573
self.assertEqual(l[3], (9, '9'))
575
def test_readv_out_of_order(self):
576
server = self.get_readonly_server()
577
t = self._transport(server.get_url())
578
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
579
self.assertEqual(l[0], (1, '1'))
580
self.assertEqual(l[1], (9, '9'))
581
self.assertEqual(l[2], (0, '0'))
582
self.assertEqual(l[3], (3, '34'))
584
def test_readv_invalid_ranges(self):
585
server = self.get_readonly_server()
586
t = self._transport(server.get_url())
588
# This is intentionally reading off the end of the file
589
# since we are sure that it cannot get there
590
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
591
t.readv, 'a', [(1,1), (8,10)])
593
# This is trying to seek past the end of the file, it should
594
# also raise a special error
595
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
596
t.readv, 'a', [(12,2)])
599
class TestSingleRangeRequestServer(TestRangeRequestServer):
600
"""Test readv against a server which accept only single range requests"""
602
def create_transport_readonly_server(self):
603
return HttpServer(SingleRangeRequestHandler)
606
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
607
TestCaseWithWebserver):
608
"""Tests single range requests accepting server for urllib implementation"""
610
_transport = HttpTransport_urllib
613
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
614
TestSingleRangeRequestServer,
615
TestCaseWithWebserver):
616
"""Tests single range requests accepting server for pycurl implementation"""
619
class TestNoRangeRequestServer(TestRangeRequestServer):
620
"""Test readv against a server which do not accept range requests"""
622
def create_transport_readonly_server(self):
623
return HttpServer(NoRangeRequestHandler)
626
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
627
TestCaseWithWebserver):
628
"""Tests range requests refusing server for urllib implementation"""
630
_transport = HttpTransport_urllib
633
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
634
TestNoRangeRequestServer,
635
TestCaseWithWebserver):
636
"""Tests range requests refusing server for pycurl implementation"""
639
class TestProxyHttpServer(object):
640
"""Tests proxy server.
642
This MUST be used by daughter classes that also inherit from
643
TestCaseWithTwoWebservers.
645
We can't inherit directly from TestCaseWithTwoWebservers or
646
the test framework will try to create an instance which
647
cannot run, its implementation being incomplete.
649
Be aware that we do not setup a real proxy here. Instead, we
650
check that the *connection* goes through the proxy by serving
651
different content (the faked proxy server append '-proxied'
655
# FIXME: We don't have an https server available, so we don't
656
# test https connections.
659
TestCaseWithTwoWebservers.setUp(self)
660
self.build_tree_contents([('foo', 'contents of foo\n'),
661
('foo-proxied', 'proxied contents of foo\n')])
662
# Let's setup some attributes for tests
663
self.server = self.get_readonly_server()
664
self.no_proxy_host = 'localhost:%d' % self.server.port
665
# The secondary server is the proxy
666
self.proxy = self.get_secondary_server()
667
self.proxy_url = self.proxy.get_url()
670
def create_transport_secondary_server(self):
671
"""Creates an http server that will serve files with
672
'-proxied' appended to their names.
674
return HttpServer(FakeProxyRequestHandler)
676
def _set_and_capture_env_var(self, name, new_value):
677
"""Set an environment variable, and reset it when finished."""
678
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
680
def _install_env(self, env):
681
for name, value in env.iteritems():
682
self._set_and_capture_env_var(name, value)
684
def _restore_env(self):
685
for name, value in self._old_env.iteritems():
686
osutils.set_or_unset_env(name, value)
688
def proxied_in_env(self, env):
689
self._install_env(env)
690
url = self.server.get_url()
691
t = self._transport(url)
693
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
697
def not_proxied_in_env(self, env):
698
self._install_env(env)
699
url = self.server.get_url()
700
t = self._transport(url)
702
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
706
def test_http_proxy(self):
707
self.proxied_in_env({'http_proxy': self.proxy_url})
709
def test_HTTP_PROXY(self):
710
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
712
def test_all_proxy(self):
713
self.proxied_in_env({'all_proxy': self.proxy_url})
715
def test_ALL_PROXY(self):
716
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
718
def test_http_proxy_with_no_proxy(self):
719
self.not_proxied_in_env({'http_proxy': self.proxy_url,
720
'no_proxy': self.no_proxy_host})
722
def test_HTTP_PROXY_with_NO_PROXY(self):
723
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
724
'NO_PROXY': self.no_proxy_host})
726
def test_all_proxy_with_no_proxy(self):
727
self.not_proxied_in_env({'all_proxy': self.proxy_url,
728
'no_proxy': self.no_proxy_host})
730
def test_ALL_PROXY_with_NO_PROXY(self):
731
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
732
'NO_PROXY': self.no_proxy_host})
735
class TestProxyHttpServer_urllib(TestProxyHttpServer,
736
TestCaseWithTwoWebservers):
737
"""Tests proxy server for urllib implementation"""
739
_transport = HttpTransport_urllib
742
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
744
TestCaseWithTwoWebservers):
745
"""Tests proxy server for pycurl implementation"""
748
TestProxyHttpServer.setUp(self)
749
# Oh my ! pycurl does not check for the port as part of
750
# no_proxy :-( So we just test the host part
751
self.no_proxy_host = 'localhost'
753
def test_HTTP_PROXY(self):
754
# pycurl do not check HTTP_PROXY for security reasons
755
# (for use in a CGI context that we do not care
756
# about. Should we ?)
759
def test_HTTP_PROXY_with_NO_PROXY(self):
763
class TestRanges(object):
764
"""Test the Range header in GET methods..
766
This MUST be used by daughter classes that also inherit from
767
TestCaseWithWebserver.
769
We can't inherit directly from TestCaseWithWebserver or the
770
test framework will try to create an instance which cannot
771
run, its implementation being incomplete.
775
TestCaseWithWebserver.setUp(self)
776
self.build_tree_contents([('a', '0123456789')],)
777
server = self.get_readonly_server()
778
self.transport = self._transport(server.get_url())
780
def _file_contents(self, relpath, ranges, tail_amount=0):
781
code, data = self.transport._get(relpath, ranges)
782
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
783
for start, end in ranges:
785
yield data.read(end - start + 1)
787
def _file_tail(self, relpath, tail_amount):
788
code, data = self.transport._get(relpath, [], tail_amount)
789
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
790
data.seek(-tail_amount + 1, 2)
791
return data.read(tail_amount)
793
def test_range_header(self):
795
map(self.assertEqual,['0', '234'],
796
list(self._file_contents('a', [(0,0), (2,4)])),)
798
self.assertEqual('789', self._file_tail('a', 3))
799
# Syntactically invalid range
800
self.assertRaises(errors.InvalidRange,
801
self.transport._get, 'a', [(4, 3)])
802
# Semantically invalid range
803
self.assertRaises(errors.InvalidRange,
804
self.transport._get, 'a', [(42, 128)])
807
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
808
"""Test the Range header in GET methods for urllib implementation"""
810
_transport = HttpTransport_urllib
813
class TestRanges_pycurl(TestWithTransport_pycurl,
815
TestCaseWithWebserver):
816
"""Test the Range header in GET methods for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
