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