1
# Copyright (C) 2004, 2005 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
19
from cStringIO import StringIO
21
from bzrlib.errors import (NoSuchFile, FileExists,
22
TransportNotPossible, ConnectionError)
23
from bzrlib.tests import TestCase, TestCaseInTempDir
24
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
25
from bzrlib.transport import memory, urlescape
29
"""Append the given text (file-like object) to the supplied filename."""
36
class TestTransport(TestCase):
37
"""Test the non transport-concrete class functionality."""
39
def test_urlescape(self):
40
self.assertEqual('%25', urlescape('%'))
43
class TestTransportMixIn(object):
44
"""Subclass this, and it will provide a series of tests for a Transport.
45
It assumes that the Transport object is connected to the
46
current working directory. So that whatever is done
47
through the transport, should show up in the working
48
directory, and vice-versa.
50
This also tests to make sure that the functions work with both
51
generators and lists (assuming iter(list) is effectively a generator)
54
def get_transport(self):
55
"""Children should override this to return the Transport object.
57
raise NotImplementedError
59
def assertListRaises(self, excClass, func, *args, **kwargs):
60
"""Many transport functions can return generators this makes sure
61
to wrap them in a list() call to make sure the whole generator
62
is run, and that the proper exception is raised.
65
list(func(*args, **kwargs))
69
if hasattr(excClass,'__name__'): excName = excClass.__name__
70
else: excName = str(excClass)
71
raise self.failureException, "%s not raised" % excName
74
t = self.get_transport()
76
files = ['a', 'b', 'e', 'g', '%']
77
self.build_tree(files)
78
self.assertEqual(True, t.has('a'))
79
self.assertEqual(False, t.has('c'))
80
self.assertEqual(True, t.has(urlescape('%')))
81
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
82
[True, True, False, False, True, False, True, False])
83
self.assertEqual(True, t.has_any(['a', 'b', 'c']))
84
self.assertEqual(False, t.has_any(['c', 'd', 'f', urlescape('%%')]))
85
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
86
[True, True, False, False, True, False, True, False])
87
self.assertEqual(False, t.has_any(['c', 'c', 'c']))
88
self.assertEqual(True, t.has_any(['b', 'b', 'b']))
91
t = self.get_transport()
93
files = ['a', 'b', 'e', 'g']
94
self.build_tree(files)
95
self.assertEqual(open('a', 'rb').read(), t.get('a').read())
96
content_f = t.get_multi(files)
97
for path,f in zip(files, content_f):
98
self.assertEqual(open(path).read(), f.read())
100
content_f = t.get_multi(iter(files))
101
for path,f in zip(files, content_f):
102
self.assertEqual(f.read(), open(path).read())
104
self.assertRaises(NoSuchFile, t.get, 'c')
105
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
106
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
109
t = self.get_transport()
112
self.assertRaises(TransportNotPossible,
113
t.put, 'a', 'some text for a\n')
114
open('a', 'wb').write('some text for a\n')
116
t.put('a', 'some text for a\n')
117
self.assert_(os.path.exists('a'))
118
self.check_file_contents('a', 'some text for a\n')
119
self.assertEqual(t.get('a').read(), 'some text for a\n')
120
# Make sure 'has' is updated
121
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
122
[True, False, False, False, False])
124
self.assertRaises(TransportNotPossible,
126
[('a', 'new\ncontents for\na\n'),
127
('d', 'contents\nfor d\n')])
128
open('a', 'wb').write('new\ncontents for\na\n')
129
open('d', 'wb').write('contents\nfor d\n')
131
# Put also replaces contents
132
self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
133
('d', 'contents\nfor d\n')]),
135
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
136
[True, False, False, True, False])
137
self.check_file_contents('a', 'new\ncontents for\na\n')
138
self.check_file_contents('d', 'contents\nfor d\n')
141
self.assertRaises(TransportNotPossible,
142
t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
143
('d', 'another contents\nfor d\n')]))
144
open('a', 'wb').write('diff\ncontents for\na\n')
145
open('d', 'wb').write('another contents\nfor d\n')
148
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
149
('d', 'another contents\nfor d\n')]))
151
self.check_file_contents('a', 'diff\ncontents for\na\n')
152
self.check_file_contents('d', 'another contents\nfor d\n')
155
self.assertRaises(TransportNotPossible,
156
t.put, 'path/doesnt/exist/c', 'contents')
158
self.assertRaises(NoSuchFile,
159
t.put, 'path/doesnt/exist/c', 'contents')
161
def test_put_file(self):
162
t = self.get_transport()
164
# Test that StringIO can be used as a file-like object with put
165
f1 = StringIO('this is a string\nand some more stuff\n')
167
open('f1', 'wb').write(f1.read())
173
self.check_file_contents('f1',
174
'this is a string\nand some more stuff\n')
176
f2 = StringIO('here is some text\nand a bit more\n')
177
f3 = StringIO('some text for the\nthird file created\n')
180
open('f2', 'wb').write(f2.read())
181
open('f3', 'wb').write(f3.read())
183
t.put_multi([('f2', f2), ('f3', f3)])
187
self.check_file_contents('f2', 'here is some text\nand a bit more\n')
188
self.check_file_contents('f3', 'some text for the\nthird file created\n')
190
# Test that an actual file object can be used with put
191
f4 = open('f1', 'rb')
193
open('f4', 'wb').write(f4.read())
199
self.check_file_contents('f4',
200
'this is a string\nand some more stuff\n')
202
f5 = open('f2', 'rb')
203
f6 = open('f3', 'rb')
205
open('f5', 'wb').write(f5.read())
206
open('f6', 'wb').write(f6.read())
208
t.put_multi([('f5', f5), ('f6', f6)])
212
self.check_file_contents('f5', 'here is some text\nand a bit more\n')
213
self.check_file_contents('f6', 'some text for the\nthird file created\n')
215
def test_mkdir(self):
216
t = self.get_transport()
220
self.assertEqual(t.has('dir_a'), True)
221
self.assertEqual(t.has('dir_b'), False)
224
self.assertRaises(TransportNotPossible,
229
self.assertEqual(t.has('dir_b'), True)
230
self.assert_(os.path.isdir('dir_b'))
233
self.assertRaises(TransportNotPossible,
234
t.mkdir_multi, ['dir_c', 'dir_d'])
238
t.mkdir_multi(['dir_c', 'dir_d'])
241
self.assertRaises(TransportNotPossible,
242
t.mkdir_multi, iter(['dir_e', 'dir_f']))
246
t.mkdir_multi(iter(['dir_e', 'dir_f']))
247
self.assertEqual(list(t.has_multi(
248
['dir_a', 'dir_b', 'dir_c', 'dir_q',
249
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
250
[True, True, True, False,
251
True, True, True, True])
252
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
253
self.assert_(os.path.isdir(d))
255
if not self.readonly:
256
self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
257
self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
259
# Make sure the transport recognizes when a
260
# directory is created by other means
261
# Caching Transports will fail, because dir_e was already seen not
262
# to exist. So instead, we will search for a new directory
264
#if not self.readonly:
265
# self.assertRaises(FileExists, t.mkdir, 'dir_e')
268
if not self.readonly:
269
self.assertRaises(FileExists, t.mkdir, 'dir_g')
271
# Test get/put in sub-directories
273
open('dir_a/a', 'wb').write('contents of dir_a/a')
274
open('dir_b/b', 'wb').write('contents of dir_b/b')
277
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
278
('dir_b/b', 'contents of dir_b/b')])
280
for f in ('dir_a/a', 'dir_b/b'):
281
self.assertEqual(t.get(f).read(), open(f).read())
283
def test_copy_to(self):
285
from bzrlib.transport.local import LocalTransport
287
t = self.get_transport()
289
files = ['a', 'b', 'c', 'd']
290
self.build_tree(files)
292
dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
293
dtmp_base = os.path.basename(dtmp)
294
local_t = LocalTransport(dtmp)
296
t.copy_to(files, local_t)
298
self.assertEquals(open(f).read(),
299
open(os.path.join(dtmp_base, f)).read())
301
# Test that copying into a missing directory raises
304
open('e/f', 'wb').write('contents of e')
305
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
307
os.mkdir(os.path.join(dtmp_base, 'e'))
308
t.copy_to(['e/f'], local_t)
310
del dtmp, dtmp_base, local_t
312
dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
313
dtmp_base = os.path.basename(dtmp)
314
local_t = LocalTransport(dtmp)
316
files = ['a', 'b', 'c', 'd']
317
t.copy_to(iter(files), local_t)
319
self.assertEquals(open(f).read(),
320
open(os.path.join(dtmp_base, f)).read())
322
del dtmp, dtmp_base, local_t
324
def test_append(self):
325
t = self.get_transport()
328
open('a', 'wb').write('diff\ncontents for\na\n')
329
open('b', 'wb').write('contents\nfor b\n')
332
('a', 'diff\ncontents for\na\n'),
333
('b', 'contents\nfor b\n')
337
self.assertRaises(TransportNotPossible,
338
t.append, 'a', 'add\nsome\nmore\ncontents\n')
339
_append('a', 'add\nsome\nmore\ncontents\n')
341
t.append('a', 'add\nsome\nmore\ncontents\n')
343
self.check_file_contents('a',
344
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
347
self.assertRaises(TransportNotPossible,
349
[('a', 'and\nthen\nsome\nmore\n'),
350
('b', 'some\nmore\nfor\nb\n')])
351
_append('a', 'and\nthen\nsome\nmore\n')
352
_append('b', 'some\nmore\nfor\nb\n')
354
t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
355
('b', 'some\nmore\nfor\nb\n')])
356
self.check_file_contents('a',
357
'diff\ncontents for\na\n'
358
'add\nsome\nmore\ncontents\n'
359
'and\nthen\nsome\nmore\n')
360
self.check_file_contents('b',
362
'some\nmore\nfor\nb\n')
365
_append('a', 'a little bit more\n')
366
_append('b', 'from an iterator\n')
368
t.append_multi(iter([('a', 'a little bit more\n'),
369
('b', 'from an iterator\n')]))
370
self.check_file_contents('a',
371
'diff\ncontents for\na\n'
372
'add\nsome\nmore\ncontents\n'
373
'and\nthen\nsome\nmore\n'
374
'a little bit more\n')
375
self.check_file_contents('b',
377
'some\nmore\nfor\nb\n'
378
'from an iterator\n')
381
_append('c', 'some text\nfor a missing file\n')
382
_append('a', 'some text in a\n')
383
_append('d', 'missing file r\n')
385
t.append('c', 'some text\nfor a missing file\n')
386
t.append_multi([('a', 'some text in a\n'),
387
('d', 'missing file r\n')])
388
self.check_file_contents('a',
389
'diff\ncontents for\na\n'
390
'add\nsome\nmore\ncontents\n'
391
'and\nthen\nsome\nmore\n'
392
'a little bit more\n'
394
self.check_file_contents('c', 'some text\nfor a missing file\n')
395
self.check_file_contents('d', 'missing file r\n')
397
def test_append_file(self):
398
t = self.get_transport()
401
('f1', 'this is a string\nand some more stuff\n'),
402
('f2', 'here is some text\nand a bit more\n'),
403
('f3', 'some text for the\nthird file created\n'),
404
('f4', 'this is a string\nand some more stuff\n'),
405
('f5', 'here is some text\nand a bit more\n'),
406
('f6', 'some text for the\nthird file created\n')
410
for f, val in contents:
411
open(f, 'wb').write(val)
413
t.put_multi(contents)
415
a1 = StringIO('appending to\none\n')
417
_append('f1', a1.read())
423
self.check_file_contents('f1',
424
'this is a string\nand some more stuff\n'
425
'appending to\none\n')
427
a2 = StringIO('adding more\ntext to two\n')
428
a3 = StringIO('some garbage\nto put in three\n')
431
_append('f2', a2.read())
432
_append('f3', a3.read())
434
t.append_multi([('f2', a2), ('f3', a3)])
438
self.check_file_contents('f2',
439
'here is some text\nand a bit more\n'
440
'adding more\ntext to two\n')
441
self.check_file_contents('f3',
442
'some text for the\nthird file created\n'
443
'some garbage\nto put in three\n')
445
# Test that an actual file object can be used with put
446
a4 = open('f1', 'rb')
448
_append('f4', a4.read())
454
self.check_file_contents('f4',
455
'this is a string\nand some more stuff\n'
456
'this is a string\nand some more stuff\n'
457
'appending to\none\n')
459
a5 = open('f2', 'rb')
460
a6 = open('f3', 'rb')
462
_append('f5', a5.read())
463
_append('f6', a6.read())
465
t.append_multi([('f5', a5), ('f6', a6)])
469
self.check_file_contents('f5',
470
'here is some text\nand a bit more\n'
471
'here is some text\nand a bit more\n'
472
'adding more\ntext to two\n')
473
self.check_file_contents('f6',
474
'some text for the\nthird file created\n'
475
'some text for the\nthird file created\n'
476
'some garbage\nto put in three\n')
478
a5 = open('f2', 'rb')
479
a6 = open('f2', 'rb')
480
a7 = open('f3', 'rb')
482
_append('c', a5.read())
483
_append('a', a6.read())
484
_append('d', a7.read())
487
t.append_multi([('a', a6), ('d', a7)])
489
self.check_file_contents('c', open('f2', 'rb').read())
490
self.check_file_contents('d', open('f3', 'rb').read())
493
def test_delete(self):
494
# TODO: Test Transport.delete
495
t = self.get_transport()
497
# Not much to do with a readonly transport
501
open('a', 'wb').write('a little bit of text\n')
502
self.failUnless(t.has('a'))
503
self.failUnlessExists('a')
505
self.failIf(os.path.lexists('a'))
507
self.assertRaises(NoSuchFile, t.delete, 'a')
509
open('a', 'wb').write('a text\n')
510
open('b', 'wb').write('b text\n')
511
open('c', 'wb').write('c text\n')
512
self.assertEqual([True, True, True],
513
list(t.has_multi(['a', 'b', 'c'])))
514
t.delete_multi(['a', 'c'])
515
self.assertEqual([False, True, False],
516
list(t.has_multi(['a', 'b', 'c'])))
517
self.failIf(os.path.lexists('a'))
518
self.failUnlessExists('b')
519
self.failIf(os.path.lexists('c'))
521
self.assertRaises(NoSuchFile,
522
t.delete_multi, ['a', 'b', 'c'])
524
self.assertRaises(NoSuchFile,
525
t.delete_multi, iter(['a', 'b', 'c']))
527
open('a', 'wb').write('another a text\n')
528
open('c', 'wb').write('another c text\n')
529
t.delete_multi(iter(['a', 'b', 'c']))
531
# We should have deleted everything
532
# SftpServer creates control files in the
533
# working directory, so we can just do a
535
# self.assertEqual([], os.listdir('.'))
538
t = self.get_transport()
543
# TODO: I would like to use os.listdir() to
544
# make sure there are no extra files, but SftpServer
545
# creates control files in the working directory
546
# perhaps all of this could be done in a subdirectory
548
open('a', 'wb').write('a first file\n')
549
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
552
self.failUnlessExists('b')
553
self.failIf(os.path.lexists('a'))
555
self.check_file_contents('b', 'a first file\n')
556
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
559
open('c', 'wb').write('c this file\n')
561
self.failIf(os.path.lexists('c'))
562
self.check_file_contents('b', 'c this file\n')
564
# TODO: Try to write a test for atomicity
565
# TODO: Test moving into a non-existant subdirectory
566
# TODO: Test Transport.move_multi
569
t = self.get_transport()
574
open('a', 'wb').write('a file\n')
576
self.check_file_contents('b', 'a file\n')
578
self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
580
# What should the assert be if you try to copy a
581
# file over a directory?
582
#self.assertRaises(Something, t.copy, 'a', 'c')
583
open('d', 'wb').write('text in d\n')
585
self.check_file_contents('b', 'text in d\n')
587
# TODO: test copy_multi
589
def test_connection_error(self):
590
"""ConnectionError is raised when connection is impossible"""
591
if not hasattr(self, "get_bogus_transport"):
593
t = self.get_bogus_transport()
596
except (ConnectionError, NoSuchFile), e:
598
except (Exception), e:
599
self.failIf(True, 'Wrong exception thrown: %s' % e)
601
self.failIf(True, 'Did not get the expected exception.')
604
# TODO: Test stat, just try once, and if it throws, stop testing
605
from stat import S_ISDIR, S_ISREG
607
t = self.get_transport()
611
except TransportNotPossible, e:
612
# This transport cannot stat
615
paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
616
self.build_tree(paths)
622
local_st = os.stat(p)
624
self.failUnless(S_ISDIR(st.st_mode))
626
self.failUnless(S_ISREG(st.st_mode))
627
self.assertEqual(local_st.st_size, st.st_size)
628
self.assertEqual(local_st.st_mode, st.st_mode)
629
local_stats.append(local_st)
631
remote_stats = list(t.stat_multi(paths))
632
remote_iter_stats = list(t.stat_multi(iter(paths)))
634
for local, remote, remote_iter in \
635
zip(local_stats, remote_stats, remote_iter_stats):
636
self.assertEqual(local.st_mode, remote.st_mode)
637
self.assertEqual(local.st_mode, remote_iter.st_mode)
639
self.assertEqual(local.st_size, remote.st_size)
640
self.assertEqual(local.st_size, remote_iter.st_size)
641
# Should we test UID/GID?
643
self.assertRaises(NoSuchFile, t.stat, 'q')
644
self.assertRaises(NoSuchFile, t.stat, 'b/a')
646
self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
647
self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
649
def test_list_dir(self):
650
# TODO: Test list_dir, just try once, and if it throws, stop testing
651
t = self.get_transport()
654
self.assertRaises(TransportNotPossible, t.list_dir, '.')
658
l = list(t.list_dir(d))
662
# SftpServer creates control files in the working directory
663
# so lets move down a directory to be safe
668
self.assertEqual([], sorted_list(u'.'))
669
self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e'])
671
self.assertEqual([u'a', u'b', u'c'], sorted_list(u'.'))
672
self.assertEqual([u'd', u'e'], sorted_list(u'c'))
676
self.assertEqual([u'a', u'c'], sorted_list('.'))
677
self.assertEqual([u'e'], sorted_list(u'c'))
679
self.assertListRaises(NoSuchFile, t.list_dir, 'q')
680
self.assertListRaises(NoSuchFile, t.list_dir, 'c/f')
681
self.assertListRaises(NoSuchFile, t.list_dir, 'a')
683
def test_clone(self):
684
# TODO: Test that clone moves up and down the filesystem
685
t1 = self.get_transport()
687
self.build_tree(['a', 'b/', 'b/c'])
689
self.failUnless(t1.has('a'))
690
self.failUnless(t1.has('b/c'))
691
self.failIf(t1.has('c'))
694
self.failUnless(t2.has('c'))
695
self.failIf(t2.has('a'))
698
self.failUnless(t3.has('a'))
699
self.failIf(t3.has('c'))
701
self.failIf(t1.has('b/d'))
702
self.failIf(t2.has('d'))
703
self.failIf(t3.has('b/d'))
706
open('b/d', 'wb').write('newfile\n')
708
t2.put('d', 'newfile\n')
710
self.failUnless(t1.has('b/d'))
711
self.failUnless(t2.has('d'))
712
self.failUnless(t3.has('b/d'))
715
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
716
def get_transport(self):
717
from bzrlib.transport.local import LocalTransport
718
return LocalTransport(u'.')
721
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
725
def get_transport(self):
726
from bzrlib.transport.http import HttpTransport
727
url = self.get_remote_url(u'.')
728
return HttpTransport(url)
730
def get_bogus_transport(self):
731
from bzrlib.transport.http import HttpTransport
732
return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
735
class TestMemoryTransport(TestCase):
737
def test_get_transport(self):
738
memory.MemoryTransport()
740
def test_clone(self):
741
transport = memory.MemoryTransport()
742
self.failUnless(transport.clone() is transport)
744
def test_abspath(self):
745
transport = memory.MemoryTransport()
746
self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
748
def test_relpath(self):
749
transport = memory.MemoryTransport()
751
def test_append_and_get(self):
752
transport = memory.MemoryTransport()
753
transport.append('path', StringIO('content'))
754
self.assertEqual(transport.get('path').read(), 'content')
755
transport.append('path', StringIO('content'))
756
self.assertEqual(transport.get('path').read(), 'contentcontent')
758
def test_put_and_get(self):
759
transport = memory.MemoryTransport()
760
transport.put('path', StringIO('content'))
761
self.assertEqual(transport.get('path').read(), 'content')
762
transport.put('path', StringIO('content'))
763
self.assertEqual(transport.get('path').read(), 'content')
765
def test_append_without_dir_fails(self):
766
transport = memory.MemoryTransport()
767
self.assertRaises(NoSuchFile,
768
transport.append, 'dir/path', StringIO('content'))
770
def test_put_without_dir_fails(self):
771
transport = memory.MemoryTransport()
772
self.assertRaises(NoSuchFile,
773
transport.put, 'dir/path', StringIO('content'))
775
def test_get_missing(self):
776
transport = memory.MemoryTransport()
777
self.assertRaises(NoSuchFile, transport.get, 'foo')
779
def test_has_missing(self):
780
transport = memory.MemoryTransport()
781
self.assertEquals(False, transport.has('foo'))
783
def test_has_present(self):
784
transport = memory.MemoryTransport()
785
transport.append('foo', StringIO('content'))
786
self.assertEquals(True, transport.has('foo'))
788
def test_mkdir(self):
789
transport = memory.MemoryTransport()
790
transport.mkdir('dir')
791
transport.append('dir/path', StringIO('content'))
792
self.assertEqual(transport.get('dir/path').read(), 'content')
794
def test_mkdir_missing_parent(self):
795
transport = memory.MemoryTransport()
796
self.assertRaises(NoSuchFile,
797
transport.mkdir, 'dir/dir')
799
def test_mkdir_twice(self):
800
transport = memory.MemoryTransport()
801
transport.mkdir('dir')
802
self.assertRaises(FileExists, transport.mkdir, 'dir')
804
def test_parameters(self):
805
transport = memory.MemoryTransport()
806
self.assertEqual(True, transport.listable())
807
self.assertEqual(False, transport.should_cache())
809
def test_iter_files_recursive(self):
810
transport = memory.MemoryTransport()
811
transport.mkdir('dir')
812
transport.put('dir/foo', StringIO('content'))
813
transport.put('dir/bar', StringIO('content'))
814
transport.put('bar', StringIO('content'))
815
paths = set(transport.iter_files_recursive())
816
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
819
transport = memory.MemoryTransport()
820
transport.put('foo', StringIO('content'))
821
transport.put('bar', StringIO('phowar'))
822
self.assertEqual(7, transport.stat('foo').st_size)
823
self.assertEqual(6, transport.stat('bar').st_size)