1
# Copyright (C) 2004, 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
21
from cStringIO import StringIO
24
from bzrlib import urlutils
25
from bzrlib.errors import (
34
from bzrlib.tests import TestCase, TestCaseInTempDir
35
from bzrlib.transport import (_CoalescedOffset,
36
_get_protocol_handlers,
37
_get_transport_modules,
39
register_lazy_transport,
40
_set_protocol_handlers,
43
from bzrlib.transport.memory import MemoryTransport
44
from bzrlib.transport.local import LocalTransport
47
# TODO: Should possibly split transport-specific tests into their own files.
50
class TestTransport(TestCase):
51
"""Test the non transport-concrete class functionality."""
53
def test__get_set_protocol_handlers(self):
54
handlers = _get_protocol_handlers()
55
self.assertNotEqual({}, handlers)
57
_set_protocol_handlers({})
58
self.assertEqual({}, _get_protocol_handlers())
60
_set_protocol_handlers(handlers)
62
def test_get_transport_modules(self):
63
handlers = _get_protocol_handlers()
64
class SampleHandler(object):
65
"""I exist, isnt that enough?"""
68
_set_protocol_handlers(my_handlers)
69
register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
70
register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
71
self.assertEqual([SampleHandler.__module__],
72
_get_transport_modules())
74
_set_protocol_handlers(handlers)
76
def test_transport_dependency(self):
77
"""Transport with missing dependency causes no error"""
78
saved_handlers = _get_protocol_handlers()
80
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
81
'BadTransportHandler')
83
get_transport('foo://fooserver/foo')
84
except UnsupportedProtocol, e:
86
self.assertEquals('Unsupported protocol'
87
' for url "foo://fooserver/foo":'
88
' Unable to import library "some_lib":'
89
' testing missing dependency', str(e))
91
self.fail('Did not raise UnsupportedProtocol')
93
# restore original values
94
_set_protocol_handlers(saved_handlers)
96
def test_transport_fallback(self):
97
"""Transport with missing dependency causes no error"""
98
saved_handlers = _get_protocol_handlers()
100
_set_protocol_handlers({})
101
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
102
'BackupTransportHandler')
103
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
104
'BadTransportHandler')
105
t = get_transport('foo://fooserver/foo')
106
self.assertTrue(isinstance(t, BackupTransportHandler))
108
_set_protocol_handlers(saved_handlers)
110
def test__combine_paths(self):
112
self.assertEqual('/home/sarah/project/foo',
113
t._combine_paths('/home/sarah', 'project/foo'))
114
self.assertEqual('/etc',
115
t._combine_paths('/home/sarah', '../../etc'))
116
self.assertEqual('/etc',
117
t._combine_paths('/home/sarah', '../../../etc'))
118
self.assertEqual('/etc',
119
t._combine_paths('/home/sarah', '/etc'))
122
class TestCoalesceOffsets(TestCase):
124
def check(self, expected, offsets, limit=0, fudge=0):
125
coalesce = Transport._coalesce_offsets
126
exp = [_CoalescedOffset(*x) for x in expected]
127
out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
128
self.assertEqual(exp, out)
130
def test_coalesce_empty(self):
133
def test_coalesce_simple(self):
134
self.check([(0, 10, [(0, 10)])], [(0, 10)])
136
def test_coalesce_unrelated(self):
137
self.check([(0, 10, [(0, 10)]),
139
], [(0, 10), (20, 10)])
141
def test_coalesce_unsorted(self):
142
self.check([(20, 10, [(0, 10)]),
144
], [(20, 10), (0, 10)])
146
def test_coalesce_nearby(self):
147
self.check([(0, 20, [(0, 10), (10, 10)])],
150
def test_coalesce_overlapped(self):
151
self.check([(0, 15, [(0, 10), (5, 10)])],
154
def test_coalesce_limit(self):
155
self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
156
(30, 10), (40, 10)]),
157
(60, 50, [(0, 10), (10, 10), (20, 10),
158
(30, 10), (40, 10)]),
159
], [(10, 10), (20, 10), (30, 10), (40, 10),
160
(50, 10), (60, 10), (70, 10), (80, 10),
161
(90, 10), (100, 10)],
164
def test_coalesce_no_limit(self):
165
self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
166
(30, 10), (40, 10), (50, 10),
167
(60, 10), (70, 10), (80, 10),
169
], [(10, 10), (20, 10), (30, 10), (40, 10),
170
(50, 10), (60, 10), (70, 10), (80, 10),
171
(90, 10), (100, 10)])
173
def test_coalesce_fudge(self):
174
self.check([(10, 30, [(0, 10), (20, 10)]),
175
(100, 10, [(0, 10),]),
176
], [(10, 10), (30, 10), (100, 10)],
181
class TestMemoryTransport(TestCase):
183
def test_get_transport(self):
186
def test_clone(self):
187
transport = MemoryTransport()
188
self.assertTrue(isinstance(transport, MemoryTransport))
189
self.assertEqual("memory:///", transport.clone("/").base)
191
def test_abspath(self):
192
transport = MemoryTransport()
193
self.assertEqual("memory:///relpath", transport.abspath('relpath'))
195
def test_abspath_of_root(self):
196
transport = MemoryTransport()
197
self.assertEqual("memory:///", transport.base)
198
self.assertEqual("memory:///", transport.abspath('/'))
200
def test_abspath_of_relpath_starting_at_root(self):
201
transport = MemoryTransport()
202
self.assertEqual("memory:///foo", transport.abspath('/foo'))
204
def test_append_and_get(self):
205
transport = MemoryTransport()
206
transport.append_bytes('path', 'content')
207
self.assertEqual(transport.get('path').read(), 'content')
208
transport.append_file('path', StringIO('content'))
209
self.assertEqual(transport.get('path').read(), 'contentcontent')
211
def test_put_and_get(self):
212
transport = MemoryTransport()
213
transport.put_file('path', StringIO('content'))
214
self.assertEqual(transport.get('path').read(), 'content')
215
transport.put_bytes('path', 'content')
216
self.assertEqual(transport.get('path').read(), 'content')
218
def test_append_without_dir_fails(self):
219
transport = MemoryTransport()
220
self.assertRaises(NoSuchFile,
221
transport.append_bytes, 'dir/path', 'content')
223
def test_put_without_dir_fails(self):
224
transport = MemoryTransport()
225
self.assertRaises(NoSuchFile,
226
transport.put_file, 'dir/path', StringIO('content'))
228
def test_get_missing(self):
229
transport = MemoryTransport()
230
self.assertRaises(NoSuchFile, transport.get, 'foo')
232
def test_has_missing(self):
233
transport = MemoryTransport()
234
self.assertEquals(False, transport.has('foo'))
236
def test_has_present(self):
237
transport = MemoryTransport()
238
transport.append_bytes('foo', 'content')
239
self.assertEquals(True, transport.has('foo'))
241
def test_mkdir(self):
242
transport = MemoryTransport()
243
transport.mkdir('dir')
244
transport.append_bytes('dir/path', 'content')
245
self.assertEqual(transport.get('dir/path').read(), 'content')
247
def test_mkdir_missing_parent(self):
248
transport = MemoryTransport()
249
self.assertRaises(NoSuchFile,
250
transport.mkdir, 'dir/dir')
252
def test_mkdir_twice(self):
253
transport = MemoryTransport()
254
transport.mkdir('dir')
255
self.assertRaises(FileExists, transport.mkdir, 'dir')
257
def test_parameters(self):
258
transport = MemoryTransport()
259
self.assertEqual(True, transport.listable())
260
self.assertEqual(False, transport.should_cache())
261
self.assertEqual(False, transport.is_readonly())
263
def test_iter_files_recursive(self):
264
transport = MemoryTransport()
265
transport.mkdir('dir')
266
transport.put_bytes('dir/foo', 'content')
267
transport.put_bytes('dir/bar', 'content')
268
transport.put_bytes('bar', 'content')
269
paths = set(transport.iter_files_recursive())
270
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
273
transport = MemoryTransport()
274
transport.put_bytes('foo', 'content')
275
transport.put_bytes('bar', 'phowar')
276
self.assertEqual(7, transport.stat('foo').st_size)
277
self.assertEqual(6, transport.stat('bar').st_size)
280
class ChrootDecoratorTransportTest(TestCase):
281
"""Chroot decoration specific tests."""
283
def test_construct(self):
284
from bzrlib.transport import chroot
285
transport = chroot.ChrootTransportDecorator('chroot+memory:///pathA/')
286
self.assertEqual('memory:///pathA/', transport.chroot_url)
288
transport = chroot.ChrootTransportDecorator(
289
'chroot+memory:///path/B', chroot='memory:///path/')
290
self.assertEqual('memory:///path/', transport.chroot_url)
292
def test_append_file(self):
293
transport = get_transport('chroot+file:///foo/bar')
294
self.assertRaises(PathNotChild, transport.append_file, '/foo', None)
296
def test_append_bytes(self):
297
transport = get_transport('chroot+file:///foo/bar')
298
self.assertRaises(PathNotChild, transport.append_bytes, '/foo', 'bytes')
300
def test_clone(self):
301
transport = get_transport('chroot+file:///foo/bar')
302
self.assertRaises(PathNotChild, transport.clone, '/foo')
304
def test_delete(self):
305
transport = get_transport('chroot+file:///foo/bar')
306
self.assertRaises(PathNotChild, transport.delete, '/foo')
308
def test_delete_tree(self):
309
transport = get_transport('chroot+file:///foo/bar')
310
self.assertRaises(PathNotChild, transport.delete_tree, '/foo')
313
transport = get_transport('chroot+file:///foo/bar')
314
self.assertRaises(PathNotChild, transport.get, '/foo')
316
def test_get_bytes(self):
317
transport = get_transport('chroot+file:///foo/bar')
318
self.assertRaises(PathNotChild, transport.get_bytes, '/foo')
321
transport = get_transport('chroot+file:///foo/bar')
322
self.assertRaises(PathNotChild, transport.has, '/foo')
324
def test_list_dir(self):
325
transport = get_transport('chroot+file:///foo/bar')
326
self.assertRaises(PathNotChild, transport.list_dir, '/foo')
328
def test_lock_read(self):
329
transport = get_transport('chroot+file:///foo/bar')
330
self.assertRaises(PathNotChild, transport.lock_read, '/foo')
332
def test_lock_write(self):
333
transport = get_transport('chroot+file:///foo/bar')
334
self.assertRaises(PathNotChild, transport.lock_write, '/foo')
336
def test_mkdir(self):
337
transport = get_transport('chroot+file:///foo/bar')
338
self.assertRaises(PathNotChild, transport.mkdir, '/foo')
340
def test_put_bytes(self):
341
transport = get_transport('chroot+file:///foo/bar')
342
self.assertRaises(PathNotChild, transport.put_bytes, '/foo', 'bytes')
344
def test_put_file(self):
345
transport = get_transport('chroot+file:///foo/bar')
346
self.assertRaises(PathNotChild, transport.put_file, '/foo', None)
348
def test_rename(self):
349
transport = get_transport('chroot+file:///foo/bar')
350
self.assertRaises(PathNotChild, transport.rename, '/aaa', 'bbb')
351
self.assertRaises(PathNotChild, transport.rename, 'ccc', '/d')
353
def test_rmdir(self):
354
transport = get_transport('chroot+file:///foo/bar')
355
self.assertRaises(PathNotChild, transport.rmdir, '/foo')
358
transport = get_transport('chroot+file:///foo/bar')
359
self.assertRaises(PathNotChild, transport.stat, '/foo')
362
class ReadonlyDecoratorTransportTest(TestCase):
363
"""Readonly decoration specific tests."""
365
def test_local_parameters(self):
366
import bzrlib.transport.readonly as readonly
367
# connect to . in readonly mode
368
transport = readonly.ReadonlyTransportDecorator('readonly+.')
369
self.assertEqual(True, transport.listable())
370
self.assertEqual(False, transport.should_cache())
371
self.assertEqual(True, transport.is_readonly())
373
def test_http_parameters(self):
374
import bzrlib.transport.readonly as readonly
375
from bzrlib.transport.http import HttpServer
376
# connect to . via http which is not listable
377
server = HttpServer()
380
transport = get_transport('readonly+' + server.get_url())
381
self.failUnless(isinstance(transport,
382
readonly.ReadonlyTransportDecorator))
383
self.assertEqual(False, transport.listable())
384
self.assertEqual(True, transport.should_cache())
385
self.assertEqual(True, transport.is_readonly())
390
class FakeNFSDecoratorTests(TestCaseInTempDir):
391
"""NFS decorator specific tests."""
393
def get_nfs_transport(self, url):
394
import bzrlib.transport.fakenfs as fakenfs
395
# connect to url with nfs decoration
396
return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
398
def test_local_parameters(self):
399
# the listable, should_cache and is_readonly parameters
400
# are not changed by the fakenfs decorator
401
transport = self.get_nfs_transport('.')
402
self.assertEqual(True, transport.listable())
403
self.assertEqual(False, transport.should_cache())
404
self.assertEqual(False, transport.is_readonly())
406
def test_http_parameters(self):
407
# the listable, should_cache and is_readonly parameters
408
# are not changed by the fakenfs decorator
409
from bzrlib.transport.http import HttpServer
410
# connect to . via http which is not listable
411
server = HttpServer()
414
transport = self.get_nfs_transport(server.get_url())
415
self.assertIsInstance(
416
transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
417
self.assertEqual(False, transport.listable())
418
self.assertEqual(True, transport.should_cache())
419
self.assertEqual(True, transport.is_readonly())
423
def test_fakenfs_server_default(self):
424
# a FakeNFSServer() should bring up a local relpath server for itself
425
import bzrlib.transport.fakenfs as fakenfs
426
server = fakenfs.FakeNFSServer()
429
# the url should be decorated appropriately
430
self.assertStartsWith(server.get_url(), 'fakenfs+')
431
# and we should be able to get a transport for it
432
transport = get_transport(server.get_url())
433
# which must be a FakeNFSTransportDecorator instance.
434
self.assertIsInstance(
435
transport, fakenfs.FakeNFSTransportDecorator)
439
def test_fakenfs_rename_semantics(self):
440
# a FakeNFS transport must mangle the way rename errors occur to
441
# look like NFS problems.
442
transport = self.get_nfs_transport('.')
443
self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
445
self.assertRaises(bzrlib.errors.ResourceBusy,
446
transport.rename, 'from', 'to')
449
class FakeVFATDecoratorTests(TestCaseInTempDir):
450
"""Tests for simulation of VFAT restrictions"""
452
def get_vfat_transport(self, url):
453
"""Return vfat-backed transport for test directory"""
454
from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
455
return FakeVFATTransportDecorator('vfat+' + url)
457
def test_transport_creation(self):
458
from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
459
transport = self.get_vfat_transport('.')
460
self.assertIsInstance(transport, FakeVFATTransportDecorator)
462
def test_transport_mkdir(self):
463
transport = self.get_vfat_transport('.')
464
transport.mkdir('HELLO')
465
self.assertTrue(transport.has('hello'))
466
self.assertTrue(transport.has('Hello'))
468
def test_forbidden_chars(self):
469
transport = self.get_vfat_transport('.')
470
self.assertRaises(ValueError, transport.has, "<NU>")
473
class BadTransportHandler(Transport):
474
def __init__(self, base_url):
475
raise DependencyNotPresent('some_lib', 'testing missing dependency')
478
class BackupTransportHandler(Transport):
479
"""Test transport that works as a backup for the BadTransportHandler"""
483
class TestTransportImplementation(TestCaseInTempDir):
484
"""Implementation verification for transports.
486
To verify a transport we need a server factory, which is a callable
487
that accepts no parameters and returns an implementation of
488
bzrlib.transport.Server.
490
That Server is then used to construct transport instances and test
491
the transport via loopback activity.
493
Currently this assumes that the Transport object is connected to the
494
current working directory. So that whatever is done
495
through the transport, should show up in the working
496
directory, and vice-versa. This is a bug, because its possible to have
497
URL schemes which provide access to something that may not be
498
result in storage on the local disk, i.e. due to file system limits, or
499
due to it being a database or some other non-filesystem tool.
501
This also tests to make sure that the functions work with both
502
generators and lists (assuming iter(list) is effectively a generator)
506
super(TestTransportImplementation, self).setUp()
507
self._server = self.transport_server()
511
super(TestTransportImplementation, self).tearDown()
512
self._server.tearDown()
514
def get_transport(self):
515
"""Return a connected transport to the local directory."""
516
base_url = self._server.get_url()
517
# try getting the transport via the regular interface:
518
t = get_transport(base_url)
519
if not isinstance(t, self.transport_class):
520
# we did not get the correct transport class type. Override the
521
# regular connection behaviour by direct construction.
522
t = self.transport_class(base_url)
526
class TestLocalTransports(TestCase):
528
def test_get_transport_from_abspath(self):
529
here = os.path.abspath('.')
530
t = get_transport(here)
531
self.assertIsInstance(t, LocalTransport)
532
self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
534
def test_get_transport_from_relpath(self):
535
here = os.path.abspath('.')
536
t = get_transport('.')
537
self.assertIsInstance(t, LocalTransport)
538
self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
540
def test_get_transport_from_local_url(self):
541
here = os.path.abspath('.')
542
here_url = urlutils.local_path_to_url(here) + '/'
543
t = get_transport(here_url)
544
self.assertIsInstance(t, LocalTransport)
545
self.assertEquals(t.base, here_url)
added
removed
bzrlib/tests/test_transport.py
