1
# Copyright (C) 2005, 2006 Canonical
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?
25
from bzrlib.errors import (
30
from bzrlib.tests import (
34
from bzrlib.tests.HttpServer import (
39
from bzrlib.tests.HTTPTestUtil import (
40
BadProtocolRequestHandler,
41
BadStatusRequestHandler,
42
InvalidStatusRequestHandler,
43
TestCaseWithWebserver,
46
from bzrlib.transport import (
50
from bzrlib.transport.http import (
54
from bzrlib.transport.http._urllib import HttpTransport_urllib
57
class FakeManager (object):
62
def add_password(self, realm, host, username, password):
63
self.credentials.append([realm, host, username, password])
66
class TestHttpUrls(TestCase):
68
def test_url_parsing(self):
70
url = extract_auth('http://example.com', f)
71
self.assertEquals('http://example.com', url)
72
self.assertEquals(0, len(f.credentials))
73
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
74
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
75
self.assertEquals(1, len(f.credentials))
76
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
79
def test_abs_url(self):
80
"""Construction of absolute http URLs"""
81
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
82
eq = self.assertEqualDiff
84
'http://bazaar-vcs.org/bzr/bzr.dev')
85
eq(t.abspath('foo/bar'),
86
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
88
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
89
eq(t.abspath('.bzr/1//2/./3'),
90
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
92
def test_invalid_http_urls(self):
93
"""Trap invalid construction of urls"""
94
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
95
self.assertRaises(ValueError,
99
def test_http_root_urls(self):
100
"""Construction of URLs from server root"""
101
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
102
eq = self.assertEqualDiff
103
eq(t.abspath('.bzr/tree-version'),
104
'http://bzr.ozlabs.org/.bzr/tree-version')
106
def test_http_impl_urls(self):
107
"""There are servers which ask for particular clients to connect"""
108
server = HttpServer_PyCurl()
111
url = server.get_url()
112
self.assertTrue(url.startswith('http+pycurl://'))
117
class TestHttpConnections(object):
118
"""Test the http connections.
120
This MUST be used by daughter classes that also inherit from
121
TestCaseWithWebserver.
123
We can't inherit directly from TestCaseWithWebserver or the
124
test framework will try to create an instance which cannot
125
run, its implementation being incomplete.
129
TestCaseWithWebserver.setUp(self)
130
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
131
transport=self.get_transport())
133
def test_http_has(self):
134
server = self.get_readonly_server()
135
t = self._transport(server.get_url())
136
self.assertEqual(t.has('foo/bar'), True)
137
self.assertEqual(len(server.logs), 1)
138
self.assertContainsRe(server.logs[0],
139
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
141
def test_http_has_not_found(self):
142
server = self.get_readonly_server()
143
t = self._transport(server.get_url())
144
self.assertEqual(t.has('not-found'), False)
145
self.assertContainsRe(server.logs[1],
146
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
148
def test_http_get(self):
149
server = self.get_readonly_server()
150
t = self._transport(server.get_url())
151
fp = t.get('foo/bar')
152
self.assertEqualDiff(
154
'contents of foo/bar\n')
155
self.assertEqual(len(server.logs), 1)
156
self.assertTrue(server.logs[0].find(
157
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
158
% bzrlib.__version__) > -1)
160
def test_has_on_bogus_host(self):
161
# Get a free address and don't 'accept' on it, so that we
162
# can be sure there is no http handler there, but set a
163
# reasonable timeout to not slow down tests too much.
164
default_timeout = socket.getdefaulttimeout()
166
socket.setdefaulttimeout(2)
168
s.bind(('localhost', 0))
169
t = self._transport('http://%s:%s/' % s.getsockname())
170
self.assertRaises(ConnectionError, t.has, 'foo/bar')
172
socket.setdefaulttimeout(default_timeout)
175
class TestWithTransport_pycurl(object):
176
"""Test case to inherit from if pycurl is present"""
177
def _get_pycurl_maybe(self):
179
from bzrlib.transport.http._pycurl import PyCurlTransport
180
return PyCurlTransport
181
except DependencyNotPresent:
182
raise TestSkipped('pycurl not present')
184
_transport = property(_get_pycurl_maybe)
187
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
188
"""Test http connections with urllib"""
190
_transport = HttpTransport_urllib
194
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
196
TestCaseWithWebserver):
197
"""Test http connections with pycurl"""
200
class TestHttpTransportRegistration(TestCase):
201
"""Test registrations of various http implementations"""
203
def test_http_registered(self):
204
# urlllib should always be present
205
t = get_transport('http+urllib://bzr.google.com/')
206
self.assertIsInstance(t, Transport)
207
self.assertIsInstance(t, HttpTransport_urllib)
210
class TestOffsets(TestCase):
211
"""Test offsets_to_ranges method"""
213
def test_offsets_to_ranges_simple(self):
214
to_range = HttpTransportBase.offsets_to_ranges
215
ranges = to_range([(10, 1)])
216
self.assertEqual([[10, 10]], ranges)
218
ranges = to_range([(0, 1), (1, 1)])
219
self.assertEqual([[0, 1]], ranges)
221
ranges = to_range([(1, 1), (0, 1)])
222
self.assertEqual([[0, 1]], ranges)
224
def test_offset_to_ranges_overlapped(self):
225
to_range = HttpTransportBase.offsets_to_ranges
227
ranges = to_range([(10, 1), (20, 2), (22, 5)])
228
self.assertEqual([[10, 10], [20, 26]], ranges)
230
ranges = to_range([(10, 1), (11, 2), (22, 5)])
231
self.assertEqual([[10, 12], [22, 26]], ranges)
234
class TestRangeHeader(TestCase):
235
"""Test range_header method"""
237
def check_header(self, value, ranges=[], tail=0):
238
range_header = HttpTransportBase.range_header
239
self.assertEqual(value, range_header(ranges, tail))
241
def test_range_header_single(self):
242
self.check_header('0-9', ranges=[[0,9]])
243
self.check_header('100-109', ranges=[[100,109]])
245
def test_range_header_tail(self):
246
self.check_header('-10', tail=10)
247
self.check_header('-50', tail=50)
249
def test_range_header_multi(self):
250
self.check_header('0-9,100-200,300-5000',
251
ranges=[(0,9), (100, 200), (300,5000)])
253
def test_range_header_mixed(self):
254
self.check_header('0-9,300-5000,-50',
255
ranges=[(0,9), (300,5000)],
259
class TestWallServer(object):
260
"""Tests exceptions during the connection phase"""
262
def create_transport_readonly_server(self):
263
return HttpServer(WallRequestHandler)
265
def test_http_has(self):
266
server = self.get_readonly_server()
267
t = self._transport(server.get_url())
268
self.assertRaises(ConnectionError, t.has, 'foo/bar')
270
def test_http_get(self):
271
server = self.get_readonly_server()
272
t = self._transport(server.get_url())
273
self.assertRaises(ConnectionError, t.get, 'foo/bar')
276
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
277
"""Tests WallServer for urllib implementation"""
279
_transport = HttpTransport_urllib
282
class TestWallServer_pycurl(TestWithTransport_pycurl,
284
TestCaseWithWebserver):
285
"""Tests WallServer for pycurl implementation"""
288
class TestBadStatusServer(object):
289
"""Tests bad status from server."""
291
def create_transport_readonly_server(self):
292
return HttpServer(BadStatusRequestHandler)
294
def test_http_has(self):
295
server = self.get_readonly_server()
296
t = self._transport(server.get_url())
297
self.assertRaises(InvalidHttpResponse, t.has, 'foo/bar')
299
def test_http_get(self):
300
server = self.get_readonly_server()
301
t = self._transport(server.get_url())
302
self.assertRaises(InvalidHttpResponse, t.get, 'foo/bar')
305
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
306
"""Tests BadStatusServer for urllib implementation"""
308
_transport = HttpTransport_urllib
311
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
313
TestCaseWithWebserver):
314
"""Tests BadStatusServer for pycurl implementation"""
317
class TestInvalidStatusServer(TestBadStatusServer):
318
"""Tests invalid status from server.
320
Both implementations raises the same error as for a bad status.
323
def create_transport_readonly_server(self):
324
return HttpServer(InvalidStatusRequestHandler)
327
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
328
TestCaseWithWebserver):
329
"""Tests InvalidStatusServer for urllib implementation"""
331
_transport = HttpTransport_urllib
334
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
335
TestInvalidStatusServer,
336
TestCaseWithWebserver):
337
"""Tests InvalidStatusServer for pycurl implementation"""
340
class TestBadProtocolServer(object):
341
"""Tests bad status from server."""
343
def create_transport_readonly_server(self):
344
return HttpServer(BadProtocolRequestHandler)
346
def test_http_has(self):
347
server = self.get_readonly_server()
348
t = self._transport(server.get_url())
349
self.assertRaises(InvalidHttpResponse, t.has, 'foo/bar')
351
def test_http_get(self):
352
server = self.get_readonly_server()
353
t = self._transport(server.get_url())
354
self.assertRaises(InvalidHttpResponse, t.get, 'foo/bar')
357
class TestBadProtocolServer_urllib(TestBadProtocolServer,
358
TestCaseWithWebserver):
359
"""Tests BadProtocolServer for urllib implementation"""
361
_transport = HttpTransport_urllib
363
# curl don't check the protocol version
364
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
365
# TestBadProtocolServer,
366
# TestCaseWithWebserver):
367
# """Tests BadProtocolServer for pycurl implementation"""
370
class TestRangesServer(object):
371
"""Tests range requests against a server.
373
This MUST be used by daughter classes that also inherit from
374
TestCaseWithWebserver.
376
We can't inherit directly from TestCaseWithWebserver or the
377
test framework will try to create an instance which cannot
378
run, its implementation being incomplete.
382
TestCaseWithWebserver.setUp(self)
383
transport = self.get_transport()
384
if transport.is_readonly():
385
file('a', 'w').write('0123456789')
387
transport.put_bytes('a', '0123456789')
389
def test_single_range(self):
390
server = self.get_readonly_server()
391
t = self._transport(server.get_url())
392
content = t.readv(_file_10, )
394
self.assertRaises(errors.InvalidRange, out.read, 20)
397
self.assertEqual(_single_range_response[2], out.read(100))
399
def test_single_range_no_content(self):
400
out = self.get_response(_single_range_no_content_type)
401
self.assertIsInstance(out, response.HttpRangeResponse)
403
self.assertRaises(errors.InvalidRange, out.read, 20)
406
self.assertEqual(_single_range_no_content_type[2], out.read(100))
408
def test_multi_range(self):
409
out = self.get_response(_multipart_range_response)
410
self.assertIsInstance(out, response.HttpMultipartRangeResponse)
412
# Just make sure we can read the right contents
419
def test_multi_squid_range(self):
420
out = self.get_response(_multipart_squid_range_response)
421
self.assertIsInstance(out, response.HttpMultipartRangeResponse)
423
# Just make sure we can read the right contents
430
def test_full_text_no_content_type(self):
431
# We should not require Content-Type for a full response
432
a_response = _full_text_response
433
headers = http._extract_headers(a_response[1], 'http://foo')
434
del headers['Content-Type']
435
out = response.handle_response('http://foo', a_response[0], headers,
436
StringIO(a_response[2]))
437
self.assertEqual(_full_text_response[2], out.read())
439
def test_missing_content_range(self):
440
a_response = _single_range_response
441
headers = http._extract_headers(a_response[1], 'http://nocontent')
442
del headers['Content-Range']
443
self.assertRaises(errors.InvalidHttpResponse,
444
response.handle_response, 'http://nocontent', a_response[0],
445
headers, StringIO(a_response[2]))
added
removed
bzrlib/tests/test_http.py
