« back to all changes in this revision
Viewing changes to bzrlib/tests/test_transport_implementations.py
- browse files at revision 1540.3.6
- compare with another revision
- download diff
- download tarball
- view history from revision 1540.3.6
- Committer: Martin Pool
- Date: 2006-02-10 06:07:22 UTC
- mfrom: (1553.5.4 bzr.dev (Main development branch))
- mto: This revision was merged to the branch mainline in revision 1611.
- Revision ID: mbp@sourcefrog.net-20060210060722-89e4007ffbceba04
[merge] update from bzr.dev
- files added:
- bzrlib/decorators.py
- bzrlib/doc
- bzrlib/doc/api
- bzrlib/tests/branch_implementations
- tools/doc_generate
- files removed:
- bzrlib/clone.py
- files renamed:
- bzrlib/changeset.py => bzrlib/_changeset.py
- bzrlib/merge_core.py => bzrlib/_merge_core.py
- bzrlib/tests/test_status.py => bzrlib/tests/blackbox/test_status.py
- bzrlib/tests/test_branch.py => bzrlib/tests/branch_implementations/test_branch.py
- bzrlib/xml.py => bzrlib/xml_serializer.py
- bzr_man.py => tools/doc_generate/autodoc_man.py
- files modified:
- .bzrignore
added
removed
bzrlib/tests/test_transport_implementations.py
1
# Copyright (C) 2004, 2005 by Canonical Ltd
2
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.
7
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.
12
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
16
17
"""Tests for Transport implementations.
18
19
Transport implementations tested here are supplied by
20
TransportTestProviderAdapter.
21
"""
22
23
import os
24
from cStringIO import StringIO
25
import stat
26
import sys
27
28
from bzrlib.errors import (NoSuchFile, FileExists,
29
LockError,
30
TransportNotPossible, ConnectionError)
31
from bzrlib.tests import TestCaseInTempDir, TestSkipped
32
from bzrlib.transport import memory, urlescape
33
import bzrlib.transport
34
35
36
def _append(fn, txt):
37
"""Append the given text (file-like object) to the supplied filename."""
38
f = open(fn, 'ab')
39
try:
40
f.write(txt.read())
41
finally:
42
f.close()
43
44
45
class TestTransportImplementation(TestCaseInTempDir):
46
"""Implementation verification for transports.
47
48
To verify a transport we need a server factory, which is a callable
49
that accepts no parameters and returns an implementation of
50
bzrlib.transport.Server.
51
52
That Server is then used to construct transport instances and test
53
the transport via loopback activity.
54
55
Currently this assumes that the Transport object is connected to the
56
current working directory. So that whatever is done
57
through the transport, should show up in the working
58
directory, and vice-versa. This is a bug, because its possible to have
59
URL schemes which provide access to something that may not be
60
result in storage on the local disk, i.e. due to file system limits, or
61
due to it being a database or some other non-filesystem tool.
62
63
This also tests to make sure that the functions work with both
64
generators and lists (assuming iter(list) is effectively a generator)
65
"""
66
67
def setUp(self):
68
super(TestTransportImplementation, self).setUp()
69
self._server = self.transport_server()
70
self._server.setUp()
71
72
def tearDown(self):
73
super(TestTransportImplementation, self).tearDown()
74
self._server.tearDown()
75
76
def check_transport_contents(self, content, transport, relpath):
77
"""Check that transport.get(relpath).read() == content."""
78
self.assertEqualDiff(content, transport.get(relpath).read())
79
80
def get_transport(self):
81
"""Return a connected transport to the local directory."""
82
t = bzrlib.transport.get_transport(self._server.get_url())
83
self.failUnless(isinstance(t, self.transport_class),
84
"Got the wrong class from get_transport"
85
"(%r, expected %r)" % (t.__class__,
86
self.transport_class))
87
return t
88
89
def assertListRaises(self, excClass, func, *args, **kwargs):
90
"""Fail unless excClass is raised when the iterator from func is used.
91
92
Many transport functions can return generators this makes sure
93
to wrap them in a list() call to make sure the whole generator
94
is run, and that the proper exception is raised.
95
"""
96
try:
97
list(func(*args, **kwargs))
98
except excClass:
99
return
100
else:
101
if hasattr(excClass,'__name__'): excName = excClass.__name__
102
else: excName = str(excClass)
103
raise self.failureException, "%s not raised" % excName
104
105
def test_has(self):
106
t = self.get_transport()
107
108
files = ['a', 'b', 'e', 'g', '%']
109
self.build_tree(files, transport=t)
110
self.assertEqual(True, t.has('a'))
111
self.assertEqual(False, t.has('c'))
112
self.assertEqual(True, t.has(urlescape('%')))
113
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
114
[True, True, False, False, True, False, True, False])
115
self.assertEqual(True, t.has_any(['a', 'b', 'c']))
116
self.assertEqual(False, t.has_any(['c', 'd', 'f', urlescape('%%')]))
117
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
118
[True, True, False, False, True, False, True, False])
119
self.assertEqual(False, t.has_any(['c', 'c', 'c']))
120
self.assertEqual(True, t.has_any(['b', 'b', 'b']))
121
122
def test_get(self):
123
t = self.get_transport()
124
125
files = ['a', 'b', 'e', 'g']
126
contents = ['contents of a\n',
127
'contents of b\n',
128
'contents of e\n',
129
'contents of g\n',
130
]
131
self.build_tree(files, transport=t)
132
self.check_transport_contents('contents of a\n', t, 'a')
133
content_f = t.get_multi(files)
134
for content, f in zip(contents, content_f):
135
self.assertEqual(content, f.read())
136
137
content_f = t.get_multi(iter(files))
138
for content, f in zip(contents, content_f):
139
self.assertEqual(content, f.read())
140
141
self.assertRaises(NoSuchFile, t.get, 'c')
142
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
143
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
144
145
def test_put(self):
146
t = self.get_transport()
147
148
if t.is_readonly():
149
self.assertRaises(TransportNotPossible,
150
t.put, 'a', 'some text for a\n')
151
return
152
153
t.put('a', StringIO('some text for a\n'))
154
self.failUnless(t.has('a'))
155
self.check_transport_contents('some text for a\n', t, 'a')
156
# Make sure 'has' is updated
157
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
158
[True, False, False, False, False])
159
# Put also replaces contents
160
self.assertEqual(t.put_multi([('a', StringIO('new\ncontents for\na\n')),
161
('d', StringIO('contents\nfor d\n'))]),
162
2)
163
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
164
[True, False, False, True, False])
165
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
166
self.check_transport_contents('contents\nfor d\n', t, 'd')
167
168
self.assertEqual(
169
t.put_multi(iter([('a', StringIO('diff\ncontents for\na\n')),
170
('d', StringIO('another contents\nfor d\n'))])),
171
2)
172
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
173
self.check_transport_contents('another contents\nfor d\n', t, 'd')
174
175
self.assertRaises(NoSuchFile,
176
t.put, 'path/doesnt/exist/c', 'contents')
177
178
def test_put_permissions(self):
179
t = self.get_transport()
180
181
if t.is_readonly():
182
return
183
t.put('mode644', StringIO('test text\n'), mode=0644)
184
self.assertTransportMode(t, 'mode644', 0644)
185
t.put('mode666', StringIO('test text\n'), mode=0666)
186
self.assertTransportMode(t, 'mode666', 0666)
187
t.put('mode600', StringIO('test text\n'), mode=0600)
188
self.assertTransportMode(t, 'mode600', 0600)
189
# Yes, you can put a file such that it becomes readonly
190
t.put('mode400', StringIO('test text\n'), mode=0400)
191
self.assertTransportMode(t, 'mode400', 0400)
192
t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
193
self.assertTransportMode(t, 'mmode644', 0644)
194
195
def test_mkdir(self):
196
t = self.get_transport()
197
198
if t.is_readonly():
199
# cannot mkdir on readonly transports. We're not testing for
200
# cache coherency because cache behaviour is not currently
201
# defined for the transport interface.
202
self.assertRaises(TransportNotPossible, t.mkdir, '.')
203
self.assertRaises(TransportNotPossible, t.mkdir, 'new_dir')
204
self.assertRaises(TransportNotPossible, t.mkdir_multi, ['new_dir'])
205
self.assertRaises(TransportNotPossible, t.mkdir, 'path/doesnt/exist')
206
return
207
# Test mkdir
208
t.mkdir('dir_a')
209
self.assertEqual(t.has('dir_a'), True)
210
self.assertEqual(t.has('dir_b'), False)
211
212
t.mkdir('dir_b')
213
self.assertEqual(t.has('dir_b'), True)
214
215
t.mkdir_multi(['dir_c', 'dir_d'])
216
217
t.mkdir_multi(iter(['dir_e', 'dir_f']))
218
self.assertEqual(list(t.has_multi(
219
['dir_a', 'dir_b', 'dir_c', 'dir_q',
220
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
221
[True, True, True, False,
222
True, True, True, True])
223
224
# we were testing that a local mkdir followed by a transport
225
# mkdir failed thusly, but given that we * in one process * do not
226
# concurrently fiddle with disk dirs and then use transport to do
227
# things, the win here seems marginal compared to the constraint on
228
# the interface. RBC 20051227
229
t.mkdir('dir_g')
230
self.assertRaises(FileExists, t.mkdir, 'dir_g')
231
232
# Test get/put in sub-directories
233
self.assertEqual(
234
t.put_multi([('dir_a/a', StringIO('contents of dir_a/a')),
235
('dir_b/b', StringIO('contents of dir_b/b'))])
236
, 2)
237
self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
238
self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
239
240
# mkdir of a dir with an absent parent
241
self.assertRaises(NoSuchFile, t.mkdir, 'missing/dir')
242
243
def test_mkdir_permissions(self):
244
t = self.get_transport()
245
if t.is_readonly():
246
return
247
# Test mkdir with a mode
248
t.mkdir('dmode755', mode=0755)
249
self.assertTransportMode(t, 'dmode755', 0755)
250
t.mkdir('dmode555', mode=0555)
251
self.assertTransportMode(t, 'dmode555', 0555)
252
t.mkdir('dmode777', mode=0777)
253
self.assertTransportMode(t, 'dmode777', 0777)
254
t.mkdir('dmode700', mode=0700)
255
self.assertTransportMode(t, 'dmode700', 0700)
256
# TODO: jam 20051215 test mkdir_multi with a mode
257
t.mkdir_multi(['mdmode755'], mode=0755)
258
self.assertTransportMode(t, 'mdmode755', 0755)
259
260
def test_copy_to(self):
261
# FIXME: test: same server to same server (partly done)
262
# same protocol two servers
263
# and different protocols (done for now except for MemoryTransport.
264
# - RBC 20060122
265
from bzrlib.transport.memory import MemoryTransport
266
267
def simple_copy_files(transport_from, transport_to):
268
files = ['a', 'b', 'c', 'd']
269
self.build_tree(files, transport=transport_from)
270
transport_from.copy_to(files, transport_to)
271
for f in files:
272
self.check_transport_contents(transport_to.get(f).read(),
273
transport_from, f)
274
275
t = self.get_transport()
276
temp_transport = MemoryTransport('memory:/')
277
simple_copy_files(t, temp_transport)
278
if not t.is_readonly():
279
t.mkdir('copy_to_simple')
280
t2 = t.clone('copy_to_simple')
281
simple_copy_files(t, t2)
282
283
284
# Test that copying into a missing directory raises
285
# NoSuchFile
286
if t.is_readonly():
287
self.build_tree(['e/', 'e/f'])
288
else:
289
t.mkdir('e')
290
t.put('e/f', StringIO('contents of e'))
291
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
292
temp_transport.mkdir('e')
293
t.copy_to(['e/f'], temp_transport)
294
295
del temp_transport
296
temp_transport = MemoryTransport('memory:/')
297
298
files = ['a', 'b', 'c', 'd']
299
t.copy_to(iter(files), temp_transport)
300
for f in files:
301
self.check_transport_contents(temp_transport.get(f).read(),
302
t, f)
303
del temp_transport
304
305
for mode in (0666, 0644, 0600, 0400):
306
temp_transport = MemoryTransport("memory:/")
307
t.copy_to(files, temp_transport, mode=mode)
308
for f in files:
309
self.assertTransportMode(temp_transport, f, mode)
310
311
def test_append(self):
312
t = self.get_transport()
313
314
if t.is_readonly():
315
open('a', 'wb').write('diff\ncontents for\na\n')
316
open('b', 'wb').write('contents\nfor b\n')
317
else:
318
t.put_multi([
319
('a', StringIO('diff\ncontents for\na\n')),
320
('b', StringIO('contents\nfor b\n'))
321
])
322
323
if t.is_readonly():
324
self.assertRaises(TransportNotPossible,
325
t.append, 'a', 'add\nsome\nmore\ncontents\n')
326
_append('a', StringIO('add\nsome\nmore\ncontents\n'))
327
else:
328
t.append('a', StringIO('add\nsome\nmore\ncontents\n'))
329
330
self.check_transport_contents(
331
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
332
t, 'a')
333
334
if t.is_readonly():
335
self.assertRaises(TransportNotPossible,
336
t.append_multi,
337
[('a', 'and\nthen\nsome\nmore\n'),
338
('b', 'some\nmore\nfor\nb\n')])
339
_append('a', StringIO('and\nthen\nsome\nmore\n'))
340
_append('b', StringIO('some\nmore\nfor\nb\n'))
341
else:
342
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
343
('b', StringIO('some\nmore\nfor\nb\n'))])
344
self.check_transport_contents(
345
'diff\ncontents for\na\n'
346
'add\nsome\nmore\ncontents\n'
347
'and\nthen\nsome\nmore\n',
348
t, 'a')
349
self.check_transport_contents(
350
'contents\nfor b\n'
351
'some\nmore\nfor\nb\n',
352
t, 'b')
353
354
if t.is_readonly():
355
_append('a', StringIO('a little bit more\n'))
356
_append('b', StringIO('from an iterator\n'))
357
else:
358
t.append_multi(iter([('a', StringIO('a little bit more\n')),
359
('b', StringIO('from an iterator\n'))]))
360
self.check_transport_contents(
361
'diff\ncontents for\na\n'
362
'add\nsome\nmore\ncontents\n'
363
'and\nthen\nsome\nmore\n'
364
'a little bit more\n',
365
t, 'a')
366
self.check_transport_contents(
367
'contents\nfor b\n'
368
'some\nmore\nfor\nb\n'
369
'from an iterator\n',
370
t, 'b')
371
372
if t.is_readonly():
373
_append('c', StringIO('some text\nfor a missing file\n'))
374
_append('a', StringIO('some text in a\n'))
375
_append('d', StringIO('missing file r\n'))
376
else:
377
t.append('c', StringIO('some text\nfor a missing file\n'))
378
t.append_multi([('a', StringIO('some text in a\n')),
379
('d', StringIO('missing file r\n'))])
380
self.check_transport_contents(
381
'diff\ncontents for\na\n'
382
'add\nsome\nmore\ncontents\n'
383
'and\nthen\nsome\nmore\n'
384
'a little bit more\n'
385
'some text in a\n',
386
t, 'a')
387
self.check_transport_contents('some text\nfor a missing file\n',
388
t, 'c')
389
self.check_transport_contents('missing file r\n', t, 'd')
390
391
# a file with no parent should fail..
392
if not t.is_readonly():
393
self.assertRaises(NoSuchFile,
394
t.append, 'missing/path',
395
StringIO('content'))
396
397
def test_append_file(self):
398
t = self.get_transport()
399
400
contents = [
401
('f1', StringIO('this is a string\nand some more stuff\n')),
402
('f2', StringIO('here is some text\nand a bit more\n')),
403
('f3', StringIO('some text for the\nthird file created\n')),
404
('f4', StringIO('this is a string\nand some more stuff\n')),
405
('f5', StringIO('here is some text\nand a bit more\n')),
406
('f6', StringIO('some text for the\nthird file created\n'))
407
]
408
409
if t.is_readonly():
410
for f, val in contents:
411
open(f, 'wb').write(val.read())
412
else:
413
t.put_multi(contents)
414
415
a1 = StringIO('appending to\none\n')
416
if t.is_readonly():
417
_append('f1', a1)
418
else:
419
t.append('f1', a1)
420
421
del a1
422
423
self.check_transport_contents(
424
'this is a string\nand some more stuff\n'
425
'appending to\none\n',
426
t, 'f1')
427
428
a2 = StringIO('adding more\ntext to two\n')
429
a3 = StringIO('some garbage\nto put in three\n')
430
431
if t.is_readonly():
432
_append('f2', a2)
433
_append('f3', a3)
434
else:
435
t.append_multi([('f2', a2), ('f3', a3)])
436
437
del a2, a3
438
439
self.check_transport_contents(
440
'here is some text\nand a bit more\n'
441
'adding more\ntext to two\n',
442
t, 'f2')
443
self.check_transport_contents(
444
'some text for the\nthird file created\n'
445
'some garbage\nto put in three\n',
446
t, 'f3')
447
448
# Test that an actual file object can be used with put
449
a4 = t.get('f1')
450
if t.is_readonly():
451
_append('f4', a4)
452
else:
453
t.append('f4', a4)
454
455
del a4
456
457
self.check_transport_contents(
458
'this is a string\nand some more stuff\n'
459
'this is a string\nand some more stuff\n'
460
'appending to\none\n',
461
t, 'f4')
462
463
a5 = t.get('f2')
464
a6 = t.get('f3')
465
if t.is_readonly():
466
_append('f5', a5)
467
_append('f6', a6)
468
else:
469
t.append_multi([('f5', a5), ('f6', a6)])
470
471
del a5, a6
472
473
self.check_transport_contents(
474
'here is some text\nand a bit more\n'
475
'here is some text\nand a bit more\n'
476
'adding more\ntext to two\n',
477
t, 'f5')
478
self.check_transport_contents(
479
'some text for the\nthird file created\n'
480
'some text for the\nthird file created\n'
481
'some garbage\nto put in three\n',
482
t, 'f6')
483
484
a5 = t.get('f2')
485
a6 = t.get('f2')
486
a7 = t.get('f3')
487
if t.is_readonly():
488
_append('c', a5)
489
_append('a', a6)
490
_append('d', a7)
491
else:
492
t.append('c', a5)
493
t.append_multi([('a', a6), ('d', a7)])
494
del a5, a6, a7
495
self.check_transport_contents(t.get('f2').read(), t, 'c')
496
self.check_transport_contents(t.get('f3').read(), t, 'd')
497
498
def test_delete(self):
499
# TODO: Test Transport.delete
500
t = self.get_transport()
501
502
# Not much to do with a readonly transport
503
if t.is_readonly():
504
self.assertRaises(TransportNotPossible, t.delete, 'missing')
505
return
506
507
t.put('a', StringIO('a little bit of text\n'))
508
self.failUnless(t.has('a'))
509
t.delete('a')
510
self.failIf(t.has('a'))
511
512
self.assertRaises(NoSuchFile, t.delete, 'a')
513
514
t.put('a', StringIO('a text\n'))
515
t.put('b', StringIO('b text\n'))
516
t.put('c', StringIO('c text\n'))
517
self.assertEqual([True, True, True],
518
list(t.has_multi(['a', 'b', 'c'])))
519
t.delete_multi(['a', 'c'])
520
self.assertEqual([False, True, False],
521
list(t.has_multi(['a', 'b', 'c'])))
522
self.failIf(t.has('a'))
523
self.failUnless(t.has('b'))
524
self.failIf(t.has('c'))
525
526
self.assertRaises(NoSuchFile,
527
t.delete_multi, ['a', 'b', 'c'])
528
529
self.assertRaises(NoSuchFile,
530
t.delete_multi, iter(['a', 'b', 'c']))
531
532
t.put('a', StringIO('another a text\n'))
533
t.put('c', StringIO('another c text\n'))
534
t.delete_multi(iter(['a', 'b', 'c']))
535
536
# We should have deleted everything
537
# SftpServer creates control files in the
538
# working directory, so we can just do a
539
# plain "listdir".
540
# self.assertEqual([], os.listdir('.'))
541
542
def test_rmdir(self):
543
t = self.get_transport()
544
# Not much to do with a readonly transport
545
if t.is_readonly():
546
self.assertRaises(TransportNotPossible, t.rmdir, 'missing')
547
return
548
t.mkdir('adir')
549
t.mkdir('adir/bdir')
550
t.rmdir('adir/bdir')
551
self.assertRaises(NoSuchFile, t.stat, 'adir/bdir')
552
t.rmdir('adir')
553
self.assertRaises(NoSuchFile, t.stat, 'adir')
554
555
def test_delete_tree(self):
556
t = self.get_transport()
557
558
# Not much to do with a readonly transport
559
if t.is_readonly():
560
self.assertRaises(TransportNotPossible, t.delete_tree, 'missing')
561
return
562
563
# and does it like listing ?
564
t.mkdir('adir')
565
try:
566
t.delete_tree('adir')
567
except TransportNotPossible:
568
# ok, this transport does not support delete_tree
569
return
570
571
# did it delete that trivial case?
572
self.assertRaises(NoSuchFile, t.stat, 'adir')
573
574
self.build_tree(['adir/',
575
'adir/file',
576
'adir/subdir/',
577
'adir/subdir/file',
578
'adir/subdir2/',
579
'adir/subdir2/file',
580
], transport=t)
581
582
t.delete_tree('adir')
583
# adir should be gone now.
584
self.assertRaises(NoSuchFile, t.stat, 'adir')
585
586
def test_move(self):
587
t = self.get_transport()
588
589
if t.is_readonly():
590
return
591
592
# TODO: I would like to use os.listdir() to
593
# make sure there are no extra files, but SftpServer
594
# creates control files in the working directory
595
# perhaps all of this could be done in a subdirectory
596
597
t.put('a', StringIO('a first file\n'))
598
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
599
600
t.move('a', 'b')
601
self.failUnless(t.has('b'))
602
self.failIf(t.has('a'))
603
604
self.check_transport_contents('a first file\n', t, 'b')
605
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
606
607
# Overwrite a file
608
t.put('c', StringIO('c this file\n'))
609
t.move('c', 'b')
610
self.failIf(t.has('c'))
611
self.check_transport_contents('c this file\n', t, 'b')
612
613
# TODO: Try to write a test for atomicity
614
# TODO: Test moving into a non-existant subdirectory
615
# TODO: Test Transport.move_multi
616
617
def test_copy(self):
618
t = self.get_transport()
619
620
if t.is_readonly():
621
return
622
623
t.put('a', StringIO('a file\n'))
624
t.copy('a', 'b')
625
self.check_transport_contents('a file\n', t, 'b')
626
627
self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
628
os.mkdir('c')
629
# What should the assert be if you try to copy a
630
# file over a directory?
631
#self.assertRaises(Something, t.copy, 'a', 'c')
632
t.put('d', StringIO('text in d\n'))
633
t.copy('d', 'b')
634
self.check_transport_contents('text in d\n', t, 'b')
635
636
# TODO: test copy_multi
637
638
def test_connection_error(self):
639
"""ConnectionError is raised when connection is impossible"""
640
try:
641
url = self._server.get_bogus_url()
642
except NotImplementedError:
643
raise TestSkipped("Transport %s has no bogus URL support." %
644
self._server.__class__)
645
t = bzrlib.transport.get_transport(url)
646
try:
647
t.get('.bzr/branch')
648
except (ConnectionError, NoSuchFile), e:
649
pass
650
except (Exception), e:
651
self.failIf(True, 'Wrong exception thrown: %s' % e)
652
else:
653
self.failIf(True, 'Did not get the expected exception.')
654
655
def test_stat(self):
656
# TODO: Test stat, just try once, and if it throws, stop testing
657
from stat import S_ISDIR, S_ISREG
658
659
t = self.get_transport()
660
661
try:
662
st = t.stat('.')
663
except TransportNotPossible, e:
664
# This transport cannot stat
665
return
666
667
paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
668
sizes = [14, 0, 16, 0, 18]
669
self.build_tree(paths, transport=t)
670
671
for path, size in zip(paths, sizes):
672
st = t.stat(path)
673
if path.endswith('/'):
674
self.failUnless(S_ISDIR(st.st_mode))
675
# directory sizes are meaningless
676
else:
677
self.failUnless(S_ISREG(st.st_mode))
678
self.assertEqual(size, st.st_size)
679
680
remote_stats = list(t.stat_multi(paths))
681
remote_iter_stats = list(t.stat_multi(iter(paths)))
682
683
self.assertRaises(NoSuchFile, t.stat, 'q')
684
self.assertRaises(NoSuchFile, t.stat, 'b/a')
685
686
self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
687
self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
688
self.build_tree(['subdir/', 'subdir/file'], transport=t)
689
subdir = t.clone('subdir')
690
subdir.stat('./file')
691
subdir.stat('.')
692
693
def test_list_dir(self):
694
# TODO: Test list_dir, just try once, and if it throws, stop testing
695
t = self.get_transport()
696
697
if not t.listable():
698
self.assertRaises(TransportNotPossible, t.list_dir, '.')
699
return
700
701
def sorted_list(d):
702
l = list(t.list_dir(d))
703
l.sort()
704
return l
705
706
# SftpServer creates control files in the working directory
707
# so lets move down a directory to avoid those.
708
if not t.is_readonly():
709
t.mkdir('wd')
710
else:
711
os.mkdir('wd')
712
t = t.clone('wd')
713
714
self.assertEqual([], sorted_list(u'.'))
715
# c2 is precisely one letter longer than c here to test that
716
# suffixing is not confused.
717
if not t.is_readonly():
718
self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e', 'c2/'], transport=t)
719
else:
720
self.build_tree(['wd/a', 'wd/b', 'wd/c/', 'wd/c/d', 'wd/c/e', 'wd/c2/'])
721
722
self.assertEqual([u'a', u'b', u'c', u'c2'], sorted_list(u'.'))
723
self.assertEqual([u'd', u'e'], sorted_list(u'c'))
724
725
if not t.is_readonly():
726
t.delete('c/d')
727
t.delete('b')
728
else:
729
os.unlink('wd/c/d')
730
os.unlink('wd/b')
731
732
self.assertEqual([u'a', u'c', u'c2'], sorted_list('.'))
733
self.assertEqual([u'e'], sorted_list(u'c'))
734
735
self.assertListRaises(NoSuchFile, t.list_dir, 'q')
736
self.assertListRaises(NoSuchFile, t.list_dir, 'c/f')
737
self.assertListRaises(NoSuchFile, t.list_dir, 'a')
738
739
def test_clone(self):
740
# TODO: Test that clone moves up and down the filesystem
741
t1 = self.get_transport()
742
743
self.build_tree(['a', 'b/', 'b/c'], transport=t1)
744
745
self.failUnless(t1.has('a'))
746
self.failUnless(t1.has('b/c'))
747
self.failIf(t1.has('c'))
748
749
t2 = t1.clone('b')
750
self.assertEqual(t1.base + 'b/', t2.base)
751
752
self.failUnless(t2.has('c'))
753
self.failIf(t2.has('a'))
754
755
t3 = t2.clone('..')
756
self.failUnless(t3.has('a'))
757
self.failIf(t3.has('c'))
758
759
self.failIf(t1.has('b/d'))
760
self.failIf(t2.has('d'))
761
self.failIf(t3.has('b/d'))
762
763
if t1.is_readonly():
764
open('b/d', 'wb').write('newfile\n')
765
else:
766
t2.put('d', StringIO('newfile\n'))
767
768
self.failUnless(t1.has('b/d'))
769
self.failUnless(t2.has('d'))
770
self.failUnless(t3.has('b/d'))
771
772
def test_relpath(self):
773
t = self.get_transport()
774
self.assertEqual('', t.relpath(t.base))
775
# base ends with /
776
self.assertEqual('', t.relpath(t.base[:-1]))
777
# subdirs which dont exist should still give relpaths.
778
self.assertEqual('foo', t.relpath(t.base + 'foo'))
779
# trailing slash should be the same.
780
self.assertEqual('foo', t.relpath(t.base + 'foo/'))
781
782
def test_abspath(self):
783
# smoke test for abspath. Corner cases for backends like unix fs's
784
# that have aliasing problems like symlinks should go in backend
785
# specific test cases.
786
transport = self.get_transport()
787
self.assertEqual(transport.base + 'relpath',
788
transport.abspath('relpath'))
789
790
def test_iter_files_recursive(self):
791
transport = self.get_transport()
792
if not transport.listable():
793
self.assertRaises(TransportNotPossible,
794
transport.iter_files_recursive)
795
return
796
self.build_tree(['isolated/',
797
'isolated/dir/',
798
'isolated/dir/foo',
799
'isolated/dir/bar',
800
'isolated/bar'],
801
transport=transport)
802
transport = transport.clone('isolated')
803
paths = set(transport.iter_files_recursive())
804
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
805
806
def test_connect_twice_is_same_content(self):
807
# check that our server (whatever it is) is accessable reliably
808
# via get_transport and multiple connections share content.
809
transport = self.get_transport()
810
if transport.is_readonly():
811
return
812
transport.put('foo', StringIO('bar'))
813
transport2 = self.get_transport()
814
self.check_transport_contents('bar', transport2, 'foo')
815
# its base should be usable.
816
transport2 = bzrlib.transport.get_transport(transport.base)
817
self.check_transport_contents('bar', transport2, 'foo')
818
819
# now opening at a relative url should give use a sane result:
820
transport.mkdir('newdir')
821
transport2 = bzrlib.transport.get_transport(transport.base + "newdir")
822
transport2 = transport2.clone('..')
823
self.check_transport_contents('bar', transport2, 'foo')
824
825
def test_lock_write(self):
826
transport = self.get_transport()
827
if transport.is_readonly():
828
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
829
return
830
transport.put('lock', StringIO())
831
lock = transport.lock_write('lock')
832
# TODO make this consistent on all platforms:
833
# self.assertRaises(LockError, transport.lock_write, 'lock')
834
lock.unlock()
835
836
def test_lock_read(self):
837
transport = self.get_transport()
838
if transport.is_readonly():
839
file('lock', 'w').close()
840
else:
841
transport.put('lock', StringIO())
842
lock = transport.lock_read('lock')
843
# TODO make this consistent on all platforms:
844
# self.assertRaises(LockError, transport.lock_read, 'lock')
845
lock.unlock()
Loggerhead is a web-based interface for Breezy
Version: 2.0.1