« back to all changes in this revision
Viewing changes to bzrlib/tests/test_transport.py
- browse files at revision 1543
- compare with another revision
- compare with revision 5401
- download diff
- download tarball
- view history from revision 1543
- Committer: Martin Pool
- Date: 2006-01-12 06:37:23 UTC
- mfrom: (1534.1.6 integration)
- Revision ID: mbp@sourcefrog.net-20060112063723-4ec91b5ff30f0830
[merge] robertc-integration
- files added:
- bzrlib/tests/test_transport_implementations.py
- files modified:
- NEWS
added
removed
bzrlib/tests/test_transport.py
22
22
23
23
from bzrlib.errors import (NoSuchFile, FileExists,
24
24
TransportNotPossible, ConnectionError)
25
from bzrlib.tests import TestCase, TestCaseInTempDir
26
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
27
from bzrlib.transport import memory, urlescape
28
from bzrlib.osutils import pathjoin
29
30
31
def _append(fn, txt):
32
"""Append the given text (file-like object) to the supplied filename."""
33
f = open(fn, 'ab')
34
f.write(txt)
35
f.flush()
36
f.close()
37
del f
38
39
40
if sys.platform != 'win32':
41
def check_mode(test, path, mode):
42
"""Check that a particular path has the correct mode."""
43
actual_mode = stat.S_IMODE(os.stat(path).st_mode)
44
test.assertEqual(mode, actual_mode,
45
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
46
else:
47
def check_mode(test, path, mode):
48
"""On win32 chmod doesn't have any effect,
49
so don't actually check anything
50
"""
51
return
25
from bzrlib.tests import TestCase
26
from bzrlib.transport import (_get_protocol_handlers,
27
_get_transport_modules,
28
register_lazy_transport,
29
_set_protocol_handlers,
30
urlescape,
31
)
52
32
53
33
54
34
class TestTransport(TestCase):
57
37
def test_urlescape(self):
58
38
self.assertEqual('%25', urlescape('%'))
59
39
60
61
class TestTransportMixIn(object):
62
"""Subclass this, and it will provide a series of tests for a Transport.
63
It assumes that the Transport object is connected to the
64
current working directory. So that whatever is done
65
through the transport, should show up in the working
66
directory, and vice-versa.
67
68
This also tests to make sure that the functions work with both
69
generators and lists (assuming iter(list) is effectively a generator)
70
"""
71
readonly = False
72
def get_transport(self):
73
"""Children should override this to return the Transport object.
74
"""
75
raise NotImplementedError
76
77
def assertListRaises(self, excClass, func, *args, **kwargs):
78
"""Many transport functions can return generators this makes sure
79
to wrap them in a list() call to make sure the whole generator
80
is run, and that the proper exception is raised.
81
"""
82
try:
83
list(func(*args, **kwargs))
84
except excClass:
85
return
86
else:
87
if hasattr(excClass,'__name__'): excName = excClass.__name__
88
else: excName = str(excClass)
89
raise self.failureException, "%s not raised" % excName
90
91
def test_has(self):
92
t = self.get_transport()
93
94
files = ['a', 'b', 'e', 'g', '%']
95
self.build_tree(files)
96
self.assertEqual(True, t.has('a'))
97
self.assertEqual(False, t.has('c'))
98
self.assertEqual(True, t.has(urlescape('%')))
99
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
100
[True, True, False, False, True, False, True, False])
101
self.assertEqual(True, t.has_any(['a', 'b', 'c']))
102
self.assertEqual(False, t.has_any(['c', 'd', 'f', urlescape('%%')]))
103
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
104
[True, True, False, False, True, False, True, False])
105
self.assertEqual(False, t.has_any(['c', 'c', 'c']))
106
self.assertEqual(True, t.has_any(['b', 'b', 'b']))
107
108
def test_get(self):
109
t = self.get_transport()
110
111
files = ['a', 'b', 'e', 'g']
112
self.build_tree(files)
113
self.assertEqual(open('a', 'rb').read(), t.get('a').read())
114
content_f = t.get_multi(files)
115
for path,f in zip(files, content_f):
116
self.assertEqual(f.read(), open(path, 'rb').read())
117
118
content_f = t.get_multi(iter(files))
119
for path,f in zip(files, content_f):
120
self.assertEqual(f.read(), open(path, 'rb').read())
121
122
self.assertRaises(NoSuchFile, t.get, 'c')
123
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
124
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
125
126
def test_put(self):
127
t = self.get_transport()
128
129
# TODO: jam 20051215 No need to do anything if the test is readonly
130
# origininally it was thought that it would give
131
# more of a workout to readonly tests. By now the
132
# suite is probably thorough enough without testing
133
# readonly protocols in write sections
134
# The only thing that needs to be tested is that the
135
# right error is raised
136
137
if self.readonly:
138
self.assertRaises(TransportNotPossible,
139
t.put, 'a', 'some text for a\n')
140
open('a', 'wb').write('some text for a\n')
141
else:
142
t.put('a', 'some text for a\n')
143
self.assert_(os.path.exists('a'))
144
self.check_file_contents('a', 'some text for a\n')
145
self.assertEqual(t.get('a').read(), 'some text for a\n')
146
# Make sure 'has' is updated
147
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
148
[True, False, False, False, False])
149
if self.readonly:
150
self.assertRaises(TransportNotPossible,
151
t.put_multi,
152
[('a', 'new\ncontents for\na\n'),
153
('d', 'contents\nfor d\n')])
154
open('a', 'wb').write('new\ncontents for\na\n')
155
open('d', 'wb').write('contents\nfor d\n')
156
else:
157
# Put also replaces contents
158
self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
159
('d', 'contents\nfor d\n')]),
160
2)
161
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
162
[True, False, False, True, False])
163
self.check_file_contents('a', 'new\ncontents for\na\n')
164
self.check_file_contents('d', 'contents\nfor d\n')
165
166
if self.readonly:
167
self.assertRaises(TransportNotPossible,
168
t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
169
('d', 'another contents\nfor d\n')]))
170
open('a', 'wb').write('diff\ncontents for\na\n')
171
open('d', 'wb').write('another contents\nfor d\n')
172
else:
173
self.assertEqual(
174
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
175
('d', 'another contents\nfor d\n')]))
176
, 2)
177
self.check_file_contents('a', 'diff\ncontents for\na\n')
178
self.check_file_contents('d', 'another contents\nfor d\n')
179
180
if self.readonly:
181
self.assertRaises(TransportNotPossible,
182
t.put, 'path/doesnt/exist/c', 'contents')
183
else:
184
self.assertRaises(NoSuchFile,
185
t.put, 'path/doesnt/exist/c', 'contents')
186
187
if not self.readonly:
188
t.put('mode644', 'test text\n', mode=0644)
189
check_mode(self, 'mode644', 0644)
190
191
t.put('mode666', 'test text\n', mode=0666)
192
check_mode(self, 'mode666', 0666)
193
194
t.put('mode600', 'test text\n', mode=0600)
195
check_mode(self, 'mode600', 0600)
196
197
# Yes, you can put a file such that it becomes readonly
198
t.put('mode400', 'test text\n', mode=0400)
199
check_mode(self, 'mode400', 0400)
200
201
t.put_multi([('mmode644', 'text\n')], mode=0644)
202
check_mode(self, 'mmode644', 0644)
203
204
# TODO: jam 20051215 test put_multi with a mode. I didn't bother because
205
# it seems most people don't like the _multi functions
206
207
def test_put_file(self):
208
t = self.get_transport()
209
210
# Test that StringIO can be used as a file-like object with put
211
f1 = StringIO('this is a string\nand some more stuff\n')
212
if self.readonly:
213
open('f1', 'wb').write(f1.read())
214
else:
215
t.put('f1', f1)
216
217
del f1
218
219
self.check_file_contents('f1',
220
'this is a string\nand some more stuff\n')
221
222
f2 = StringIO('here is some text\nand a bit more\n')
223
f3 = StringIO('some text for the\nthird file created\n')
224
225
if self.readonly:
226
open('f2', 'wb').write(f2.read())
227
open('f3', 'wb').write(f3.read())
228
else:
229
t.put_multi([('f2', f2), ('f3', f3)])
230
231
del f2, f3
232
233
self.check_file_contents('f2', 'here is some text\nand a bit more\n')
234
self.check_file_contents('f3', 'some text for the\nthird file created\n')
235
236
# Test that an actual file object can be used with put
237
f4 = open('f1', 'rb')
238
if self.readonly:
239
open('f4', 'wb').write(f4.read())
240
else:
241
t.put('f4', f4)
242
243
del f4
244
245
self.check_file_contents('f4',
246
'this is a string\nand some more stuff\n')
247
248
f5 = open('f2', 'rb')
249
f6 = open('f3', 'rb')
250
if self.readonly:
251
open('f5', 'wb').write(f5.read())
252
open('f6', 'wb').write(f6.read())
253
else:
254
t.put_multi([('f5', f5), ('f6', f6)])
255
256
del f5, f6
257
258
self.check_file_contents('f5', 'here is some text\nand a bit more\n')
259
self.check_file_contents('f6', 'some text for the\nthird file created\n')
260
261
if not self.readonly:
262
sio = StringIO('test text\n')
263
t.put('mode644', sio, mode=0644)
264
check_mode(self, 'mode644', 0644)
265
266
a = open('mode644', 'rb')
267
t.put('mode666', a, mode=0666)
268
check_mode(self, 'mode666', 0666)
269
270
a = open('mode644', 'rb')
271
t.put('mode600', a, mode=0600)
272
check_mode(self, 'mode600', 0600)
273
274
# Yes, you can put a file such that it becomes readonly
275
a = open('mode644', 'rb')
276
t.put('mode400', a, mode=0400)
277
check_mode(self, 'mode400', 0400)
278
279
def test_mkdir(self):
280
t = self.get_transport()
281
282
# Test mkdir
283
os.mkdir('dir_a')
284
self.assertEqual(t.has('dir_a'), True)
285
self.assertEqual(t.has('dir_b'), False)
286
287
if self.readonly:
288
self.assertRaises(TransportNotPossible,
289
t.mkdir, 'dir_b')
290
os.mkdir('dir_b')
291
else:
292
t.mkdir('dir_b')
293
self.assertEqual(t.has('dir_b'), True)
294
self.assert_(os.path.isdir('dir_b'))
295
296
if self.readonly:
297
self.assertRaises(TransportNotPossible,
298
t.mkdir_multi, ['dir_c', 'dir_d'])
299
os.mkdir('dir_c')
300
os.mkdir('dir_d')
301
else:
302
t.mkdir_multi(['dir_c', 'dir_d'])
303
304
if self.readonly:
305
self.assertRaises(TransportNotPossible,
306
t.mkdir_multi, iter(['dir_e', 'dir_f']))
307
os.mkdir('dir_e')
308
os.mkdir('dir_f')
309
else:
310
t.mkdir_multi(iter(['dir_e', 'dir_f']))
311
self.assertEqual(list(t.has_multi(
312
['dir_a', 'dir_b', 'dir_c', 'dir_q',
313
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
314
[True, True, True, False,
315
True, True, True, True])
316
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
317
self.assert_(os.path.isdir(d))
318
319
if not self.readonly:
320
self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
321
self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
322
323
# Make sure the transport recognizes when a
324
# directory is created by other means
325
# Caching Transports will fail, because dir_e was already seen not
326
# to exist. So instead, we will search for a new directory
327
#os.mkdir('dir_e')
328
#if not self.readonly:
329
# self.assertRaises(FileExists, t.mkdir, 'dir_e')
330
331
os.mkdir('dir_g')
332
if not self.readonly:
333
self.assertRaises(FileExists, t.mkdir, 'dir_g')
334
335
# Test get/put in sub-directories
336
if self.readonly:
337
open('dir_a/a', 'wb').write('contents of dir_a/a')
338
open('dir_b/b', 'wb').write('contents of dir_b/b')
339
else:
340
self.assertEqual(
341
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
342
('dir_b/b', 'contents of dir_b/b')])
343
, 2)
344
for f in ('dir_a/a', 'dir_b/b'):
345
self.assertEqual(t.get(f).read(), open(f, 'rb').read())
346
347
if not self.readonly:
348
# Test mkdir with a mode
349
t.mkdir('dmode755', mode=0755)
350
check_mode(self, 'dmode755', 0755)
351
352
t.mkdir('dmode555', mode=0555)
353
check_mode(self, 'dmode555', 0555)
354
355
t.mkdir('dmode777', mode=0777)
356
check_mode(self, 'dmode777', 0777)
357
358
t.mkdir('dmode700', mode=0700)
359
check_mode(self, 'dmode700', 0700)
360
361
# TODO: jam 20051215 test mkdir_multi with a mode
362
t.mkdir_multi(['mdmode755'], mode=0755)
363
check_mode(self, 'mdmode755', 0755)
364
365
366
def test_copy_to(self):
367
import tempfile
368
from bzrlib.transport.local import LocalTransport
369
370
t = self.get_transport()
371
372
files = ['a', 'b', 'c', 'd']
373
self.build_tree(files)
374
375
def get_temp_local():
376
dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
377
dtmp_base = os.path.basename(dtmp)
378
return dtmp_base, LocalTransport(dtmp)
379
dtmp_base, local_t = get_temp_local()
380
381
t.copy_to(files, local_t)
382
for f in files:
383
self.assertEquals(open(f, 'rb').read(),
384
open(pathjoin(dtmp_base, f), 'rb').read())
385
386
# Test that copying into a missing directory raises
387
# NoSuchFile
388
os.mkdir('e')
389
open('e/f', 'wb').write('contents of e')
390
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
391
392
os.mkdir(pathjoin(dtmp_base, 'e'))
393
t.copy_to(['e/f'], local_t)
394
395
del dtmp_base, local_t
396
397
dtmp_base, local_t = get_temp_local()
398
399
files = ['a', 'b', 'c', 'd']
400
t.copy_to(iter(files), local_t)
401
for f in files:
402
self.assertEquals(open(f, 'rb').read(),
403
open(pathjoin(dtmp_base, f), 'rb').read())
404
405
del dtmp_base, local_t
406
407
for mode in (0666, 0644, 0600, 0400):
408
dtmp_base, local_t = get_temp_local()
409
t.copy_to(files, local_t, mode=mode)
410
for f in files:
411
check_mode(self, os.path.join(dtmp_base, f), mode)
412
413
def test_append(self):
414
t = self.get_transport()
415
416
if self.readonly:
417
open('a', 'wb').write('diff\ncontents for\na\n')
418
open('b', 'wb').write('contents\nfor b\n')
419
else:
420
t.put_multi([
421
('a', 'diff\ncontents for\na\n'),
422
('b', 'contents\nfor b\n')
423
])
424
425
if self.readonly:
426
self.assertRaises(TransportNotPossible,
427
t.append, 'a', 'add\nsome\nmore\ncontents\n')
428
_append('a', 'add\nsome\nmore\ncontents\n')
429
else:
430
t.append('a', 'add\nsome\nmore\ncontents\n')
431
432
self.check_file_contents('a',
433
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
434
435
if self.readonly:
436
self.assertRaises(TransportNotPossible,
437
t.append_multi,
438
[('a', 'and\nthen\nsome\nmore\n'),
439
('b', 'some\nmore\nfor\nb\n')])
440
_append('a', 'and\nthen\nsome\nmore\n')
441
_append('b', 'some\nmore\nfor\nb\n')
442
else:
443
t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
444
('b', 'some\nmore\nfor\nb\n')])
445
self.check_file_contents('a',
446
'diff\ncontents for\na\n'
447
'add\nsome\nmore\ncontents\n'
448
'and\nthen\nsome\nmore\n')
449
self.check_file_contents('b',
450
'contents\nfor b\n'
451
'some\nmore\nfor\nb\n')
452
453
if self.readonly:
454
_append('a', 'a little bit more\n')
455
_append('b', 'from an iterator\n')
456
else:
457
t.append_multi(iter([('a', 'a little bit more\n'),
458
('b', 'from an iterator\n')]))
459
self.check_file_contents('a',
460
'diff\ncontents for\na\n'
461
'add\nsome\nmore\ncontents\n'
462
'and\nthen\nsome\nmore\n'
463
'a little bit more\n')
464
self.check_file_contents('b',
465
'contents\nfor b\n'
466
'some\nmore\nfor\nb\n'
467
'from an iterator\n')
468
469
if self.readonly:
470
_append('c', 'some text\nfor a missing file\n')
471
_append('a', 'some text in a\n')
472
_append('d', 'missing file r\n')
473
else:
474
t.append('c', 'some text\nfor a missing file\n')
475
t.append_multi([('a', 'some text in a\n'),
476
('d', 'missing file r\n')])
477
self.check_file_contents('a',
478
'diff\ncontents for\na\n'
479
'add\nsome\nmore\ncontents\n'
480
'and\nthen\nsome\nmore\n'
481
'a little bit more\n'
482
'some text in a\n')
483
self.check_file_contents('c', 'some text\nfor a missing file\n')
484
self.check_file_contents('d', 'missing file r\n')
485
486
def test_append_file(self):
487
t = self.get_transport()
488
489
contents = [
490
('f1', 'this is a string\nand some more stuff\n'),
491
('f2', 'here is some text\nand a bit more\n'),
492
('f3', 'some text for the\nthird file created\n'),
493
('f4', 'this is a string\nand some more stuff\n'),
494
('f5', 'here is some text\nand a bit more\n'),
495
('f6', 'some text for the\nthird file created\n')
496
]
497
498
if self.readonly:
499
for f, val in contents:
500
open(f, 'wb').write(val)
501
else:
502
t.put_multi(contents)
503
504
a1 = StringIO('appending to\none\n')
505
if self.readonly:
506
_append('f1', a1.read())
507
else:
508
t.append('f1', a1)
509
510
del a1
511
512
self.check_file_contents('f1',
513
'this is a string\nand some more stuff\n'
514
'appending to\none\n')
515
516
a2 = StringIO('adding more\ntext to two\n')
517
a3 = StringIO('some garbage\nto put in three\n')
518
519
if self.readonly:
520
_append('f2', a2.read())
521
_append('f3', a3.read())
522
else:
523
t.append_multi([('f2', a2), ('f3', a3)])
524
525
del a2, a3
526
527
self.check_file_contents('f2',
528
'here is some text\nand a bit more\n'
529
'adding more\ntext to two\n')
530
self.check_file_contents('f3',
531
'some text for the\nthird file created\n'
532
'some garbage\nto put in three\n')
533
534
# Test that an actual file object can be used with put
535
a4 = open('f1', 'rb')
536
if self.readonly:
537
_append('f4', a4.read())
538
else:
539
t.append('f4', a4)
540
541
del a4
542
543
self.check_file_contents('f4',
544
'this is a string\nand some more stuff\n'
545
'this is a string\nand some more stuff\n'
546
'appending to\none\n')
547
548
a5 = open('f2', 'rb')
549
a6 = open('f3', 'rb')
550
if self.readonly:
551
_append('f5', a5.read())
552
_append('f6', a6.read())
553
else:
554
t.append_multi([('f5', a5), ('f6', a6)])
555
556
del a5, a6
557
558
self.check_file_contents('f5',
559
'here is some text\nand a bit more\n'
560
'here is some text\nand a bit more\n'
561
'adding more\ntext to two\n')
562
self.check_file_contents('f6',
563
'some text for the\nthird file created\n'
564
'some text for the\nthird file created\n'
565
'some garbage\nto put in three\n')
566
567
a5 = open('f2', 'rb')
568
a6 = open('f2', 'rb')
569
a7 = open('f3', 'rb')
570
if self.readonly:
571
_append('c', a5.read())
572
_append('a', a6.read())
573
_append('d', a7.read())
574
else:
575
t.append('c', a5)
576
t.append_multi([('a', a6), ('d', a7)])
577
del a5, a6, a7
578
self.check_file_contents('c', open('f2', 'rb').read())
579
self.check_file_contents('d', open('f3', 'rb').read())
580
581
582
def test_delete(self):
583
# TODO: Test Transport.delete
584
t = self.get_transport()
585
586
# Not much to do with a readonly transport
587
if self.readonly:
588
return
589
590
open('a', 'wb').write('a little bit of text\n')
591
self.failUnless(t.has('a'))
592
self.failUnlessExists('a')
593
t.delete('a')
594
self.failIf(os.path.lexists('a'))
595
596
self.assertRaises(NoSuchFile, t.delete, 'a')
597
598
open('a', 'wb').write('a text\n')
599
open('b', 'wb').write('b text\n')
600
open('c', 'wb').write('c text\n')
601
self.assertEqual([True, True, True],
602
list(t.has_multi(['a', 'b', 'c'])))
603
t.delete_multi(['a', 'c'])
604
self.assertEqual([False, True, False],
605
list(t.has_multi(['a', 'b', 'c'])))
606
self.failIf(os.path.lexists('a'))
607
self.failUnlessExists('b')
608
self.failIf(os.path.lexists('c'))
609
610
self.assertRaises(NoSuchFile,
611
t.delete_multi, ['a', 'b', 'c'])
612
613
self.assertRaises(NoSuchFile,
614
t.delete_multi, iter(['a', 'b', 'c']))
615
616
open('a', 'wb').write('another a text\n')
617
open('c', 'wb').write('another c text\n')
618
t.delete_multi(iter(['a', 'b', 'c']))
619
620
# We should have deleted everything
621
# SftpServer creates control files in the
622
# working directory, so we can just do a
623
# plain "listdir".
624
# self.assertEqual([], os.listdir('.'))
625
626
def test_move(self):
627
t = self.get_transport()
628
629
if self.readonly:
630
return
631
632
# TODO: I would like to use os.listdir() to
633
# make sure there are no extra files, but SftpServer
634
# creates control files in the working directory
635
# perhaps all of this could be done in a subdirectory
636
637
open('a', 'wb').write('a first file\n')
638
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
639
640
t.move('a', 'b')
641
self.failUnlessExists('b')
642
self.failIf(os.path.lexists('a'))
643
644
self.check_file_contents('b', 'a first file\n')
645
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
646
647
# Overwrite a file
648
open('c', 'wb').write('c this file\n')
649
t.move('c', 'b')
650
self.failIf(os.path.lexists('c'))
651
self.check_file_contents('b', 'c this file\n')
652
653
# TODO: Try to write a test for atomicity
654
# TODO: Test moving into a non-existant subdirectory
655
# TODO: Test Transport.move_multi
656
657
def test_copy(self):
658
t = self.get_transport()
659
660
if self.readonly:
661
return
662
663
open('a', 'wb').write('a file\n')
664
t.copy('a', 'b')
665
self.check_file_contents('b', 'a file\n')
666
667
self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
668
os.mkdir('c')
669
# What should the assert be if you try to copy a
670
# file over a directory?
671
#self.assertRaises(Something, t.copy, 'a', 'c')
672
open('d', 'wb').write('text in d\n')
673
t.copy('d', 'b')
674
self.check_file_contents('b', 'text in d\n')
675
676
# TODO: test copy_multi
677
678
def test_connection_error(self):
679
"""ConnectionError is raised when connection is impossible"""
680
if not hasattr(self, "get_bogus_transport"):
681
return
682
t = self.get_bogus_transport()
683
try:
684
t.get('.bzr/branch')
685
except (ConnectionError, NoSuchFile), e:
686
pass
687
except (Exception), e:
688
self.failIf(True, 'Wrong exception thrown: %s' % e)
689
else:
690
self.failIf(True, 'Did not get the expected exception.')
691
692
def test_stat(self):
693
# TODO: Test stat, just try once, and if it throws, stop testing
694
from stat import S_ISDIR, S_ISREG
695
696
t = self.get_transport()
697
698
try:
699
st = t.stat('.')
700
except TransportNotPossible, e:
701
# This transport cannot stat
702
return
703
704
paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
705
self.build_tree(paths)
706
707
local_stats = []
708
709
for p in paths:
710
st = t.stat(p)
711
local_st = os.stat(p)
712
if p.endswith('/'):
713
self.failUnless(S_ISDIR(st.st_mode))
714
else:
715
self.failUnless(S_ISREG(st.st_mode))
716
self.assertEqual(local_st.st_size, st.st_size)
717
self.assertEqual(local_st.st_mode, st.st_mode)
718
local_stats.append(local_st)
719
720
remote_stats = list(t.stat_multi(paths))
721
remote_iter_stats = list(t.stat_multi(iter(paths)))
722
723
for local, remote, remote_iter in \
724
zip(local_stats, remote_stats, remote_iter_stats):
725
self.assertEqual(local.st_mode, remote.st_mode)
726
self.assertEqual(local.st_mode, remote_iter.st_mode)
727
728
self.assertEqual(local.st_size, remote.st_size)
729
self.assertEqual(local.st_size, remote_iter.st_size)
730
# Should we test UID/GID?
731
732
self.assertRaises(NoSuchFile, t.stat, 'q')
733
self.assertRaises(NoSuchFile, t.stat, 'b/a')
734
735
self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
736
self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
737
738
def test_list_dir(self):
739
# TODO: Test list_dir, just try once, and if it throws, stop testing
740
t = self.get_transport()
741
742
if not t.listable():
743
self.assertRaises(TransportNotPossible, t.list_dir, '.')
744
return
745
746
def sorted_list(d):
747
l = list(t.list_dir(d))
748
l.sort()
749
return l
750
751
# SftpServer creates control files in the working directory
752
# so lets move down a directory to be safe
753
os.mkdir('wd')
754
os.chdir('wd')
755
t = t.clone('wd')
756
757
self.assertEqual([], sorted_list(u'.'))
758
self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e'])
759
760
self.assertEqual([u'a', u'b', u'c'], sorted_list(u'.'))
761
self.assertEqual([u'd', u'e'], sorted_list(u'c'))
762
763
os.remove('c/d')
764
os.remove('b')
765
self.assertEqual([u'a', u'c'], sorted_list('.'))
766
self.assertEqual([u'e'], sorted_list(u'c'))
767
768
self.assertListRaises(NoSuchFile, t.list_dir, 'q')
769
self.assertListRaises(NoSuchFile, t.list_dir, 'c/f')
770
self.assertListRaises(NoSuchFile, t.list_dir, 'a')
771
772
def test_clone(self):
773
# TODO: Test that clone moves up and down the filesystem
774
t1 = self.get_transport()
775
776
self.build_tree(['a', 'b/', 'b/c'])
777
778
self.failUnless(t1.has('a'))
779
self.failUnless(t1.has('b/c'))
780
self.failIf(t1.has('c'))
781
782
t2 = t1.clone('b')
783
self.failUnless(t2.has('c'))
784
self.failIf(t2.has('a'))
785
786
t3 = t2.clone('..')
787
self.failUnless(t3.has('a'))
788
self.failIf(t3.has('c'))
789
790
self.failIf(t1.has('b/d'))
791
self.failIf(t2.has('d'))
792
self.failIf(t3.has('b/d'))
793
794
if self.readonly:
795
open('b/d', 'wb').write('newfile\n')
796
else:
797
t2.put('d', 'newfile\n')
798
799
self.failUnless(t1.has('b/d'))
800
self.failUnless(t2.has('d'))
801
self.failUnless(t3.has('b/d'))
802
803
804
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
805
def get_transport(self):
806
from bzrlib.transport.local import LocalTransport
807
return LocalTransport(u'.')
808
809
810
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
811
812
readonly = True
813
814
def get_transport(self):
815
from bzrlib.transport.http import HttpTransport
816
url = self.get_remote_url(u'.')
817
return HttpTransport(url)
818
819
def get_bogus_transport(self):
820
from bzrlib.transport.http import HttpTransport
821
return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
822
823
824
class TestMemoryTransport(TestCase):
825
826
def test_get_transport(self):
827
memory.MemoryTransport()
828
829
def test_clone(self):
830
transport = memory.MemoryTransport()
831
self.failUnless(transport.clone() is transport)
832
833
def test_abspath(self):
834
transport = memory.MemoryTransport()
835
self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
836
837
def test_relpath(self):
838
transport = memory.MemoryTransport()
839
840
def test_append_and_get(self):
841
transport = memory.MemoryTransport()
842
transport.append('path', StringIO('content'))
843
self.assertEqual(transport.get('path').read(), 'content')
844
transport.append('path', StringIO('content'))
845
self.assertEqual(transport.get('path').read(), 'contentcontent')
846
847
def test_put_and_get(self):
848
transport = memory.MemoryTransport()
849
transport.put('path', StringIO('content'))
850
self.assertEqual(transport.get('path').read(), 'content')
851
transport.put('path', StringIO('content'))
852
self.assertEqual(transport.get('path').read(), 'content')
853
854
def test_append_without_dir_fails(self):
855
transport = memory.MemoryTransport()
856
self.assertRaises(NoSuchFile,
857
transport.append, 'dir/path', StringIO('content'))
858
859
def test_put_without_dir_fails(self):
860
transport = memory.MemoryTransport()
861
self.assertRaises(NoSuchFile,
862
transport.put, 'dir/path', StringIO('content'))
863
864
def test_get_missing(self):
865
transport = memory.MemoryTransport()
866
self.assertRaises(NoSuchFile, transport.get, 'foo')
867
868
def test_has_missing(self):
869
transport = memory.MemoryTransport()
870
self.assertEquals(False, transport.has('foo'))
871
872
def test_has_present(self):
873
transport = memory.MemoryTransport()
874
transport.append('foo', StringIO('content'))
875
self.assertEquals(True, transport.has('foo'))
876
877
def test_mkdir(self):
878
transport = memory.MemoryTransport()
879
transport.mkdir('dir')
880
transport.append('dir/path', StringIO('content'))
881
self.assertEqual(transport.get('dir/path').read(), 'content')
882
883
def test_mkdir_missing_parent(self):
884
transport = memory.MemoryTransport()
885
self.assertRaises(NoSuchFile,
886
transport.mkdir, 'dir/dir')
887
888
def test_mkdir_twice(self):
889
transport = memory.MemoryTransport()
890
transport.mkdir('dir')
891
self.assertRaises(FileExists, transport.mkdir, 'dir')
40
def test__get_set_protocol_handlers(self):
41
handlers = _get_protocol_handlers()
42
self.assertNotEqual({}, handlers)
43
try:
44
_set_protocol_handlers({})
45
self.assertEqual({}, _get_protocol_handlers())
46
finally:
47
_set_protocol_handlers(handlers)
48
49
def test_get_transport_modules(self):
50
handlers = _get_protocol_handlers()
51
class SampleHandler(object):
52
"""I exist, isnt that enough?"""
53
try:
54
my_handlers = {}
55
_set_protocol_handlers(my_handlers)
56
register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
57
register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
58
self.assertEqual([SampleHandler.__module__],
59
_get_transport_modules())
60
finally:
61
_set_protocol_handlers(handlers)
62
63
64
class MemoryTransportTest(TestCase):
65
"""Memory transport specific tests."""
892
66
893
67
def test_parameters(self):
68
import bzrlib.transport.memory as memory
894
69
transport = memory.MemoryTransport()
895
70
self.assertEqual(True, transport.listable())
896
71
self.assertEqual(False, transport.should_cache())
897
898
def test_iter_files_recursive(self):
899
transport = memory.MemoryTransport()
900
transport.mkdir('dir')
901
transport.put('dir/foo', StringIO('content'))
902
transport.put('dir/bar', StringIO('content'))
903
transport.put('bar', StringIO('content'))
904
paths = set(transport.iter_files_recursive())
905
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
906
907
def test_stat(self):
908
transport = memory.MemoryTransport()
909
transport.put('foo', StringIO('content'))
910
transport.put('bar', StringIO('phowar'))
911
self.assertEqual(7, transport.stat('foo').st_size)
912
self.assertEqual(6, transport.stat('bar').st_size)
913
72
self.assertEqual(False, transport.is_readonly())
Loggerhead is a web-based interface for Breezy
Version: 2.0.1