13
13
# You should have received a copy of the GNU General Public License
14
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
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18
18
from cStringIO import StringIO
21
24
from bzrlib import (
26
from bzrlib.errors import (DependencyNotPresent,
34
from bzrlib.tests import TestCase, TestCaseInTempDir
35
from bzrlib.transport import (_clear_protocol_handlers,
38
_get_protocol_handlers,
39
_set_protocol_handlers,
40
_get_transport_modules,
43
register_lazy_transport,
44
register_transport_proto,
47
from bzrlib.transport.chroot import ChrootServer
48
from bzrlib.transport.memory import MemoryTransport
49
from bzrlib.transport.local import (LocalTransport,
50
EmulatedWin32LocalTransport)
31
from bzrlib.transport import (
39
from bzrlib.tests import (
53
45
# TODO: Should possibly split transport-specific tests into their own files.
56
class TestTransport(TestCase):
48
class TestTransport(tests.TestCase):
57
49
"""Test the non transport-concrete class functionality."""
51
# FIXME: These tests should use addCleanup() and/or overrideAttr() instead
52
# of try/finally -- vila 20100205
59
54
def test__get_set_protocol_handlers(self):
60
handlers = _get_protocol_handlers()
55
handlers = transport._get_protocol_handlers()
61
56
self.assertNotEqual([], handlers.keys( ))
63
_clear_protocol_handlers()
64
self.assertEqual([], _get_protocol_handlers().keys())
58
transport._clear_protocol_handlers()
59
self.assertEqual([], transport._get_protocol_handlers().keys())
66
_set_protocol_handlers(handlers)
61
transport._set_protocol_handlers(handlers)
68
63
def test_get_transport_modules(self):
69
handlers = _get_protocol_handlers()
64
handlers = transport._get_protocol_handlers()
70
65
# don't pollute the current handlers
71
_clear_protocol_handlers()
66
transport._clear_protocol_handlers()
72
67
class SampleHandler(object):
73
68
"""I exist, isnt that enough?"""
75
_clear_protocol_handlers()
76
register_transport_proto('foo')
77
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
78
'TestTransport.SampleHandler')
79
register_transport_proto('bar')
80
register_lazy_transport('bar', 'bzrlib.tests.test_transport',
81
'TestTransport.SampleHandler')
70
transport._clear_protocol_handlers()
71
transport.register_transport_proto('foo')
72
transport.register_lazy_transport('foo',
73
'bzrlib.tests.test_transport',
74
'TestTransport.SampleHandler')
75
transport.register_transport_proto('bar')
76
transport.register_lazy_transport('bar',
77
'bzrlib.tests.test_transport',
78
'TestTransport.SampleHandler')
82
79
self.assertEqual([SampleHandler.__module__,
83
'bzrlib.transport.chroot'],
84
_get_transport_modules())
80
'bzrlib.transport.chroot',
81
'bzrlib.transport.pathfilter'],
82
transport._get_transport_modules())
86
_set_protocol_handlers(handlers)
84
transport._set_protocol_handlers(handlers)
88
86
def test_transport_dependency(self):
89
87
"""Transport with missing dependency causes no error"""
90
saved_handlers = _get_protocol_handlers()
88
saved_handlers = transport._get_protocol_handlers()
91
89
# don't pollute the current handlers
92
_clear_protocol_handlers()
90
transport._clear_protocol_handlers()
94
register_transport_proto('foo')
95
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
96
'BadTransportHandler')
92
transport.register_transport_proto('foo')
93
transport.register_lazy_transport(
94
'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
98
get_transport('foo://fooserver/foo')
99
except UnsupportedProtocol, e:
96
transport.get_transport('foo://fooserver/foo')
97
except errors.UnsupportedProtocol, e:
101
99
self.assertEquals('Unsupported protocol'
102
100
' for url "foo://fooserver/foo":'
106
104
self.fail('Did not raise UnsupportedProtocol')
108
106
# restore original values
109
_set_protocol_handlers(saved_handlers)
107
transport._set_protocol_handlers(saved_handlers)
111
109
def test_transport_fallback(self):
112
110
"""Transport with missing dependency causes no error"""
113
saved_handlers = _get_protocol_handlers()
111
saved_handlers = transport._get_protocol_handlers()
115
_clear_protocol_handlers()
116
register_transport_proto('foo')
117
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
118
'BackupTransportHandler')
119
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
120
'BadTransportHandler')
121
t = get_transport('foo://fooserver/foo')
113
transport._clear_protocol_handlers()
114
transport.register_transport_proto('foo')
115
transport.register_lazy_transport(
116
'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
117
transport.register_lazy_transport(
118
'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
119
t = transport.get_transport('foo://fooserver/foo')
122
120
self.assertTrue(isinstance(t, BackupTransportHandler))
124
_set_protocol_handlers(saved_handlers)
122
transport._set_protocol_handlers(saved_handlers)
124
def test_ssh_hints(self):
125
"""Transport ssh:// should raise an error pointing out bzr+ssh://"""
127
transport.get_transport('ssh://fooserver/foo')
128
except errors.UnsupportedProtocol, e:
130
self.assertEquals('Unsupported protocol'
131
' for url "ssh://fooserver/foo":'
132
' bzr supports bzr+ssh to operate over ssh,'
133
' use "bzr+ssh://fooserver/foo".',
136
self.fail('Did not raise UnsupportedProtocol')
126
138
def test_LateReadError(self):
127
139
"""The LateReadError helper should raise on read()."""
128
a_file = LateReadError('a path')
140
a_file = transport.LateReadError('a path')
131
except ReadError, error:
143
except errors.ReadError, error:
132
144
self.assertEqual('a path', error.path)
133
self.assertRaises(ReadError, a_file.read, 40)
145
self.assertRaises(errors.ReadError, a_file.read, 40)
136
148
def test__combine_paths(self):
149
t = transport.Transport('/')
138
150
self.assertEqual('/home/sarah/project/foo',
139
151
t._combine_paths('/home/sarah', 'project/foo'))
140
152
self.assertEqual('/etc',
225
237
[(10, 10), (20, 10), (30, 50), (80, 100)],
229
class TestMemoryTransport(TestCase):
240
def test_coalesce_default_limit(self):
241
# By default we use a 100MB max size.
242
ten_mb = 10*1024*1024
243
self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
244
(10*ten_mb, ten_mb, [(0, ten_mb)])],
245
[(i*ten_mb, ten_mb) for i in range(11)])
246
self.check([(0, 11*ten_mb, [(i*ten_mb, ten_mb) for i in range(11)]),],
247
[(i*ten_mb, ten_mb) for i in range(11)],
248
max_size=1*1024*1024*1024)
251
class TestMemoryServer(tests.TestCase):
253
def test_create_server(self):
254
server = memory.MemoryServer()
255
server.start_server()
256
url = server.get_url()
257
self.assertTrue(url in transport.transport_list_registry)
258
t = transport.get_transport(url)
261
self.assertFalse(url in transport.transport_list_registry)
262
self.assertRaises(errors.UnsupportedProtocol,
263
transport.get_transport, url)
266
class TestMemoryTransport(tests.TestCase):
231
268
def test_get_transport(self):
269
memory.MemoryTransport()
234
271
def test_clone(self):
235
transport = MemoryTransport()
236
self.assertTrue(isinstance(transport, MemoryTransport))
237
self.assertEqual("memory:///", transport.clone("/").base)
272
t = memory.MemoryTransport()
273
self.assertTrue(isinstance(t, memory.MemoryTransport))
274
self.assertEqual("memory:///", t.clone("/").base)
239
276
def test_abspath(self):
240
transport = MemoryTransport()
241
self.assertEqual("memory:///relpath", transport.abspath('relpath'))
277
t = memory.MemoryTransport()
278
self.assertEqual("memory:///relpath", t.abspath('relpath'))
243
280
def test_abspath_of_root(self):
244
transport = MemoryTransport()
245
self.assertEqual("memory:///", transport.base)
246
self.assertEqual("memory:///", transport.abspath('/'))
281
t = memory.MemoryTransport()
282
self.assertEqual("memory:///", t.base)
283
self.assertEqual("memory:///", t.abspath('/'))
248
285
def test_abspath_of_relpath_starting_at_root(self):
249
transport = MemoryTransport()
250
self.assertEqual("memory:///foo", transport.abspath('/foo'))
286
t = memory.MemoryTransport()
287
self.assertEqual("memory:///foo", t.abspath('/foo'))
252
289
def test_append_and_get(self):
253
transport = MemoryTransport()
254
transport.append_bytes('path', 'content')
255
self.assertEqual(transport.get('path').read(), 'content')
256
transport.append_file('path', StringIO('content'))
257
self.assertEqual(transport.get('path').read(), 'contentcontent')
290
t = memory.MemoryTransport()
291
t.append_bytes('path', 'content')
292
self.assertEqual(t.get('path').read(), 'content')
293
t.append_file('path', StringIO('content'))
294
self.assertEqual(t.get('path').read(), 'contentcontent')
259
296
def test_put_and_get(self):
260
transport = MemoryTransport()
261
transport.put_file('path', StringIO('content'))
262
self.assertEqual(transport.get('path').read(), 'content')
263
transport.put_bytes('path', 'content')
264
self.assertEqual(transport.get('path').read(), 'content')
297
t = memory.MemoryTransport()
298
t.put_file('path', StringIO('content'))
299
self.assertEqual(t.get('path').read(), 'content')
300
t.put_bytes('path', 'content')
301
self.assertEqual(t.get('path').read(), 'content')
266
303
def test_append_without_dir_fails(self):
267
transport = MemoryTransport()
268
self.assertRaises(NoSuchFile,
269
transport.append_bytes, 'dir/path', 'content')
304
t = memory.MemoryTransport()
305
self.assertRaises(errors.NoSuchFile,
306
t.append_bytes, 'dir/path', 'content')
271
308
def test_put_without_dir_fails(self):
272
transport = MemoryTransport()
273
self.assertRaises(NoSuchFile,
274
transport.put_file, 'dir/path', StringIO('content'))
309
t = memory.MemoryTransport()
310
self.assertRaises(errors.NoSuchFile,
311
t.put_file, 'dir/path', StringIO('content'))
276
313
def test_get_missing(self):
277
transport = MemoryTransport()
278
self.assertRaises(NoSuchFile, transport.get, 'foo')
314
transport = memory.MemoryTransport()
315
self.assertRaises(errors.NoSuchFile, transport.get, 'foo')
280
317
def test_has_missing(self):
281
transport = MemoryTransport()
282
self.assertEquals(False, transport.has('foo'))
318
t = memory.MemoryTransport()
319
self.assertEquals(False, t.has('foo'))
284
321
def test_has_present(self):
285
transport = MemoryTransport()
286
transport.append_bytes('foo', 'content')
287
self.assertEquals(True, transport.has('foo'))
322
t = memory.MemoryTransport()
323
t.append_bytes('foo', 'content')
324
self.assertEquals(True, t.has('foo'))
289
326
def test_list_dir(self):
290
transport = MemoryTransport()
291
transport.put_bytes('foo', 'content')
292
transport.mkdir('dir')
293
transport.put_bytes('dir/subfoo', 'content')
294
transport.put_bytes('dirlike', 'content')
327
t = memory.MemoryTransport()
328
t.put_bytes('foo', 'content')
330
t.put_bytes('dir/subfoo', 'content')
331
t.put_bytes('dirlike', 'content')
296
self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
297
self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
333
self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
334
self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))
299
336
def test_mkdir(self):
300
transport = MemoryTransport()
301
transport.mkdir('dir')
302
transport.append_bytes('dir/path', 'content')
303
self.assertEqual(transport.get('dir/path').read(), 'content')
337
t = memory.MemoryTransport()
339
t.append_bytes('dir/path', 'content')
340
self.assertEqual(t.get('dir/path').read(), 'content')
305
342
def test_mkdir_missing_parent(self):
306
transport = MemoryTransport()
307
self.assertRaises(NoSuchFile,
308
transport.mkdir, 'dir/dir')
343
t = memory.MemoryTransport()
344
self.assertRaises(errors.NoSuchFile, t.mkdir, 'dir/dir')
310
346
def test_mkdir_twice(self):
311
transport = MemoryTransport()
312
transport.mkdir('dir')
313
self.assertRaises(FileExists, transport.mkdir, 'dir')
347
t = memory.MemoryTransport()
349
self.assertRaises(errors.FileExists, t.mkdir, 'dir')
315
351
def test_parameters(self):
316
transport = MemoryTransport()
317
self.assertEqual(True, transport.listable())
318
self.assertEqual(False, transport.is_readonly())
352
t = memory.MemoryTransport()
353
self.assertEqual(True, t.listable())
354
self.assertEqual(False, t.is_readonly())
320
356
def test_iter_files_recursive(self):
321
transport = MemoryTransport()
322
transport.mkdir('dir')
323
transport.put_bytes('dir/foo', 'content')
324
transport.put_bytes('dir/bar', 'content')
325
transport.put_bytes('bar', 'content')
326
paths = set(transport.iter_files_recursive())
357
t = memory.MemoryTransport()
359
t.put_bytes('dir/foo', 'content')
360
t.put_bytes('dir/bar', 'content')
361
t.put_bytes('bar', 'content')
362
paths = set(t.iter_files_recursive())
327
363
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
329
365
def test_stat(self):
330
transport = MemoryTransport()
331
transport.put_bytes('foo', 'content')
332
transport.put_bytes('bar', 'phowar')
333
self.assertEqual(7, transport.stat('foo').st_size)
334
self.assertEqual(6, transport.stat('bar').st_size)
337
class ChrootDecoratorTransportTest(TestCase):
366
t = memory.MemoryTransport()
367
t.put_bytes('foo', 'content')
368
t.put_bytes('bar', 'phowar')
369
self.assertEqual(7, t.stat('foo').st_size)
370
self.assertEqual(6, t.stat('bar').st_size)
373
class ChrootDecoratorTransportTest(tests.TestCase):
338
374
"""Chroot decoration specific tests."""
340
376
def test_abspath(self):
341
377
# The abspath is always relative to the chroot_url.
342
server = ChrootServer(get_transport('memory:///foo/bar/'))
344
transport = get_transport(server.get_url())
345
self.assertEqual(server.get_url(), transport.abspath('/'))
378
server = chroot.ChrootServer(
379
transport.get_transport('memory:///foo/bar/'))
380
self.start_server(server)
381
t = transport.get_transport(server.get_url())
382
self.assertEqual(server.get_url(), t.abspath('/'))
347
subdir_transport = transport.clone('subdir')
348
self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
384
subdir_t = t.clone('subdir')
385
self.assertEqual(server.get_url(), subdir_t.abspath('/'))
351
387
def test_clone(self):
352
server = ChrootServer(get_transport('memory:///foo/bar/'))
354
transport = get_transport(server.get_url())
388
server = chroot.ChrootServer(
389
transport.get_transport('memory:///foo/bar/'))
390
self.start_server(server)
391
t = transport.get_transport(server.get_url())
355
392
# relpath from root and root path are the same
356
relpath_cloned = transport.clone('foo')
357
abspath_cloned = transport.clone('/foo')
393
relpath_cloned = t.clone('foo')
394
abspath_cloned = t.clone('/foo')
358
395
self.assertEqual(server, relpath_cloned.server)
359
396
self.assertEqual(server, abspath_cloned.server)
362
398
def test_chroot_url_preserves_chroot(self):
363
399
"""Calling get_transport on a chroot transport's base should produce a
364
400
transport with exactly the same behaviour as the original chroot
384
420
This is so that it is not possible to escape a chroot by doing::
385
421
url = chroot_transport.base
386
422
parent_url = urlutils.join(url, '..')
387
new_transport = get_transport(parent_url)
423
new_t = transport.get_transport(parent_url)
389
server = ChrootServer(get_transport('memory:///path/'))
391
transport = get_transport(server.get_url())
425
server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
426
self.start_server(server)
427
t = transport.get_transport(server.get_url())
392
428
self.assertRaises(
393
InvalidURLJoin, urlutils.join, transport.base, '..')
397
class ChrootServerTest(TestCase):
429
errors.InvalidURLJoin, urlutils.join, t.base, '..')
432
class TestChrootServer(tests.TestCase):
399
434
def test_construct(self):
400
backing_transport = MemoryTransport()
401
server = ChrootServer(backing_transport)
435
backing_transport = memory.MemoryTransport()
436
server = chroot.ChrootServer(backing_transport)
402
437
self.assertEqual(backing_transport, server.backing_transport)
404
439
def test_setUp(self):
405
backing_transport = MemoryTransport()
406
server = ChrootServer(backing_transport)
408
self.assertTrue(server.scheme in _get_protocol_handlers().keys())
440
backing_transport = memory.MemoryTransport()
441
server = chroot.ChrootServer(backing_transport)
442
server.start_server()
444
self.assertTrue(server.scheme
445
in transport._get_protocol_handlers().keys())
410
def test_tearDown(self):
411
backing_transport = MemoryTransport()
412
server = ChrootServer(backing_transport)
415
self.assertFalse(server.scheme in _get_protocol_handlers().keys())
449
def test_stop_server(self):
450
backing_transport = memory.MemoryTransport()
451
server = chroot.ChrootServer(backing_transport)
452
server.start_server()
454
self.assertFalse(server.scheme
455
in transport._get_protocol_handlers().keys())
417
457
def test_get_url(self):
418
backing_transport = MemoryTransport()
419
server = ChrootServer(backing_transport)
421
self.assertEqual('chroot-%d:///' % id(server), server.get_url())
425
class ReadonlyDecoratorTransportTest(TestCase):
458
backing_transport = memory.MemoryTransport()
459
server = chroot.ChrootServer(backing_transport)
460
server.start_server()
462
self.assertEqual('chroot-%d:///' % id(server), server.get_url())
467
class PathFilteringDecoratorTransportTest(tests.TestCase):
468
"""Pathfilter decoration specific tests."""
470
def test_abspath(self):
471
# The abspath is always relative to the base of the backing transport.
472
server = pathfilter.PathFilteringServer(
473
transport.get_transport('memory:///foo/bar/'),
475
server.start_server()
476
t = transport.get_transport(server.get_url())
477
self.assertEqual(server.get_url(), t.abspath('/'))
479
subdir_t = t.clone('subdir')
480
self.assertEqual(server.get_url(), subdir_t.abspath('/'))
483
def make_pf_transport(self, filter_func=None):
484
"""Make a PathFilteringTransport backed by a MemoryTransport.
486
:param filter_func: by default this will be a no-op function. Use this
487
parameter to override it."""
488
if filter_func is None:
489
filter_func = lambda x: x
490
server = pathfilter.PathFilteringServer(
491
transport.get_transport('memory:///foo/bar/'), filter_func)
492
server.start_server()
493
self.addCleanup(server.stop_server)
494
return transport.get_transport(server.get_url())
496
def test__filter(self):
497
# _filter (with an identity func as filter_func) always returns
498
# paths relative to the base of the backing transport.
499
t = self.make_pf_transport()
500
self.assertEqual('foo', t._filter('foo'))
501
self.assertEqual('foo/bar', t._filter('foo/bar'))
502
self.assertEqual('', t._filter('..'))
503
self.assertEqual('', t._filter('/'))
504
# The base of the pathfiltering transport is taken into account too.
505
t = t.clone('subdir1/subdir2')
506
self.assertEqual('subdir1/subdir2/foo', t._filter('foo'))
507
self.assertEqual('subdir1/subdir2/foo/bar', t._filter('foo/bar'))
508
self.assertEqual('subdir1', t._filter('..'))
509
self.assertEqual('', t._filter('/'))
511
def test_filter_invocation(self):
514
filter_log.append(path)
516
t = self.make_pf_transport(filter)
518
self.assertEqual(['abc'], filter_log)
520
t.clone('abc').has('xyz')
521
self.assertEqual(['abc/xyz'], filter_log)
524
self.assertEqual(['abc'], filter_log)
526
def test_clone(self):
527
t = self.make_pf_transport()
528
# relpath from root and root path are the same
529
relpath_cloned = t.clone('foo')
530
abspath_cloned = t.clone('/foo')
531
self.assertEqual(t.server, relpath_cloned.server)
532
self.assertEqual(t.server, abspath_cloned.server)
534
def test_url_preserves_pathfiltering(self):
535
"""Calling get_transport on a pathfiltered transport's base should
536
produce a transport with exactly the same behaviour as the original
537
pathfiltered transport.
539
This is so that it is not possible to escape (accidentally or
540
otherwise) the filtering by doing::
541
url = filtered_transport.base
542
parent_url = urlutils.join(url, '..')
543
new_t = transport.get_transport(parent_url)
545
t = self.make_pf_transport()
546
new_t = transport.get_transport(t.base)
547
self.assertEqual(t.server, new_t.server)
548
self.assertEqual(t.base, new_t.base)
551
class ReadonlyDecoratorTransportTest(tests.TestCase):
426
552
"""Readonly decoration specific tests."""
428
554
def test_local_parameters(self):
429
import bzrlib.transport.readonly as readonly
430
555
# connect to . in readonly mode
431
transport = readonly.ReadonlyTransportDecorator('readonly+.')
432
self.assertEqual(True, transport.listable())
433
self.assertEqual(True, transport.is_readonly())
556
t = readonly.ReadonlyTransportDecorator('readonly+.')
557
self.assertEqual(True, t.listable())
558
self.assertEqual(True, t.is_readonly())
435
560
def test_http_parameters(self):
436
561
from bzrlib.tests.http_server import HttpServer
437
import bzrlib.transport.readonly as readonly
438
562
# connect to '.' via http which is not listable
439
563
server = HttpServer()
442
transport = get_transport('readonly+' + server.get_url())
443
self.failUnless(isinstance(transport,
444
readonly.ReadonlyTransportDecorator))
445
self.assertEqual(False, transport.listable())
446
self.assertEqual(True, transport.is_readonly())
451
class FakeNFSDecoratorTests(TestCaseInTempDir):
564
self.start_server(server)
565
t = transport.get_transport('readonly+' + server.get_url())
566
self.failUnless(isinstance(t, readonly.ReadonlyTransportDecorator))
567
self.assertEqual(False, t.listable())
568
self.assertEqual(True, t.is_readonly())
571
class FakeNFSDecoratorTests(tests.TestCaseInTempDir):
452
572
"""NFS decorator specific tests."""
454
574
def get_nfs_transport(self, url):
455
import bzrlib.transport.fakenfs as fakenfs
456
575
# connect to url with nfs decoration
457
576
return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
459
578
def test_local_parameters(self):
460
579
# the listable and is_readonly parameters
461
580
# are not changed by the fakenfs decorator
462
transport = self.get_nfs_transport('.')
463
self.assertEqual(True, transport.listable())
464
self.assertEqual(False, transport.is_readonly())
581
t = self.get_nfs_transport('.')
582
self.assertEqual(True, t.listable())
583
self.assertEqual(False, t.is_readonly())
466
585
def test_http_parameters(self):
467
586
# the listable and is_readonly parameters
469
588
from bzrlib.tests.http_server import HttpServer
470
589
# connect to '.' via http which is not listable
471
590
server = HttpServer()
474
transport = self.get_nfs_transport(server.get_url())
475
self.assertIsInstance(
476
transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
477
self.assertEqual(False, transport.listable())
478
self.assertEqual(True, transport.is_readonly())
591
self.start_server(server)
592
t = self.get_nfs_transport(server.get_url())
593
self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
594
self.assertEqual(False, t.listable())
595
self.assertEqual(True, t.is_readonly())
482
597
def test_fakenfs_server_default(self):
483
598
# a FakeNFSServer() should bring up a local relpath server for itself
484
import bzrlib.transport.fakenfs as fakenfs
485
server = fakenfs.FakeNFSServer()
488
# the url should be decorated appropriately
489
self.assertStartsWith(server.get_url(), 'fakenfs+')
490
# and we should be able to get a transport for it
491
transport = get_transport(server.get_url())
492
# which must be a FakeNFSTransportDecorator instance.
493
self.assertIsInstance(
494
transport, fakenfs.FakeNFSTransportDecorator)
599
server = test_server.FakeNFSServer()
600
self.start_server(server)
601
# the url should be decorated appropriately
602
self.assertStartsWith(server.get_url(), 'fakenfs+')
603
# and we should be able to get a transport for it
604
t = transport.get_transport(server.get_url())
605
# which must be a FakeNFSTransportDecorator instance.
606
self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
498
608
def test_fakenfs_rename_semantics(self):
499
609
# a FakeNFS transport must mangle the way rename errors occur to
500
610
# look like NFS problems.
501
transport = self.get_nfs_transport('.')
611
t = self.get_nfs_transport('.')
502
612
self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
504
self.assertRaises(errors.ResourceBusy,
505
transport.rename, 'from', 'to')
508
class FakeVFATDecoratorTests(TestCaseInTempDir):
614
self.assertRaises(errors.ResourceBusy, t.rename, 'from', 'to')
617
class FakeVFATDecoratorTests(tests.TestCaseInTempDir):
509
618
"""Tests for simulation of VFAT restrictions"""
511
620
def get_vfat_transport(self, url):
516
625
def test_transport_creation(self):
517
626
from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
518
transport = self.get_vfat_transport('.')
519
self.assertIsInstance(transport, FakeVFATTransportDecorator)
627
t = self.get_vfat_transport('.')
628
self.assertIsInstance(t, FakeVFATTransportDecorator)
521
630
def test_transport_mkdir(self):
522
transport = self.get_vfat_transport('.')
523
transport.mkdir('HELLO')
524
self.assertTrue(transport.has('hello'))
525
self.assertTrue(transport.has('Hello'))
631
t = self.get_vfat_transport('.')
633
self.assertTrue(t.has('hello'))
634
self.assertTrue(t.has('Hello'))
527
636
def test_forbidden_chars(self):
528
transport = self.get_vfat_transport('.')
529
self.assertRaises(ValueError, transport.has, "<NU>")
532
class BadTransportHandler(Transport):
637
t = self.get_vfat_transport('.')
638
self.assertRaises(ValueError, t.has, "<NU>")
641
class BadTransportHandler(transport.Transport):
533
642
def __init__(self, base_url):
534
raise DependencyNotPresent('some_lib', 'testing missing dependency')
537
class BackupTransportHandler(Transport):
643
raise errors.DependencyNotPresent('some_lib',
644
'testing missing dependency')
647
class BackupTransportHandler(transport.Transport):
538
648
"""Test transport that works as a backup for the BadTransportHandler"""
542
class TestTransportImplementation(TestCaseInTempDir):
652
class TestTransportImplementation(tests.TestCaseInTempDir):
543
653
"""Implementation verification for transports.
545
655
To verify a transport we need a server factory, which is a callable
546
656
that accepts no parameters and returns an implementation of
547
657
bzrlib.transport.Server.
549
659
That Server is then used to construct transport instances and test
550
660
the transport via loopback activity.
552
Currently this assumes that the Transport object is connected to the
553
current working directory. So that whatever is done
554
through the transport, should show up in the working
662
Currently this assumes that the Transport object is connected to the
663
current working directory. So that whatever is done
664
through the transport, should show up in the working
555
665
directory, and vice-versa. This is a bug, because its possible to have
556
URL schemes which provide access to something that may not be
557
result in storage on the local disk, i.e. due to file system limits, or
666
URL schemes which provide access to something that may not be
667
result in storage on the local disk, i.e. due to file system limits, or
558
668
due to it being a database or some other non-filesystem tool.
560
670
This also tests to make sure that the functions work with both
561
671
generators and lists (assuming iter(list) is effectively a generator)
565
675
super(TestTransportImplementation, self).setUp()
566
676
self._server = self.transport_server()
568
self.addCleanup(self._server.tearDown)
677
self.start_server(self._server)
570
679
def get_transport(self, relpath=None):
571
680
"""Return a connected transport to the local directory.
710
821
self.assertIs(new_password, c._get_credentials())
713
class TestReusedTransports(TestCase):
824
class TestReusedTransports(tests.TestCase):
714
825
"""Tests for transport reuse"""
716
827
def test_reuse_same_transport(self):
717
828
possible_transports = []
718
t1 = get_transport('http://foo/',
719
possible_transports=possible_transports)
829
t1 = transport.get_transport('http://foo/',
830
possible_transports=possible_transports)
720
831
self.assertEqual([t1], possible_transports)
721
t2 = get_transport('http://foo/', possible_transports=[t1])
832
t2 = transport.get_transport('http://foo/',
833
possible_transports=[t1])
722
834
self.assertIs(t1, t2)
724
836
# Also check that final '/' are handled correctly
725
t3 = get_transport('http://foo/path/')
726
t4 = get_transport('http://foo/path', possible_transports=[t3])
837
t3 = transport.get_transport('http://foo/path/')
838
t4 = transport.get_transport('http://foo/path',
839
possible_transports=[t3])
727
840
self.assertIs(t3, t4)
729
t5 = get_transport('http://foo/path')
730
t6 = get_transport('http://foo/path/', possible_transports=[t5])
842
t5 = transport.get_transport('http://foo/path')
843
t6 = transport.get_transport('http://foo/path/',
844
possible_transports=[t5])
731
845
self.assertIs(t5, t6)
733
847
def test_don_t_reuse_different_transport(self):
734
t1 = get_transport('http://foo/path')
735
t2 = get_transport('http://bar/path', possible_transports=[t1])
848
t1 = transport.get_transport('http://foo/path')
849
t2 = transport.get_transport('http://bar/path',
850
possible_transports=[t1])
736
851
self.assertIsNot(t1, t2)
739
class TestTransportTrace(TestCase):
854
class TestTransportTrace(tests.TestCase):
741
856
def test_get(self):
742
transport = get_transport('trace+memory://')
743
self.assertIsInstance(
744
transport, bzrlib.transport.trace.TransportTraceDecorator)
857
t = transport.get_transport('trace+memory://')
858
self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
746
860
def test_clone_preserves_activity(self):
747
transport = get_transport('trace+memory://')
748
transport2 = transport.clone('.')
749
self.assertTrue(transport is not transport2)
750
self.assertTrue(transport._activity is transport2._activity)
861
t = transport.get_transport('trace+memory://')
863
self.assertTrue(t is not t2)
864
self.assertTrue(t._activity is t2._activity)
752
866
# the following specific tests are for the operations that have made use of
753
867
# logging in tests; we could test every single operation but doing that
754
868
# still won't cause a test failure when the top level Transport API
755
869
# changes; so there is little return doing that.
756
870
def test_get(self):
757
transport = get_transport('trace+memory:///')
758
transport.put_bytes('foo', 'barish')
871
t = transport.get_transport('trace+memory:///')
872
t.put_bytes('foo', 'barish')
760
874
expected_result = []
761
875
# put_bytes records the bytes, not the content to avoid memory
763
877
expected_result.append(('put_bytes', 'foo', 6, None))
764
878
# get records the file name only.
765
879
expected_result.append(('get', 'foo'))
766
self.assertEqual(expected_result, transport._activity)
880
self.assertEqual(expected_result, t._activity)
768
882
def test_readv(self):
769
transport = get_transport('trace+memory:///')
770
transport.put_bytes('foo', 'barish')
771
list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
883
t = transport.get_transport('trace+memory:///')
884
t.put_bytes('foo', 'barish')
885
list(t.readv('foo', [(0, 1), (3, 2)],
886
adjust_for_latency=True, upper_limit=6))
773
887
expected_result = []
774
888
# put_bytes records the bytes, not the content to avoid memory
776
890
expected_result.append(('put_bytes', 'foo', 6, None))
777
891
# readv records the supplied offset request
778
892
expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
779
self.assertEqual(expected_result, transport._activity)
893
self.assertEqual(expected_result, t._activity)
896
class TestSSHConnections(tests.TestCaseWithTransport):
898
def test_bzr_connect_to_bzr_ssh(self):
899
"""User acceptance that get_transport of a bzr+ssh:// behaves correctly.
901
bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
903
# This test actually causes a bzr instance to be invoked, which is very
904
# expensive: it should be the only such test in the test suite.
905
# A reasonable evolution for this would be to simply check inside
906
# check_channel_exec_request that the command is appropriate, and then
907
# satisfy requests in-process.
908
self.requireFeature(features.paramiko)
909
# SFTPFullAbsoluteServer has a get_url method, and doesn't
910
# override the interface (doesn't change self._vendor).
911
# Note that this does encryption, so can be slow.
912
from bzrlib.tests import stub_sftp
914
# Start an SSH server
915
self.command_executed = []
916
# XXX: This is horrible -- we define a really dumb SSH server that
917
# executes commands, and manage the hooking up of stdin/out/err to the
918
# SSH channel ourselves. Surely this has already been implemented
921
class StubSSHServer(stub_sftp.StubServer):
925
def check_channel_exec_request(self, channel, command):
926
self.test.command_executed.append(command)
927
proc = subprocess.Popen(
928
command, shell=True, stdin=subprocess.PIPE,
929
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
931
# XXX: horribly inefficient, not to mention ugly.
932
# Start a thread for each of stdin/out/err, and relay bytes from
933
# the subprocess to channel and vice versa.
934
def ferry_bytes(read, write, close):
943
(channel.recv, proc.stdin.write, proc.stdin.close),
944
(proc.stdout.read, channel.sendall, channel.close),
945
(proc.stderr.read, channel.sendall_stderr, channel.close)]
947
for read, write, close in file_functions:
948
t = threading.Thread(
949
target=ferry_bytes, args=(read, write, close))
955
ssh_server = stub_sftp.SFTPFullAbsoluteServer(StubSSHServer)
956
# We *don't* want to override the default SSH vendor: the detected one
958
self.start_server(ssh_server)
959
port = ssh_server._listener.port
961
if sys.platform == 'win32':
962
bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
964
bzr_remote_path = self.get_bzr_path()
965
os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
967
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
968
# variable is used to tell bzr what command to run on the remote end.
969
path_to_branch = osutils.abspath('.')
970
if sys.platform == 'win32':
971
# On Windows, we export all drives as '/C:/, etc. So we need to
972
# prefix a '/' to get the right path.
973
path_to_branch = '/' + path_to_branch
974
url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
975
t = transport.get_transport(url)
976
self.permit_url(t.base)
980
['%s serve --inet --directory=/ --allow-writes' % bzr_remote_path],
981
self.command_executed)
982
# Make sure to disconnect, so that the remote process can stop, and we
983
# can cleanup. Then pause the test until everything is shutdown
984
t._client._medium.disconnect()
987
# First wait for the subprocess
989
# And the rest are threads
990
for t in started[1:]:
added
removed
bzrlib/tests/test_transport.py
