20
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
28
from bzrlib.errors import DependencyNotPresent, UnsupportedProtocol
29
from bzrlib.tests import TestCase, TestSkipped
30
from bzrlib.transport import get_transport, Transport
31
from bzrlib.transport.http import extract_auth, HttpTransportBase
27
from bzrlib import errors
28
from bzrlib import osutils
29
from bzrlib.tests import (
33
from bzrlib.tests.HttpServer import (
38
from bzrlib.tests.HTTPTestUtil import (
39
BadProtocolRequestHandler,
40
BadStatusRequestHandler,
41
ForbiddenRequestHandler,
42
InvalidStatusRequestHandler,
43
NoRangeRequestHandler,
44
SingleRangeRequestHandler,
45
TestCaseWithWebserver,
48
from bzrlib.transport import (
52
from bzrlib.transport.http import (
32
56
from bzrlib.transport.http._urllib import HttpTransport_urllib
33
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
36
59
class FakeManager(object):
38
61
def __init__(self):
39
62
self.credentials = []
41
64
def add_password(self, realm, host, username, password):
42
65
self.credentials.append([realm, host, username, password])
106
132
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
107
133
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
108
134
self.assertEquals(1, len(f.credentials))
109
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'], f.credentials[0])
135
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
111
138
def test_abs_url(self):
112
139
"""Construction of absolute http URLs"""
113
140
t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
114
141
eq = self.assertEqualDiff
115
142
eq(t.abspath('.'),
116
143
'http://bazaar-vcs.org/bzr/bzr.dev')
117
eq(t.abspath('foo/bar'),
144
eq(t.abspath('foo/bar'),
118
145
'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
119
146
eq(t.abspath('.bzr'),
120
147
'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
138
167
def test_http_impl_urls(self):
139
168
"""There are servers which ask for particular clients to connect"""
169
server = HttpServer_PyCurl()
141
from bzrlib.transport.http._pycurl import HttpServer_PyCurl
142
server = HttpServer_PyCurl()
145
url = server.get_url()
146
self.assertTrue(url.startswith('http+pycurl://'))
149
except DependencyNotPresent:
150
raise TestSkipped('pycurl not present')
153
class TestHttpMixins(object):
155
def _prep_tree(self):
172
url = server.get_url()
173
self.assertTrue(url.startswith('http+pycurl://'))
178
class TestHttpConnections(object):
179
"""Test the http connections.
181
This MUST be used by daughter classes that also inherit from
182
TestCaseWithWebserver.
184
We can't inherit directly from TestCaseWithWebserver or the
185
test framework will try to create an instance which cannot
186
run, its implementation being incomplete.
190
TestCaseWithWebserver.setUp(self)
156
191
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
157
192
transport=self.get_transport())
161
196
t = self._transport(server.get_url())
162
197
self.assertEqual(t.has('foo/bar'), True)
163
198
self.assertEqual(len(server.logs), 1)
164
self.assertContainsRe(server.logs[0],
199
self.assertContainsRe(server.logs[0],
165
200
r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
167
202
def test_http_has_not_found(self):
168
203
server = self.get_readonly_server()
169
204
t = self._transport(server.get_url())
170
205
self.assertEqual(t.has('not-found'), False)
171
self.assertContainsRe(server.logs[1],
206
self.assertContainsRe(server.logs[1],
172
207
r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
174
209
def test_http_get(self):
188
224
http_transport = self._transport(server.get_url())
189
225
medium = http_transport.get_smart_medium()
190
226
self.assertIs(medium, http_transport)
193
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
195
_transport = HttpTransport_urllib
198
TestCaseWithWebserver.setUp(self)
201
228
def test_has_on_bogus_host(self):
203
# Get a random address, so that we can be sure there is no
204
# http handler there.
206
s.bind(('localhost', 0))
207
t = self._transport('http://%s:%s/' % s.getsockname())
208
self.assertRaises(urllib2.URLError, t.has, 'foo/bar')
211
class TestHttpConnections_pycurl(TestCaseWithWebserver, TestHttpMixins):
229
# Get a free address and don't 'accept' on it, so that we
230
# can be sure there is no http handler there, but set a
231
# reasonable timeout to not slow down tests too much.
232
default_timeout = socket.getdefaulttimeout()
234
socket.setdefaulttimeout(2)
236
s.bind(('localhost', 0))
237
t = self._transport('http://%s:%s/' % s.getsockname())
238
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
240
socket.setdefaulttimeout(default_timeout)
243
class TestWithTransport_pycurl(object):
244
"""Test case to inherit from if pycurl is present"""
213
246
def _get_pycurl_maybe(self):
215
248
from bzrlib.transport.http._pycurl import PyCurlTransport
216
249
return PyCurlTransport
217
except DependencyNotPresent:
250
except errors.DependencyNotPresent:
218
251
raise TestSkipped('pycurl not present')
220
253
_transport = property(_get_pycurl_maybe)
223
TestCaseWithWebserver.setUp(self)
256
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
257
"""Test http connections with urllib"""
259
_transport = HttpTransport_urllib
263
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
265
TestCaseWithWebserver):
266
"""Test http connections with pycurl"""
227
269
class TestHttpTransportRegistration(TestCase):
228
270
"""Test registrations of various http implementations"""
230
272
def test_http_registered(self):
231
import bzrlib.transport.http._urllib
232
from bzrlib.transport import get_transport
233
273
# urlllib should always be present
234
274
t = get_transport('http+urllib://bzr.google.com/')
235
275
self.assertIsInstance(t, Transport)
236
self.assertIsInstance(t, bzrlib.transport.http._urllib.HttpTransport_urllib)
276
self.assertIsInstance(t, HttpTransport_urllib)
239
279
class TestOffsets(TestCase):
312
352
ranges=[(0,9), (300,5000)],
356
class TestWallServer(object):
357
"""Tests exceptions during the connection phase"""
359
def create_transport_readonly_server(self):
360
return HttpServer(WallRequestHandler)
362
def test_http_has(self):
363
server = self.get_readonly_server()
364
t = self._transport(server.get_url())
365
# Unfortunately httplib (see HTTPResponse._read_status
366
# for details) make no distinction between a closed
367
# socket and badly formatted status line, so we can't
368
# just test for ConnectionError, we have to test
369
# InvalidHttpResponse too.
370
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
373
def test_http_get(self):
374
server = self.get_readonly_server()
375
t = self._transport(server.get_url())
376
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
380
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
381
"""Tests "wall" server for urllib implementation"""
383
_transport = HttpTransport_urllib
386
class TestWallServer_pycurl(TestWithTransport_pycurl,
388
TestCaseWithWebserver):
389
"""Tests "wall" server for pycurl implementation"""
392
class TestBadStatusServer(object):
393
"""Tests bad status from server."""
395
def create_transport_readonly_server(self):
396
return HttpServer(BadStatusRequestHandler)
398
def test_http_has(self):
399
server = self.get_readonly_server()
400
t = self._transport(server.get_url())
401
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
403
def test_http_get(self):
404
server = self.get_readonly_server()
405
t = self._transport(server.get_url())
406
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
409
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
410
"""Tests bad status server for urllib implementation"""
412
_transport = HttpTransport_urllib
415
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
417
TestCaseWithWebserver):
418
"""Tests bad status server for pycurl implementation"""
421
class TestInvalidStatusServer(TestBadStatusServer):
422
"""Tests invalid status from server.
424
Both implementations raises the same error as for a bad status.
427
def create_transport_readonly_server(self):
428
return HttpServer(InvalidStatusRequestHandler)
431
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
432
TestCaseWithWebserver):
433
"""Tests invalid status server for urllib implementation"""
435
_transport = HttpTransport_urllib
438
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
439
TestInvalidStatusServer,
440
TestCaseWithWebserver):
441
"""Tests invalid status server for pycurl implementation"""
444
class TestBadProtocolServer(object):
445
"""Tests bad protocol from server."""
447
def create_transport_readonly_server(self):
448
return HttpServer(BadProtocolRequestHandler)
450
def test_http_has(self):
451
server = self.get_readonly_server()
452
t = self._transport(server.get_url())
453
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
455
def test_http_get(self):
456
server = self.get_readonly_server()
457
t = self._transport(server.get_url())
458
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
461
class TestBadProtocolServer_urllib(TestBadProtocolServer,
462
TestCaseWithWebserver):
463
"""Tests bad protocol server for urllib implementation"""
465
_transport = HttpTransport_urllib
467
# curl don't check the protocol version
468
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
469
# TestBadProtocolServer,
470
# TestCaseWithWebserver):
471
# """Tests bad protocol server for pycurl implementation"""
474
class TestForbiddenServer(object):
475
"""Tests forbidden server"""
477
def create_transport_readonly_server(self):
478
return HttpServer(ForbiddenRequestHandler)
480
def test_http_has(self):
481
server = self.get_readonly_server()
482
t = self._transport(server.get_url())
483
self.assertRaises(errors.TransportError, t.has, 'foo/bar')
485
def test_http_get(self):
486
server = self.get_readonly_server()
487
t = self._transport(server.get_url())
488
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
491
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
492
"""Tests forbidden server for urllib implementation"""
494
_transport = HttpTransport_urllib
497
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
499
TestCaseWithWebserver):
500
"""Tests forbidden server for pycurl implementation"""
316
503
class TestRecordingServer(TestCase):
318
505
def test_create(self):
340
527
sock.connect((server.host, server.port))
341
528
sock.sendall('abc')
342
529
self.assertEqual('HTTP/1.1 200 OK\r\n',
343
sock.recv(4096, socket.MSG_WAITALL))
530
osutils.recv_all(sock, 4096))
344
531
self.assertEqual('abc', server.received_bytes)
534
class TestRangeRequestServer(object):
535
"""Test the http connections.
537
This MUST be used by daughter classes that also inherit from
538
TestCaseWithWebserver.
540
We can't inherit directly from TestCaseWithWebserver or the
541
test framework will try to create an instance which cannot
542
run, its implementation being incomplete.
546
TestCaseWithWebserver.setUp(self)
547
self.build_tree_contents([('a', '0123456789')],)
549
"""Tests readv requests against server"""
551
def test_readv(self):
552
server = self.get_readonly_server()
553
t = self._transport(server.get_url())
554
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
555
self.assertEqual(l[0], (0, '0'))
556
self.assertEqual(l[1], (1, '1'))
557
self.assertEqual(l[2], (3, '34'))
558
self.assertEqual(l[3], (9, '9'))
560
def test_readv_out_of_order(self):
561
server = self.get_readonly_server()
562
t = self._transport(server.get_url())
563
l = list(t.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
564
self.assertEqual(l[0], (1, '1'))
565
self.assertEqual(l[1], (9, '9'))
566
self.assertEqual(l[2], (0, '0'))
567
self.assertEqual(l[3], (3, '34'))
569
def test_readv_short_read(self):
570
server = self.get_readonly_server()
571
t = self._transport(server.get_url())
573
# This is intentionally reading off the end of the file
574
# since we are sure that it cannot get there
575
self.assertListRaises((errors.ShortReadvError, AssertionError),
576
t.readv, 'a', [(1,1), (8,10)])
578
# This is trying to seek past the end of the file, it should
579
# also raise a special error
580
self.assertListRaises(errors.ShortReadvError,
581
t.readv, 'a', [(12,2)])
584
class TestSingleRangeRequestServer(TestRangeRequestServer):
585
"""Test readv against a server which accept only single range requests"""
587
def create_transport_readonly_server(self):
588
return HttpServer(SingleRangeRequestHandler)
591
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
592
TestCaseWithWebserver):
593
"""Tests single range requests accepting server for urllib implementation"""
595
_transport = HttpTransport_urllib
598
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
599
TestSingleRangeRequestServer,
600
TestCaseWithWebserver):
601
"""Tests single range requests accepting server for pycurl implementation"""
604
class TestNoRangeRequestServer(TestRangeRequestServer):
605
"""Test readv against a server which do not accept range requests"""
607
def create_transport_readonly_server(self):
608
return HttpServer(NoRangeRequestHandler)
611
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
612
TestCaseWithWebserver):
613
"""Tests range requests refusing server for urllib implementation"""
615
_transport = HttpTransport_urllib
618
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
619
TestNoRangeRequestServer,
620
TestCaseWithWebserver):
621
"""Tests range requests refusing server for pycurl implementation"""
added
removed
bzrlib/tests/test_http.py
