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.
120
# We might have already closed it. We don't care.
126
class TestHttpUrls(TestCase):
128
# FIXME: Some of these tests should be done for both
131
def test_url_parsing(self):
133
url = extract_auth('http://example.com', f)
134
self.assertEquals('http://example.com', url)
135
self.assertEquals(0, len(f.credentials))
136
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
137
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
138
self.assertEquals(1, len(f.credentials))
139
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
142
def test_abs_url(self):
143
"""Construction of absolute http URLs"""
144
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
145
eq = self.assertEqualDiff
147
'http://bazaar-vcs.org/bzr/bzr.dev')
148
eq(t.abspath('foo/bar'),
149
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
150
eq(t.abspath('.bzr'),
151
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
152
eq(t.abspath('.bzr/1//2/./3'),
153
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
155
def test_invalid_http_urls(self):
156
"""Trap invalid construction of urls"""
157
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
158
self.assertRaises(ValueError,
161
t = HttpTransport_urllib('http://http://bazaar-vcs.org/bzr/bzr.dev/')
162
self.assertRaises(errors.InvalidURL, t.has, 'foo/bar')
164
def test_http_root_urls(self):
165
"""Construction of URLs from server root"""
166
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
167
eq = self.assertEqualDiff
168
eq(t.abspath('.bzr/tree-version'),
169
'http://bzr.ozlabs.org/.bzr/tree-version')
171
def test_http_impl_urls(self):
172
"""There are servers which ask for particular clients to connect"""
173
server = HttpServer_PyCurl()
176
url = server.get_url()
177
self.assertTrue(url.startswith('http+pycurl://'))
182
class TestHttpConnections(object):
183
"""Test the http connections.
185
This MUST be used by daughter classes that also inherit from
186
TestCaseWithWebserver.
188
We can't inherit directly from TestCaseWithWebserver or the
189
test framework will try to create an instance which cannot
190
run, its implementation being incomplete.
194
TestCaseWithWebserver.setUp(self)
195
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
196
transport=self.get_transport())
198
def test_http_has(self):
199
server = self.get_readonly_server()
200
t = self._transport(server.get_url())
201
self.assertEqual(t.has('foo/bar'), True)
202
self.assertEqual(len(server.logs), 1)
203
self.assertContainsRe(server.logs[0],
204
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
206
def test_http_has_not_found(self):
207
server = self.get_readonly_server()
208
t = self._transport(server.get_url())
209
self.assertEqual(t.has('not-found'), False)
210
self.assertContainsRe(server.logs[1],
211
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
213
def test_http_get(self):
214
server = self.get_readonly_server()
215
t = self._transport(server.get_url())
216
fp = t.get('foo/bar')
217
self.assertEqualDiff(
219
'contents of foo/bar\n')
220
self.assertEqual(len(server.logs), 1)
221
self.assertTrue(server.logs[0].find(
222
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
223
% bzrlib.__version__) > -1)
225
def test_get_smart_medium(self):
226
# For HTTP, get_smart_medium should return the transport object.
227
server = self.get_readonly_server()
228
http_transport = self._transport(server.get_url())
229
medium = http_transport.get_smart_medium()
230
self.assertIs(medium, http_transport)
232
def test_has_on_bogus_host(self):
233
# Get a free address and don't 'accept' on it, so that we
234
# can be sure there is no http handler there, but set a
235
# reasonable timeout to not slow down tests too much.
236
default_timeout = socket.getdefaulttimeout()
238
socket.setdefaulttimeout(2)
240
s.bind(('localhost', 0))
241
t = self._transport('http://%s:%s/' % s.getsockname())
242
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
244
socket.setdefaulttimeout(default_timeout)
247
class TestWithTransport_pycurl(object):
248
"""Test case to inherit from if pycurl is present"""
250
def _get_pycurl_maybe(self):
252
from bzrlib.transport.http._pycurl import PyCurlTransport
253
return PyCurlTransport
254
except errors.DependencyNotPresent:
255
raise TestSkipped('pycurl not present')
257
_transport = property(_get_pycurl_maybe)
260
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
261
"""Test http connections with urllib"""
263
_transport = HttpTransport_urllib
267
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
269
TestCaseWithWebserver):
270
"""Test http connections with pycurl"""
273
class TestHttpTransportRegistration(TestCase):
274
"""Test registrations of various http implementations"""
276
def test_http_registered(self):
277
# urlllib should always be present
278
t = get_transport('http+urllib://bzr.google.com/')
279
self.assertIsInstance(t, Transport)
280
self.assertIsInstance(t, HttpTransport_urllib)
283
class TestOffsets(TestCase):
284
"""Test offsets_to_ranges method"""
286
def test_offsets_to_ranges_simple(self):
287
to_range = HttpTransportBase.offsets_to_ranges
288
ranges = to_range([(10, 1)])
289
self.assertEqual([[10, 10]], ranges)
291
ranges = to_range([(0, 1), (1, 1)])
292
self.assertEqual([[0, 1]], ranges)
294
ranges = to_range([(1, 1), (0, 1)])
295
self.assertEqual([[0, 1]], ranges)
297
def test_offset_to_ranges_overlapped(self):
298
to_range = HttpTransportBase.offsets_to_ranges
300
ranges = to_range([(10, 1), (20, 2), (22, 5)])
301
self.assertEqual([[10, 10], [20, 26]], ranges)
303
ranges = to_range([(10, 1), (11, 2), (22, 5)])
304
self.assertEqual([[10, 12], [22, 26]], ranges)
307
class TestPost(TestCase):
309
def _test_post_body_is_received(self, scheme):
310
server = RecordingServer(expect_body_tail='end-of-body')
312
self.addCleanup(server.tearDown)
313
url = '%s://%s:%s/' % (scheme, server.host, server.port)
315
http_transport = get_transport(url)
316
except errors.UnsupportedProtocol:
317
raise TestSkipped('%s not available' % scheme)
318
code, response = http_transport._post('abc def end-of-body')
320
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
321
self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
322
# The transport should not be assuming that the server can accept
323
# chunked encoding the first time it connects, because HTTP/1.1, so we
324
# check for the literal string.
326
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
328
def test_post_body_is_received_urllib(self):
329
self._test_post_body_is_received('http+urllib')
331
def test_post_body_is_received_pycurl(self):
332
self._test_post_body_is_received('http+pycurl')
335
class TestRangeHeader(TestCase):
336
"""Test range_header method"""
338
def check_header(self, value, ranges=[], tail=0):
339
range_header = HttpTransportBase.range_header
340
self.assertEqual(value, range_header(ranges, tail))
342
def test_range_header_single(self):
343
self.check_header('0-9', ranges=[[0,9]])
344
self.check_header('100-109', ranges=[[100,109]])
346
def test_range_header_tail(self):
347
self.check_header('-10', tail=10)
348
self.check_header('-50', tail=50)
350
def test_range_header_multi(self):
351
self.check_header('0-9,100-200,300-5000',
352
ranges=[(0,9), (100, 200), (300,5000)])
354
def test_range_header_mixed(self):
355
self.check_header('0-9,300-5000,-50',
356
ranges=[(0,9), (300,5000)],
360
class TestWallServer(object):
361
"""Tests exceptions during the connection phase"""
363
def create_transport_readonly_server(self):
364
return HttpServer(WallRequestHandler)
366
def test_http_has(self):
367
server = self.get_readonly_server()
368
t = self._transport(server.get_url())
369
# Unfortunately httplib (see HTTPResponse._read_status
370
# for details) make no distinction between a closed
371
# socket and badly formatted status line, so we can't
372
# just test for ConnectionError, we have to test
373
# InvalidHttpResponse too.
374
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
377
def test_http_get(self):
378
server = self.get_readonly_server()
379
t = self._transport(server.get_url())
380
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
384
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
385
"""Tests "wall" server for urllib implementation"""
387
_transport = HttpTransport_urllib
390
class TestWallServer_pycurl(TestWithTransport_pycurl,
392
TestCaseWithWebserver):
393
"""Tests "wall" server for pycurl implementation"""
396
class TestBadStatusServer(object):
397
"""Tests bad status from server."""
399
def create_transport_readonly_server(self):
400
return HttpServer(BadStatusRequestHandler)
402
def test_http_has(self):
403
server = self.get_readonly_server()
404
t = self._transport(server.get_url())
405
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
407
def test_http_get(self):
408
server = self.get_readonly_server()
409
t = self._transport(server.get_url())
410
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
413
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
414
"""Tests bad status server for urllib implementation"""
416
_transport = HttpTransport_urllib
419
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
421
TestCaseWithWebserver):
422
"""Tests bad status server for pycurl implementation"""
425
class TestInvalidStatusServer(TestBadStatusServer):
426
"""Tests invalid status from server.
428
Both implementations raises the same error as for a bad status.
431
def create_transport_readonly_server(self):
432
return HttpServer(InvalidStatusRequestHandler)
435
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
436
TestCaseWithWebserver):
437
"""Tests invalid status server for urllib implementation"""
439
_transport = HttpTransport_urllib
442
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
443
TestInvalidStatusServer,
444
TestCaseWithWebserver):
445
"""Tests invalid status server for pycurl implementation"""
448
class TestBadProtocolServer(object):
449
"""Tests bad protocol from server."""
451
def create_transport_readonly_server(self):
452
return HttpServer(BadProtocolRequestHandler)
454
def test_http_has(self):
455
server = self.get_readonly_server()
456
t = self._transport(server.get_url())
457
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
459
def test_http_get(self):
460
server = self.get_readonly_server()
461
t = self._transport(server.get_url())
462
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
465
class TestBadProtocolServer_urllib(TestBadProtocolServer,
466
TestCaseWithWebserver):
467
"""Tests bad protocol server for urllib implementation"""
469
_transport = HttpTransport_urllib
471
# curl don't check the protocol version
472
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
473
# TestBadProtocolServer,
474
# TestCaseWithWebserver):
475
# """Tests bad protocol server for pycurl implementation"""
478
class TestForbiddenServer(object):
479
"""Tests forbidden server"""
481
def create_transport_readonly_server(self):
482
return HttpServer(ForbiddenRequestHandler)
484
def test_http_has(self):
485
server = self.get_readonly_server()
486
t = self._transport(server.get_url())
487
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
489
def test_http_get(self):
490
server = self.get_readonly_server()
491
t = self._transport(server.get_url())
492
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
495
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
496
"""Tests forbidden server for urllib implementation"""
498
_transport = HttpTransport_urllib
501
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
503
TestCaseWithWebserver):
504
"""Tests forbidden server for pycurl implementation"""
507
class TestRecordingServer(TestCase):
509
def test_create(self):
510
server = RecordingServer(expect_body_tail=None)
511
self.assertEqual('', server.received_bytes)
512
self.assertEqual(None, server.host)
513
self.assertEqual(None, server.port)
515
def test_setUp_and_tearDown(self):
516
server = RecordingServer(expect_body_tail=None)
519
self.assertNotEqual(None, server.host)
520
self.assertNotEqual(None, server.port)
523
self.assertEqual(None, server.host)
524
self.assertEqual(None, server.port)
526
def test_send_receive_bytes(self):
527
server = RecordingServer(expect_body_tail='c')
529
self.addCleanup(server.tearDown)
530
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
531
sock.connect((server.host, server.port))
533
self.assertEqual('HTTP/1.1 200 OK\r\n',
534
osutils.recv_all(sock, 4096))
535
self.assertEqual('abc', server.received_bytes)
538
class TestRangeRequestServer(object):
539
"""Tests readv requests against server.
541
This MUST be used by daughter classes that also inherit from
542
TestCaseWithWebserver.
544
We can't inherit directly from TestCaseWithWebserver or the
545
test framework will try to create an instance which cannot
546
run, its implementation being incomplete.
550
TestCaseWithWebserver.setUp(self)
551
self.build_tree_contents([('a', '0123456789')],)
553
def test_readv(self):
554
server = self.get_readonly_server()
555
t = self._transport(server.get_url())
556
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
557
self.assertEqual(l[0], (0, '0'))
558
self.assertEqual(l[1], (1, '1'))
559
self.assertEqual(l[2], (3, '34'))
560
self.assertEqual(l[3], (9, '9'))
562
def test_readv_out_of_order(self):
563
server = self.get_readonly_server()
564
t = self._transport(server.get_url())
565
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
566
self.assertEqual(l[0], (1, '1'))
567
self.assertEqual(l[1], (9, '9'))
568
self.assertEqual(l[2], (0, '0'))
569
self.assertEqual(l[3], (3, '34'))
571
def test_readv_invalid_ranges(self):
572
server = self.get_readonly_server()
573
t = self._transport(server.get_url())
575
# This is intentionally reading off the end of the file
576
# since we are sure that it cannot get there
577
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
578
t.readv, 'a', [(1,1), (8,10)])
580
# This is trying to seek past the end of the file, it should
581
# also raise a special error
582
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
583
t.readv, 'a', [(12,2)])
586
class TestSingleRangeRequestServer(TestRangeRequestServer):
587
"""Test readv against a server which accept only single range requests"""
589
def create_transport_readonly_server(self):
590
return HttpServer(SingleRangeRequestHandler)
593
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
594
TestCaseWithWebserver):
595
"""Tests single range requests accepting server for urllib implementation"""
597
_transport = HttpTransport_urllib
600
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
601
TestSingleRangeRequestServer,
602
TestCaseWithWebserver):
603
"""Tests single range requests accepting server for pycurl implementation"""
606
class TestNoRangeRequestServer(TestRangeRequestServer):
607
"""Test readv against a server which do not accept range requests"""
609
def create_transport_readonly_server(self):
610
return HttpServer(NoRangeRequestHandler)
613
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
614
TestCaseWithWebserver):
615
"""Tests range requests refusing server for urllib implementation"""
617
_transport = HttpTransport_urllib
620
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
621
TestNoRangeRequestServer,
622
TestCaseWithWebserver):
623
"""Tests range requests refusing server for pycurl implementation"""
626
class TestProxyHttpServer(object):
627
"""Tests proxy server.
629
This MUST be used by daughter classes that also inherit from
630
TestCaseWithTwoWebservers.
632
We can't inherit directly from TestCaseWithTwoWebservers or
633
the test framework will try to create an instance which
634
cannot run, its implementation being incomplete.
636
Be aware that we do not setup a real proxy here. Instead, we
637
check that the *connection* goes through the proxy by serving
638
different content (the faked proxy server append '-proxied'
642
# FIXME: We don't have an https server available, so we don't
643
# test https connections.
646
TestCaseWithTwoWebservers.setUp(self)
647
self.build_tree_contents([('foo', 'contents of foo\n'),
648
('foo-proxied', 'proxied contents of foo\n')])
649
# Let's setup some attributes for tests
650
self.server = self.get_readonly_server()
651
self.no_proxy_host = 'localhost:%d' % self.server.port
652
# The secondary server is the proxy
653
self.proxy = self.get_secondary_server()
654
self.proxy_url = self.proxy.get_url()
657
def create_transport_secondary_server(self):
658
"""Creates an http server that will serve files with
659
'-proxied' appended to their names.
661
return HttpServer(FakeProxyRequestHandler)
663
def _set_and_capture_env_var(self, name, new_value):
664
"""Set an environment variable, and reset it when finished."""
665
self._old_env[name] = osutils.set_or_unset_env(name, new_value)
667
def _install_env(self, env):
668
for name, value in env.iteritems():
669
self._set_and_capture_env_var(name, value)
671
def _restore_env(self):
672
for name, value in self._old_env.iteritems():
673
osutils.set_or_unset_env(name, value)
675
def proxied_in_env(self, env):
676
self._install_env(env)
677
url = self.server.get_url()
678
t = self._transport(url)
680
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
684
def not_proxied_in_env(self, env):
685
self._install_env(env)
686
url = self.server.get_url()
687
t = self._transport(url)
689
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
693
def test_http_proxy(self):
694
self.proxied_in_env({'http_proxy': self.proxy_url})
696
def test_HTTP_PROXY(self):
697
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
699
def test_all_proxy(self):
700
self.proxied_in_env({'all_proxy': self.proxy_url})
702
def test_ALL_PROXY(self):
703
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
705
def test_http_proxy_with_no_proxy(self):
706
self.not_proxied_in_env({'http_proxy': self.proxy_url,
707
'no_proxy': self.no_proxy_host})
709
def test_HTTP_PROXY_with_NO_PROXY(self):
710
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
711
'NO_PROXY': self.no_proxy_host})
713
def test_all_proxy_with_no_proxy(self):
714
self.not_proxied_in_env({'all_proxy': self.proxy_url,
715
'no_proxy': self.no_proxy_host})
717
def test_ALL_PROXY_with_NO_PROXY(self):
718
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
719
'NO_PROXY': self.no_proxy_host})
722
class TestProxyHttpServer_urllib(TestProxyHttpServer,
723
TestCaseWithTwoWebservers):
724
"""Tests proxy server for urllib implementation"""
726
_transport = HttpTransport_urllib
729
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
731
TestCaseWithTwoWebservers):
732
"""Tests proxy server for pycurl implementation"""
735
TestProxyHttpServer.setUp(self)
736
# Oh my ! pycurl does not check for the port as part of
737
# no_proxy :-( So we just test the host part
738
self.no_proxy_host = 'localhost'
740
def test_HTTP_PROXY(self):
741
# pycurl do not check HTTP_PROXY for security reasons
742
# (for use in a CGI context that we do not care
743
# about. Should we ?)
746
def test_HTTP_PROXY_with_NO_PROXY(self):
750
class TestRanges(object):
751
"""Test the Range header in GET methods..
753
This MUST be used by daughter classes that also inherit from
754
TestCaseWithWebserver.
756
We can't inherit directly from TestCaseWithWebserver or the
757
test framework will try to create an instance which cannot
758
run, its implementation being incomplete.
762
TestCaseWithWebserver.setUp(self)
763
self.build_tree_contents([('a', '0123456789')],)
764
server = self.get_readonly_server()
765
self.transport = self._transport(server.get_url())
767
def _file_contents(self, relpath, ranges, tail_amount=0):
768
code, data = self.transport._get(relpath, ranges)
769
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
770
for start, end in ranges:
772
yield data.read(end - start + 1)
774
def _file_tail(self, relpath, tail_amount):
775
code, data = self.transport._get(relpath, [], tail_amount)
776
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
777
data.seek(-tail_amount + 1, 2)
778
return data.read(tail_amount)
780
def test_range_header(self):
782
map(self.assertEqual,['0', '234'],
783
list(self._file_contents('a', [(0,0), (2,4)])),)
785
self.assertEqual('789', self._file_tail('a', 3))
786
# Syntactically invalid range
787
self.assertRaises(errors.InvalidRange,
788
self.transport._get, 'a', [(4, 3)])
789
# Semantically invalid range
790
self.assertRaises(errors.InvalidRange,
791
self.transport._get, 'a', [(42, 128)])
794
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
795
"""Test the Range header in GET methods for urllib implementation"""
797
_transport = HttpTransport_urllib
800
class TestRanges_pycurl(TestWithTransport_pycurl,
802
TestCaseWithWebserver):
803
"""Test the Range header in GET methods for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
