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.transport import (
38
from bzrlib.transport.http import (
40
BadProtocolRequestHandler,
41
BadStatusRequestHandler,
43
InvalidStatusRequestHandler,
46
from bzrlib.transport.http._urllib import HttpTransport_urllib
47
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
50
class FakeManager (object):
55
def add_password(self, realm, host, username, password):
56
self.credentials.append([realm, host, username, password])
59
class TestHttpUrls(TestCase):
61
def test_url_parsing(self):
63
url = extract_auth('http://example.com', f)
64
self.assertEquals('http://example.com', url)
65
self.assertEquals(0, len(f.credentials))
66
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
67
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
68
self.assertEquals(1, len(f.credentials))
69
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
72
def test_abs_url(self):
73
"""Construction of absolute http URLs"""
74
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
75
eq = self.assertEqualDiff
77
'http://bazaar-vcs.org/bzr/bzr.dev')
78
eq(t.abspath('foo/bar'),
79
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
81
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
82
eq(t.abspath('.bzr/1//2/./3'),
83
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
85
def test_invalid_http_urls(self):
86
"""Trap invalid construction of urls"""
87
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
88
self.assertRaises(ValueError,
92
def test_http_root_urls(self):
93
"""Construction of URLs from server root"""
94
t = HttpTransport_urllib('http://bzr.ozlabs.org/')
95
eq = self.assertEqualDiff
96
eq(t.abspath('.bzr/tree-version'),
97
'http://bzr.ozlabs.org/.bzr/tree-version')
99
def test_http_impl_urls(self):
100
"""There are servers which ask for particular clients to connect"""
102
from bzrlib.transport.http._pycurl import HttpServer_PyCurl
103
server = HttpServer_PyCurl()
106
url = server.get_url()
107
self.assertTrue(url.startswith('http+pycurl://'))
110
except DependencyNotPresent:
111
raise TestSkipped('pycurl not present')
114
class TestHttpConnections(object):
115
"""Test the http connections.
117
This MUST be used by daughter classes that also inherit from
118
TestCaseWithWebserver.
120
We can't inherit directly from TestCaseWithWebserver or the
121
test framework will try to create an instance which cannot
122
run, its implementation being incomplete.
126
TestCaseWithWebserver.setUp(self)
127
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
128
transport=self.get_transport())
130
def test_http_has(self):
131
server = self.get_readonly_server()
132
t = self._transport(server.get_url())
133
self.assertEqual(t.has('foo/bar'), True)
134
self.assertEqual(len(server.logs), 1)
135
self.assertContainsRe(server.logs[0],
136
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
138
def test_http_has_not_found(self):
139
server = self.get_readonly_server()
140
t = self._transport(server.get_url())
141
self.assertEqual(t.has('not-found'), False)
142
self.assertContainsRe(server.logs[1],
143
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
145
def test_http_get(self):
146
server = self.get_readonly_server()
147
t = self._transport(server.get_url())
148
fp = t.get('foo/bar')
149
self.assertEqualDiff(
151
'contents of foo/bar\n')
152
self.assertEqual(len(server.logs), 1)
153
self.assertTrue(server.logs[0].find(
154
'"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
155
% bzrlib.__version__) > -1)
157
def test_has_on_bogus_host(self):
158
# Get a free address and don't 'accept' on it, so that we
159
# can be sure there is no http handler there, but set a
160
# reasonable timeout to not slow down tests too much.
161
default_timeout = socket.getdefaulttimeout()
163
socket.setdefaulttimeout(2)
165
s.bind(('localhost', 0))
166
t = self._transport('http://%s:%s/' % s.getsockname())
167
self.assertRaises(ConnectionError, t.has, 'foo/bar')
169
socket.setdefaulttimeout(default_timeout)
172
class TestWithTransport_pycurl(object):
173
"""Test case to inherit from if pycurl is present"""
174
def _get_pycurl_maybe(self):
176
from bzrlib.transport.http._pycurl import PyCurlTransport
177
return PyCurlTransport
178
except DependencyNotPresent:
179
raise TestSkipped('pycurl not present')
181
_transport = property(_get_pycurl_maybe)
184
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
185
"""Test http connections with urllib"""
187
_transport = HttpTransport_urllib
191
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
193
TestCaseWithWebserver):
194
"""Test http connections with pycurl"""
197
class TestHttpTransportRegistration(TestCase):
198
"""Test registrations of various http implementations"""
200
def test_http_registered(self):
201
# urlllib should always be present
202
t = get_transport('http+urllib://bzr.google.com/')
203
self.assertIsInstance(t, Transport)
204
self.assertIsInstance(t, HttpTransport_urllib)
207
class TestOffsets(TestCase):
208
"""Test offsets_to_ranges method"""
210
def test_offsets_to_ranges_simple(self):
211
to_range = HttpTransportBase.offsets_to_ranges
212
ranges = to_range([(10, 1)])
213
self.assertEqual([[10, 10]], ranges)
215
ranges = to_range([(0, 1), (1, 1)])
216
self.assertEqual([[0, 1]], ranges)
218
ranges = to_range([(1, 1), (0, 1)])
219
self.assertEqual([[0, 1]], ranges)
221
def test_offset_to_ranges_overlapped(self):
222
to_range = HttpTransportBase.offsets_to_ranges
224
ranges = to_range([(10, 1), (20, 2), (22, 5)])
225
self.assertEqual([[10, 10], [20, 26]], ranges)
227
ranges = to_range([(10, 1), (11, 2), (22, 5)])
228
self.assertEqual([[10, 12], [22, 26]], ranges)
231
class TestRangeHeader(TestCase):
232
"""Test range_header method"""
234
def check_header(self, value, ranges=[], tail=0):
235
range_header = HttpTransportBase.range_header
236
self.assertEqual(value, range_header(ranges, tail))
238
def test_range_header_single(self):
239
self.check_header('0-9', ranges=[[0,9]])
240
self.check_header('100-109', ranges=[[100,109]])
242
def test_range_header_tail(self):
243
self.check_header('-10', tail=10)
244
self.check_header('-50', tail=50)
246
def test_range_header_multi(self):
247
self.check_header('0-9,100-200,300-5000',
248
ranges=[(0,9), (100, 200), (300,5000)])
250
def test_range_header_mixed(self):
251
self.check_header('0-9,300-5000,-50',
252
ranges=[(0,9), (300,5000)],
256
class TestWallServer(object):
257
"""Tests exceptions during the connection phase"""
259
def create_transport_readonly_server(self):
260
return bzrlib.transport.http.HttpServer(WallRequestHandler)
262
def test_http_has(self):
263
server = self.get_readonly_server()
264
t = self._transport(server.get_url())
265
self.assertRaises(ConnectionError, t.has, 'foo/bar')
267
def test_http_get(self):
268
server = self.get_readonly_server()
269
t = self._transport(server.get_url())
270
self.assertRaises(ConnectionError, t.get, 'foo/bar')
273
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
274
"""Tests WallServer for urllib implementation"""
276
_transport = HttpTransport_urllib
279
class TestWallServer_pycurl(TestWithTransport_pycurl,
281
TestCaseWithWebserver):
282
"""Tests WallServer for pycurl implementation"""
285
class TestBadStatusServer(object):
286
"""Tests bad status from server."""
288
def create_transport_readonly_server(self):
289
return bzrlib.transport.http.HttpServer(BadStatusRequestHandler)
291
def test_http_has(self):
292
server = self.get_readonly_server()
293
t = self._transport(server.get_url())
294
self.assertRaises(InvalidHttpResponse, t.has, 'foo/bar')
296
def test_http_get(self):
297
server = self.get_readonly_server()
298
t = self._transport(server.get_url())
299
self.assertRaises(InvalidHttpResponse, t.get, 'foo/bar')
302
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
303
"""Tests BadStatusServer for urllib implementation"""
305
_transport = HttpTransport_urllib
308
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
310
TestCaseWithWebserver):
311
"""Tests BadStatusServer for pycurl implementation"""
314
class TestInvalidStatusServer(TestBadStatusServer):
315
"""Tests invalid status from server.
317
Both implementations raises the same error as for a bad status.
320
def create_transport_readonly_server(self):
321
return bzrlib.transport.http.HttpServer(InvalidStatusRequestHandler)
324
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
325
TestCaseWithWebserver):
326
"""Tests InvalidStatusServer for urllib implementation"""
328
_transport = HttpTransport_urllib
331
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
332
TestInvalidStatusServer,
333
TestCaseWithWebserver):
334
"""Tests InvalidStatusServer for pycurl implementation"""
337
class TestBadProtocolServer(object):
338
"""Tests bad status from server."""
340
def create_transport_readonly_server(self):
341
return bzrlib.transport.http.HttpServer(BadProtocolRequestHandler)
343
def test_http_has(self):
344
server = self.get_readonly_server()
345
t = self._transport(server.get_url())
346
self.assertRaises(InvalidHttpResponse, t.has, 'foo/bar')
348
def test_http_get(self):
349
server = self.get_readonly_server()
350
t = self._transport(server.get_url())
351
self.assertRaises(InvalidHttpResponse, t.get, 'foo/bar')
354
class TestBadProtocolServer_urllib(TestBadProtocolServer,
355
TestCaseWithWebserver):
356
"""Tests BadProtocolServer for urllib implementation"""
358
_transport = HttpTransport_urllib
360
# curl don't check the protocol version
361
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
362
# TestBadProtocolServer,
363
# TestCaseWithWebserver):
364
# """Tests BadProtocolServer for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
