1
# Copyright (C) 2004, 2005, 2006 by 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
17
"""Tests for Transport implementations.
19
Transport implementations tested here are supplied by
20
TransportTestProviderAdapter.
24
from cStringIO import StringIO
28
from bzrlib.errors import (DirectoryNotEmpty, NoSuchFile, FileExists,
31
TransportNotPossible, ConnectionError)
32
from bzrlib.tests import TestCaseInTempDir, TestSkipped
33
from bzrlib.transport import memory, urlescape
34
import bzrlib.transport
38
"""Append the given text (file-like object) to the supplied filename."""
46
class TestTransportImplementation(TestCaseInTempDir):
47
"""Implementation verification for transports.
49
To verify a transport we need a server factory, which is a callable
50
that accepts no parameters and returns an implementation of
51
bzrlib.transport.Server.
53
That Server is then used to construct transport instances and test
54
the transport via loopback activity.
56
Currently this assumes that the Transport object is connected to the
57
current working directory. So that whatever is done
58
through the transport, should show up in the working
59
directory, and vice-versa. This is a bug, because its possible to have
60
URL schemes which provide access to something that may not be
61
result in storage on the local disk, i.e. due to file system limits, or
62
due to it being a database or some other non-filesystem tool.
64
This also tests to make sure that the functions work with both
65
generators and lists (assuming iter(list) is effectively a generator)
69
super(TestTransportImplementation, self).setUp()
70
self._server = self.transport_server()
74
super(TestTransportImplementation, self).tearDown()
75
self._server.tearDown()
77
def check_transport_contents(self, content, transport, relpath):
78
"""Check that transport.get(relpath).read() == content."""
79
self.assertEqualDiff(content, transport.get(relpath).read())
81
def get_transport(self):
82
"""Return a connected transport to the local directory."""
83
t = bzrlib.transport.get_transport(self._server.get_url())
84
self.failUnless(isinstance(t, self.transport_class),
85
"Got the wrong class from get_transport"
86
"(%r, expected %r)" % (t.__class__,
87
self.transport_class))
90
def assertListRaises(self, excClass, func, *args, **kwargs):
91
"""Fail unless excClass is raised when the iterator from func is used.
93
Many transport functions can return generators this makes sure
94
to wrap them in a list() call to make sure the whole generator
95
is run, and that the proper exception is raised.
98
list(func(*args, **kwargs))
102
if hasattr(excClass,'__name__'): excName = excClass.__name__
103
else: excName = str(excClass)
104
raise self.failureException, "%s not raised" % excName
107
t = self.get_transport()
109
files = ['a', 'b', 'e', 'g', '%']
110
self.build_tree(files, transport=t)
111
self.assertEqual(True, t.has('a'))
112
self.assertEqual(False, t.has('c'))
113
self.assertEqual(True, t.has(urlescape('%')))
114
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
115
[True, True, False, False, True, False, True, False])
116
self.assertEqual(True, t.has_any(['a', 'b', 'c']))
117
self.assertEqual(False, t.has_any(['c', 'd', 'f', urlescape('%%')]))
118
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
119
[True, True, False, False, True, False, True, False])
120
self.assertEqual(False, t.has_any(['c', 'c', 'c']))
121
self.assertEqual(True, t.has_any(['b', 'b', 'b']))
124
t = self.get_transport()
126
files = ['a', 'b', 'e', 'g']
127
contents = ['contents of a\n',
132
self.build_tree(files, transport=t)
133
self.check_transport_contents('contents of a\n', t, 'a')
134
content_f = t.get_multi(files)
135
for content, f in zip(contents, content_f):
136
self.assertEqual(content, f.read())
138
content_f = t.get_multi(iter(files))
139
for content, f in zip(contents, content_f):
140
self.assertEqual(content, f.read())
142
self.assertRaises(NoSuchFile, t.get, 'c')
143
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
144
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
147
t = self.get_transport()
150
self.assertRaises(TransportNotPossible,
151
t.put, 'a', 'some text for a\n')
154
t.put('a', StringIO('some text for a\n'))
155
self.failUnless(t.has('a'))
156
self.check_transport_contents('some text for a\n', t, 'a')
157
# Make sure 'has' is updated
158
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
159
[True, False, False, False, False])
160
# Put also replaces contents
161
self.assertEqual(t.put_multi([('a', StringIO('new\ncontents for\na\n')),
162
('d', StringIO('contents\nfor d\n'))]),
164
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
165
[True, False, False, True, False])
166
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
167
self.check_transport_contents('contents\nfor d\n', t, 'd')
170
t.put_multi(iter([('a', StringIO('diff\ncontents for\na\n')),
171
('d', StringIO('another contents\nfor d\n'))])),
173
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
174
self.check_transport_contents('another contents\nfor d\n', t, 'd')
176
self.assertRaises(NoSuchFile,
177
t.put, 'path/doesnt/exist/c', 'contents')
179
def test_put_permissions(self):
180
t = self.get_transport()
184
t.put('mode644', StringIO('test text\n'), mode=0644)
185
self.assertTransportMode(t, 'mode644', 0644)
186
t.put('mode666', StringIO('test text\n'), mode=0666)
187
self.assertTransportMode(t, 'mode666', 0666)
188
t.put('mode600', StringIO('test text\n'), mode=0600)
189
self.assertTransportMode(t, 'mode600', 0600)
190
# Yes, you can put a file such that it becomes readonly
191
t.put('mode400', StringIO('test text\n'), mode=0400)
192
self.assertTransportMode(t, 'mode400', 0400)
193
t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
194
self.assertTransportMode(t, 'mmode644', 0644)
196
def test_mkdir(self):
197
t = self.get_transport()
200
# cannot mkdir on readonly transports. We're not testing for
201
# cache coherency because cache behaviour is not currently
202
# defined for the transport interface.
203
self.assertRaises(TransportNotPossible, t.mkdir, '.')
204
self.assertRaises(TransportNotPossible, t.mkdir, 'new_dir')
205
self.assertRaises(TransportNotPossible, t.mkdir_multi, ['new_dir'])
206
self.assertRaises(TransportNotPossible, t.mkdir, 'path/doesnt/exist')
210
self.assertEqual(t.has('dir_a'), True)
211
self.assertEqual(t.has('dir_b'), False)
214
self.assertEqual(t.has('dir_b'), True)
216
t.mkdir_multi(['dir_c', 'dir_d'])
218
t.mkdir_multi(iter(['dir_e', 'dir_f']))
219
self.assertEqual(list(t.has_multi(
220
['dir_a', 'dir_b', 'dir_c', 'dir_q',
221
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
222
[True, True, True, False,
223
True, True, True, True])
225
# we were testing that a local mkdir followed by a transport
226
# mkdir failed thusly, but given that we * in one process * do not
227
# concurrently fiddle with disk dirs and then use transport to do
228
# things, the win here seems marginal compared to the constraint on
229
# the interface. RBC 20051227
231
self.assertRaises(FileExists, t.mkdir, 'dir_g')
233
# Test get/put in sub-directories
235
t.put_multi([('dir_a/a', StringIO('contents of dir_a/a')),
236
('dir_b/b', StringIO('contents of dir_b/b'))])
238
self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
239
self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
241
# mkdir of a dir with an absent parent
242
self.assertRaises(NoSuchFile, t.mkdir, 'missing/dir')
244
def test_mkdir_permissions(self):
245
t = self.get_transport()
248
# Test mkdir with a mode
249
t.mkdir('dmode755', mode=0755)
250
self.assertTransportMode(t, 'dmode755', 0755)
251
t.mkdir('dmode555', mode=0555)
252
self.assertTransportMode(t, 'dmode555', 0555)
253
t.mkdir('dmode777', mode=0777)
254
self.assertTransportMode(t, 'dmode777', 0777)
255
t.mkdir('dmode700', mode=0700)
256
self.assertTransportMode(t, 'dmode700', 0700)
257
# TODO: jam 20051215 test mkdir_multi with a mode
258
t.mkdir_multi(['mdmode755'], mode=0755)
259
self.assertTransportMode(t, 'mdmode755', 0755)
261
def test_copy_to(self):
262
# FIXME: test: same server to same server (partly done)
263
# same protocol two servers
264
# and different protocols (done for now except for MemoryTransport.
266
from bzrlib.transport.memory import MemoryTransport
268
def simple_copy_files(transport_from, transport_to):
269
files = ['a', 'b', 'c', 'd']
270
self.build_tree(files, transport=transport_from)
271
transport_from.copy_to(files, transport_to)
273
self.check_transport_contents(transport_to.get(f).read(),
276
t = self.get_transport()
277
temp_transport = MemoryTransport('memory:/')
278
simple_copy_files(t, temp_transport)
279
if not t.is_readonly():
280
t.mkdir('copy_to_simple')
281
t2 = t.clone('copy_to_simple')
282
simple_copy_files(t, t2)
285
# Test that copying into a missing directory raises
288
self.build_tree(['e/', 'e/f'])
291
t.put('e/f', StringIO('contents of e'))
292
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
293
temp_transport.mkdir('e')
294
t.copy_to(['e/f'], temp_transport)
297
temp_transport = MemoryTransport('memory:/')
299
files = ['a', 'b', 'c', 'd']
300
t.copy_to(iter(files), temp_transport)
302
self.check_transport_contents(temp_transport.get(f).read(),
306
for mode in (0666, 0644, 0600, 0400):
307
temp_transport = MemoryTransport("memory:/")
308
t.copy_to(files, temp_transport, mode=mode)
310
self.assertTransportMode(temp_transport, f, mode)
312
def test_append(self):
313
t = self.get_transport()
316
open('a', 'wb').write('diff\ncontents for\na\n')
317
open('b', 'wb').write('contents\nfor b\n')
320
('a', StringIO('diff\ncontents for\na\n')),
321
('b', StringIO('contents\nfor b\n'))
325
self.assertRaises(TransportNotPossible,
326
t.append, 'a', 'add\nsome\nmore\ncontents\n')
327
_append('a', StringIO('add\nsome\nmore\ncontents\n'))
329
t.append('a', StringIO('add\nsome\nmore\ncontents\n'))
331
self.check_transport_contents(
332
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
336
self.assertRaises(TransportNotPossible,
338
[('a', 'and\nthen\nsome\nmore\n'),
339
('b', 'some\nmore\nfor\nb\n')])
340
_append('a', StringIO('and\nthen\nsome\nmore\n'))
341
_append('b', StringIO('some\nmore\nfor\nb\n'))
343
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
344
('b', StringIO('some\nmore\nfor\nb\n'))])
345
self.check_transport_contents(
346
'diff\ncontents for\na\n'
347
'add\nsome\nmore\ncontents\n'
348
'and\nthen\nsome\nmore\n',
350
self.check_transport_contents(
352
'some\nmore\nfor\nb\n',
356
_append('a', StringIO('a little bit more\n'))
357
_append('b', StringIO('from an iterator\n'))
359
t.append_multi(iter([('a', StringIO('a little bit more\n')),
360
('b', StringIO('from an iterator\n'))]))
361
self.check_transport_contents(
362
'diff\ncontents for\na\n'
363
'add\nsome\nmore\ncontents\n'
364
'and\nthen\nsome\nmore\n'
365
'a little bit more\n',
367
self.check_transport_contents(
369
'some\nmore\nfor\nb\n'
370
'from an iterator\n',
374
_append('c', StringIO('some text\nfor a missing file\n'))
375
_append('a', StringIO('some text in a\n'))
376
_append('d', StringIO('missing file r\n'))
378
t.append('c', StringIO('some text\nfor a missing file\n'))
379
t.append_multi([('a', StringIO('some text in a\n')),
380
('d', StringIO('missing file r\n'))])
381
self.check_transport_contents(
382
'diff\ncontents for\na\n'
383
'add\nsome\nmore\ncontents\n'
384
'and\nthen\nsome\nmore\n'
385
'a little bit more\n'
388
self.check_transport_contents('some text\nfor a missing file\n',
390
self.check_transport_contents('missing file r\n', t, 'd')
392
# a file with no parent should fail..
393
if not t.is_readonly():
394
self.assertRaises(NoSuchFile,
395
t.append, 'missing/path',
398
def test_append_file(self):
399
t = self.get_transport()
402
('f1', StringIO('this is a string\nand some more stuff\n')),
403
('f2', StringIO('here is some text\nand a bit more\n')),
404
('f3', StringIO('some text for the\nthird file created\n')),
405
('f4', StringIO('this is a string\nand some more stuff\n')),
406
('f5', StringIO('here is some text\nand a bit more\n')),
407
('f6', StringIO('some text for the\nthird file created\n'))
411
for f, val in contents:
412
open(f, 'wb').write(val.read())
414
t.put_multi(contents)
416
a1 = StringIO('appending to\none\n')
424
self.check_transport_contents(
425
'this is a string\nand some more stuff\n'
426
'appending to\none\n',
429
a2 = StringIO('adding more\ntext to two\n')
430
a3 = StringIO('some garbage\nto put in three\n')
436
t.append_multi([('f2', a2), ('f3', a3)])
440
self.check_transport_contents(
441
'here is some text\nand a bit more\n'
442
'adding more\ntext to two\n',
444
self.check_transport_contents(
445
'some text for the\nthird file created\n'
446
'some garbage\nto put in three\n',
449
# Test that an actual file object can be used with put
458
self.check_transport_contents(
459
'this is a string\nand some more stuff\n'
460
'this is a string\nand some more stuff\n'
461
'appending to\none\n',
470
t.append_multi([('f5', a5), ('f6', a6)])
474
self.check_transport_contents(
475
'here is some text\nand a bit more\n'
476
'here is some text\nand a bit more\n'
477
'adding more\ntext to two\n',
479
self.check_transport_contents(
480
'some text for the\nthird file created\n'
481
'some text for the\nthird file created\n'
482
'some garbage\nto put in three\n',
494
t.append_multi([('a', a6), ('d', a7)])
496
self.check_transport_contents(t.get('f2').read(), t, 'c')
497
self.check_transport_contents(t.get('f3').read(), t, 'd')
499
def test_delete(self):
500
# TODO: Test Transport.delete
501
t = self.get_transport()
503
# Not much to do with a readonly transport
505
self.assertRaises(TransportNotPossible, t.delete, 'missing')
508
t.put('a', StringIO('a little bit of text\n'))
509
self.failUnless(t.has('a'))
511
self.failIf(t.has('a'))
513
self.assertRaises(NoSuchFile, t.delete, 'a')
515
t.put('a', StringIO('a text\n'))
516
t.put('b', StringIO('b text\n'))
517
t.put('c', StringIO('c text\n'))
518
self.assertEqual([True, True, True],
519
list(t.has_multi(['a', 'b', 'c'])))
520
t.delete_multi(['a', 'c'])
521
self.assertEqual([False, True, False],
522
list(t.has_multi(['a', 'b', 'c'])))
523
self.failIf(t.has('a'))
524
self.failUnless(t.has('b'))
525
self.failIf(t.has('c'))
527
self.assertRaises(NoSuchFile,
528
t.delete_multi, ['a', 'b', 'c'])
530
self.assertRaises(NoSuchFile,
531
t.delete_multi, iter(['a', 'b', 'c']))
533
t.put('a', StringIO('another a text\n'))
534
t.put('c', StringIO('another c text\n'))
535
t.delete_multi(iter(['a', 'b', 'c']))
537
# We should have deleted everything
538
# SftpServer creates control files in the
539
# working directory, so we can just do a
541
# self.assertEqual([], os.listdir('.'))
543
def test_rmdir(self):
544
t = self.get_transport()
545
# Not much to do with a readonly transport
547
self.assertRaises(TransportNotPossible, t.rmdir, 'missing')
552
self.assertRaises(NoSuchFile, t.stat, 'adir/bdir')
554
self.assertRaises(NoSuchFile, t.stat, 'adir')
556
def test_rmdir_not_empty(self):
557
"""Deleting a non-empty directory raises an exception
559
sftp (and possibly others) don't give us a specific "directory not
560
empty" exception -- we can just see that the operation failed.
562
t = self.get_transport()
567
self.assertRaises(PathError, t.rmdir, 'adir')
569
def test_rename_dir_succeeds(self):
570
t = self.get_transport()
572
raise TestSkipped("transport is readonly")
574
t.mkdir('adir/asubdir')
575
t.rename('adir', 'bdir')
576
self.assertTrue(t.has('bdir/asubdir'))
577
self.assertFalse(t.has('adir'))
579
def test_rename_dir_nonempty(self):
580
"""Attempting to replace a nonemtpy directory should fail"""
581
t = self.get_transport()
583
raise TestSkipped("transport is readonly")
585
t.mkdir('adir/asubdir')
587
t.mkdir('bdir/bsubdir')
588
self.assertRaises(PathError, t.rename, 'bdir', 'adir')
589
# nothing was changed so it should still be as before
590
self.assertTrue(t.has('bdir/bsubdir'))
591
self.assertFalse(t.has('adir/bdir'))
592
self.assertFalse(t.has('adir/bsubdir'))
594
def test_delete_tree(self):
595
t = self.get_transport()
597
# Not much to do with a readonly transport
599
self.assertRaises(TransportNotPossible, t.delete_tree, 'missing')
602
# and does it like listing ?
605
t.delete_tree('adir')
606
except TransportNotPossible:
607
# ok, this transport does not support delete_tree
610
# did it delete that trivial case?
611
self.assertRaises(NoSuchFile, t.stat, 'adir')
613
self.build_tree(['adir/',
621
t.delete_tree('adir')
622
# adir should be gone now.
623
self.assertRaises(NoSuchFile, t.stat, 'adir')
626
t = self.get_transport()
631
# TODO: I would like to use os.listdir() to
632
# make sure there are no extra files, but SftpServer
633
# creates control files in the working directory
634
# perhaps all of this could be done in a subdirectory
636
t.put('a', StringIO('a first file\n'))
637
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
640
self.failUnless(t.has('b'))
641
self.failIf(t.has('a'))
643
self.check_transport_contents('a first file\n', t, 'b')
644
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
647
t.put('c', StringIO('c this file\n'))
649
self.failIf(t.has('c'))
650
self.check_transport_contents('c this file\n', t, 'b')
652
# TODO: Try to write a test for atomicity
653
# TODO: Test moving into a non-existant subdirectory
654
# TODO: Test Transport.move_multi
657
t = self.get_transport()
662
t.put('a', StringIO('a file\n'))
664
self.check_transport_contents('a file\n', t, 'b')
666
self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
668
# What should the assert be if you try to copy a
669
# file over a directory?
670
#self.assertRaises(Something, t.copy, 'a', 'c')
671
t.put('d', StringIO('text in d\n'))
673
self.check_transport_contents('text in d\n', t, 'b')
675
# TODO: test copy_multi
677
def test_connection_error(self):
678
"""ConnectionError is raised when connection is impossible"""
680
url = self._server.get_bogus_url()
681
except NotImplementedError:
682
raise TestSkipped("Transport %s has no bogus URL support." %
683
self._server.__class__)
684
t = bzrlib.transport.get_transport(url)
687
except (ConnectionError, NoSuchFile), e:
689
except (Exception), e:
690
self.failIf(True, 'Wrong exception thrown: %s' % e)
692
self.failIf(True, 'Did not get the expected exception.')
695
# TODO: Test stat, just try once, and if it throws, stop testing
696
from stat import S_ISDIR, S_ISREG
698
t = self.get_transport()
702
except TransportNotPossible, e:
703
# This transport cannot stat
706
paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
707
sizes = [14, 0, 16, 0, 18]
708
self.build_tree(paths, transport=t)
710
for path, size in zip(paths, sizes):
712
if path.endswith('/'):
713
self.failUnless(S_ISDIR(st.st_mode))
714
# directory sizes are meaningless
716
self.failUnless(S_ISREG(st.st_mode))
717
self.assertEqual(size, st.st_size)
719
remote_stats = list(t.stat_multi(paths))
720
remote_iter_stats = list(t.stat_multi(iter(paths)))
722
self.assertRaises(NoSuchFile, t.stat, 'q')
723
self.assertRaises(NoSuchFile, t.stat, 'b/a')
725
self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
726
self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
727
self.build_tree(['subdir/', 'subdir/file'], transport=t)
728
subdir = t.clone('subdir')
729
subdir.stat('./file')
732
def test_list_dir(self):
733
# TODO: Test list_dir, just try once, and if it throws, stop testing
734
t = self.get_transport()
737
self.assertRaises(TransportNotPossible, t.list_dir, '.')
741
l = list(t.list_dir(d))
745
# SftpServer creates control files in the working directory
746
# so lets move down a directory to avoid those.
747
if not t.is_readonly():
753
self.assertEqual([], sorted_list(u'.'))
754
# c2 is precisely one letter longer than c here to test that
755
# suffixing is not confused.
756
if not t.is_readonly():
757
self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e', 'c2/'], transport=t)
759
self.build_tree(['wd/a', 'wd/b', 'wd/c/', 'wd/c/d', 'wd/c/e', 'wd/c2/'])
761
self.assertEqual([u'a', u'b', u'c', u'c2'], sorted_list(u'.'))
762
self.assertEqual([u'd', u'e'], sorted_list(u'c'))
764
if not t.is_readonly():
771
self.assertEqual([u'a', u'c', u'c2'], sorted_list('.'))
772
self.assertEqual([u'e'], sorted_list(u'c'))
774
self.assertListRaises(NoSuchFile, t.list_dir, 'q')
775
self.assertListRaises(NoSuchFile, t.list_dir, 'c/f')
776
self.assertListRaises(NoSuchFile, t.list_dir, 'a')
778
def test_clone(self):
779
# TODO: Test that clone moves up and down the filesystem
780
t1 = self.get_transport()
782
self.build_tree(['a', 'b/', 'b/c'], transport=t1)
784
self.failUnless(t1.has('a'))
785
self.failUnless(t1.has('b/c'))
786
self.failIf(t1.has('c'))
789
self.assertEqual(t1.base + 'b/', t2.base)
791
self.failUnless(t2.has('c'))
792
self.failIf(t2.has('a'))
795
self.failUnless(t3.has('a'))
796
self.failIf(t3.has('c'))
798
self.failIf(t1.has('b/d'))
799
self.failIf(t2.has('d'))
800
self.failIf(t3.has('b/d'))
803
open('b/d', 'wb').write('newfile\n')
805
t2.put('d', StringIO('newfile\n'))
807
self.failUnless(t1.has('b/d'))
808
self.failUnless(t2.has('d'))
809
self.failUnless(t3.has('b/d'))
811
def test_relpath(self):
812
t = self.get_transport()
813
self.assertEqual('', t.relpath(t.base))
815
self.assertEqual('', t.relpath(t.base[:-1]))
816
# subdirs which dont exist should still give relpaths.
817
self.assertEqual('foo', t.relpath(t.base + 'foo'))
818
# trailing slash should be the same.
819
self.assertEqual('foo', t.relpath(t.base + 'foo/'))
821
def test_abspath(self):
822
# smoke test for abspath. Corner cases for backends like unix fs's
823
# that have aliasing problems like symlinks should go in backend
824
# specific test cases.
825
transport = self.get_transport()
826
self.assertEqual(transport.base + 'relpath',
827
transport.abspath('relpath'))
829
def test_iter_files_recursive(self):
830
transport = self.get_transport()
831
if not transport.listable():
832
self.assertRaises(TransportNotPossible,
833
transport.iter_files_recursive)
835
self.build_tree(['isolated/',
841
paths = set(transport.iter_files_recursive())
842
# nb the directories are not converted
843
self.assertEqual(paths,
844
set(['isolated/dir/foo',
847
sub_transport = transport.clone('isolated')
848
paths = set(sub_transport.iter_files_recursive())
849
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
851
def test_connect_twice_is_same_content(self):
852
# check that our server (whatever it is) is accessable reliably
853
# via get_transport and multiple connections share content.
854
transport = self.get_transport()
855
if transport.is_readonly():
857
transport.put('foo', StringIO('bar'))
858
transport2 = self.get_transport()
859
self.check_transport_contents('bar', transport2, 'foo')
860
# its base should be usable.
861
transport2 = bzrlib.transport.get_transport(transport.base)
862
self.check_transport_contents('bar', transport2, 'foo')
864
# now opening at a relative url should give use a sane result:
865
transport.mkdir('newdir')
866
transport2 = bzrlib.transport.get_transport(transport.base + "newdir")
867
transport2 = transport2.clone('..')
868
self.check_transport_contents('bar', transport2, 'foo')
870
def test_lock_write(self):
871
transport = self.get_transport()
872
if transport.is_readonly():
873
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
875
transport.put('lock', StringIO())
876
lock = transport.lock_write('lock')
877
# TODO make this consistent on all platforms:
878
# self.assertRaises(LockError, transport.lock_write, 'lock')
881
def test_lock_read(self):
882
transport = self.get_transport()
883
if transport.is_readonly():
884
file('lock', 'w').close()
886
transport.put('lock', StringIO())
887
lock = transport.lock_read('lock')
888
# TODO make this consistent on all platforms:
889
# self.assertRaises(LockError, transport.lock_read, 'lock')
added
removed
bzrlib/tests/test_transport_implementations.py
