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, TransportNotPossible
22
from bzrlib.selftest import TestCase, TestCaseInTempDir
23
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
24
from bzrlib.transport import memory
28
"""Append the given text (file-like object) to the supplied filename."""
35
class TestTransportMixIn(object):
36
"""Subclass this, and it will provide a series of tests for a Transport.
37
It assumes that the Transport object is connected to the
38
current working directory. So that whatever is done
39
through the transport, should show up in the working
40
directory, and vice-versa.
42
This also tests to make sure that the functions work with both
43
generators and lists (assuming iter(list) is effectively a generator)
46
def get_transport(self):
47
"""Children should override this to return the Transport object.
49
raise NotImplementedError
52
t = self.get_transport()
54
files = ['a', 'b', 'e', 'g']
55
self.build_tree(files)
56
self.assertEqual(t.has('a'), True)
57
self.assertEqual(t.has('c'), False)
58
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
59
[True, True, False, False, True, False, True, False])
60
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
61
[True, True, False, False, True, False, True, False])
64
t = self.get_transport()
66
files = ['a', 'b', 'e', 'g']
67
self.build_tree(files)
68
self.assertEqual(t.get('a').read(), open('a').read())
69
content_f = t.get_multi(files)
70
for path,f in zip(files, content_f):
71
self.assertEqual(open(path).read(), f.read())
73
content_f = t.get_multi(iter(files))
74
for path,f in zip(files, content_f):
75
self.assertEqual(open(path).read(), f.read())
77
self.assertRaises(NoSuchFile, t.get, 'c')
79
files = list(t.get_multi(['a', 'b', 'c']))
83
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
85
files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
89
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
92
t = self.get_transport()
95
self.assertRaises(TransportNotPossible,
96
t.put, 'a', 'some text for a\n')
97
open('a', 'wb').write('some text for a\n')
99
t.put('a', 'some text for a\n')
100
self.assert_(os.path.exists('a'))
101
self.check_file_contents('a', 'some text for a\n')
102
self.assertEqual(t.get('a').read(), 'some text for a\n')
103
# Make sure 'has' is updated
104
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
105
[True, False, False, False, False])
107
self.assertRaises(TransportNotPossible,
109
[('a', 'new\ncontents for\na\n'),
110
('d', 'contents\nfor d\n')])
111
open('a', 'wb').write('new\ncontents for\na\n')
112
open('d', 'wb').write('contents\nfor d\n')
114
# Put also replaces contents
115
self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
116
('d', 'contents\nfor d\n')]),
118
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
119
[True, False, False, True, False])
120
self.check_file_contents('a', 'new\ncontents for\na\n')
121
self.check_file_contents('d', 'contents\nfor d\n')
124
self.assertRaises(TransportNotPossible,
125
t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
126
('d', 'another contents\nfor d\n')]))
127
open('a', 'wb').write('diff\ncontents for\na\n')
128
open('d', 'wb').write('another contents\nfor d\n')
131
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
132
('d', 'another contents\nfor d\n')]))
134
self.check_file_contents('a', 'diff\ncontents for\na\n')
135
self.check_file_contents('d', 'another contents\nfor d\n')
138
self.assertRaises(TransportNotPossible,
139
t.put, 'path/doesnt/exist/c', 'contents')
141
self.assertRaises(NoSuchFile,
142
t.put, 'path/doesnt/exist/c', 'contents')
144
def test_put_file(self):
145
t = self.get_transport()
147
# Test that StringIO can be used as a file-like object with put
148
f1 = StringIO('this is a string\nand some more stuff\n')
150
open('f1', 'wb').write(f1.read())
156
self.check_file_contents('f1',
157
'this is a string\nand some more stuff\n')
159
f2 = StringIO('here is some text\nand a bit more\n')
160
f3 = StringIO('some text for the\nthird file created\n')
163
open('f2', 'wb').write(f2.read())
164
open('f3', 'wb').write(f3.read())
166
t.put_multi([('f2', f2), ('f3', f3)])
170
self.check_file_contents('f2', 'here is some text\nand a bit more\n')
171
self.check_file_contents('f3', 'some text for the\nthird file created\n')
173
# Test that an actual file object can be used with put
174
f4 = open('f1', 'rb')
176
open('f4', 'wb').write(f4.read())
182
self.check_file_contents('f4',
183
'this is a string\nand some more stuff\n')
185
f5 = open('f2', 'rb')
186
f6 = open('f3', 'rb')
188
open('f5', 'wb').write(f5.read())
189
open('f6', 'wb').write(f6.read())
191
t.put_multi([('f5', f5), ('f6', f6)])
195
self.check_file_contents('f5', 'here is some text\nand a bit more\n')
196
self.check_file_contents('f6', 'some text for the\nthird file created\n')
200
def test_mkdir(self):
201
t = self.get_transport()
205
self.assertEqual(t.has('dir_a'), True)
206
self.assertEqual(t.has('dir_b'), False)
209
self.assertRaises(TransportNotPossible,
214
self.assertEqual(t.has('dir_b'), True)
215
self.assert_(os.path.isdir('dir_b'))
218
self.assertRaises(TransportNotPossible,
219
t.mkdir_multi, ['dir_c', 'dir_d'])
223
t.mkdir_multi(['dir_c', 'dir_d'])
226
self.assertRaises(TransportNotPossible,
227
t.mkdir_multi, iter(['dir_e', 'dir_f']))
231
t.mkdir_multi(iter(['dir_e', 'dir_f']))
232
self.assertEqual(list(t.has_multi(
233
['dir_a', 'dir_b', 'dir_c', 'dir_q',
234
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
235
[True, True, True, False,
236
True, True, True, True])
237
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
238
self.assert_(os.path.isdir(d))
240
if not self.readonly:
241
self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
242
self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
244
# Make sure the transport recognizes when a
245
# directory is created by other means
246
# Caching Transports will fail, because dir_e was already seen not
247
# to exist. So instead, we will search for a new directory
249
#if not self.readonly:
250
# self.assertRaises(FileExists, t.mkdir, 'dir_e')
253
if not self.readonly:
254
self.assertRaises(FileExists, t.mkdir, 'dir_g')
256
# Test get/put in sub-directories
258
open('dir_a/a', 'wb').write('contents of dir_a/a')
259
open('dir_b/b', 'wb').write('contents of dir_b/b')
262
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
263
('dir_b/b', 'contents of dir_b/b')])
265
for f in ('dir_a/a', 'dir_b/b'):
266
self.assertEqual(t.get(f).read(), open(f).read())
268
def test_copy_to(self):
270
from bzrlib.transport.local import LocalTransport
272
t = self.get_transport()
274
files = ['a', 'b', 'c', 'd']
275
self.build_tree(files)
277
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
278
dtmp_base = os.path.basename(dtmp)
279
local_t = LocalTransport(dtmp)
281
t.copy_to(files, local_t)
283
self.assertEquals(open(f).read(),
284
open(os.path.join(dtmp_base, f)).read())
286
del dtmp, dtmp_base, local_t
288
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
289
dtmp_base = os.path.basename(dtmp)
290
local_t = LocalTransport(dtmp)
292
files = ['a', 'b', 'c', 'd']
293
t.copy_to(iter(files), local_t)
295
self.assertEquals(open(f).read(),
296
open(os.path.join(dtmp_base, f)).read())
298
del dtmp, dtmp_base, local_t
300
def test_append(self):
301
t = self.get_transport()
304
open('a', 'wb').write('diff\ncontents for\na\n')
305
open('b', 'wb').write('contents\nfor b\n')
308
('a', 'diff\ncontents for\na\n'),
309
('b', 'contents\nfor b\n')
313
self.assertRaises(TransportNotPossible,
314
t.append, 'a', 'add\nsome\nmore\ncontents\n')
315
_append('a', 'add\nsome\nmore\ncontents\n')
317
t.append('a', 'add\nsome\nmore\ncontents\n')
319
self.check_file_contents('a',
320
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
323
self.assertRaises(TransportNotPossible,
325
[('a', 'and\nthen\nsome\nmore\n'),
326
('b', 'some\nmore\nfor\nb\n')])
327
_append('a', 'and\nthen\nsome\nmore\n')
328
_append('b', 'some\nmore\nfor\nb\n')
330
t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
331
('b', 'some\nmore\nfor\nb\n')])
332
self.check_file_contents('a',
333
'diff\ncontents for\na\n'
334
'add\nsome\nmore\ncontents\n'
335
'and\nthen\nsome\nmore\n')
336
self.check_file_contents('b',
338
'some\nmore\nfor\nb\n')
341
_append('a', 'a little bit more\n')
342
_append('b', 'from an iterator\n')
344
t.append_multi(iter([('a', 'a little bit more\n'),
345
('b', 'from an iterator\n')]))
346
self.check_file_contents('a',
347
'diff\ncontents for\na\n'
348
'add\nsome\nmore\ncontents\n'
349
'and\nthen\nsome\nmore\n'
350
'a little bit more\n')
351
self.check_file_contents('b',
353
'some\nmore\nfor\nb\n'
354
'from an iterator\n')
356
def test_append_file(self):
357
t = self.get_transport()
360
('f1', 'this is a string\nand some more stuff\n'),
361
('f2', 'here is some text\nand a bit more\n'),
362
('f3', 'some text for the\nthird file created\n'),
363
('f4', 'this is a string\nand some more stuff\n'),
364
('f5', 'here is some text\nand a bit more\n'),
365
('f6', 'some text for the\nthird file created\n')
369
for f, val in contents:
370
open(f, 'wb').write(val)
372
t.put_multi(contents)
374
a1 = StringIO('appending to\none\n')
376
_append('f1', a1.read())
382
self.check_file_contents('f1',
383
'this is a string\nand some more stuff\n'
384
'appending to\none\n')
386
a2 = StringIO('adding more\ntext to two\n')
387
a3 = StringIO('some garbage\nto put in three\n')
390
_append('f2', a2.read())
391
_append('f3', a3.read())
393
t.append_multi([('f2', a2), ('f3', a3)])
397
self.check_file_contents('f2',
398
'here is some text\nand a bit more\n'
399
'adding more\ntext to two\n')
400
self.check_file_contents('f3',
401
'some text for the\nthird file created\n'
402
'some garbage\nto put in three\n')
404
# Test that an actual file object can be used with put
405
a4 = open('f1', 'rb')
407
_append('f4', a4.read())
413
self.check_file_contents('f4',
414
'this is a string\nand some more stuff\n'
415
'this is a string\nand some more stuff\n'
416
'appending to\none\n')
418
a5 = open('f2', 'rb')
419
a6 = open('f3', 'rb')
421
_append('f5', a5.read())
422
_append('f6', a6.read())
424
t.append_multi([('f5', a5), ('f6', a6)])
428
self.check_file_contents('f5',
429
'here is some text\nand a bit more\n'
430
'here is some text\nand a bit more\n'
431
'adding more\ntext to two\n')
432
self.check_file_contents('f6',
433
'some text for the\nthird file created\n'
434
'some text for the\nthird file created\n'
435
'some garbage\nto put in three\n')
437
def test_delete(self):
438
# TODO: Test Transport.delete
442
# TODO: Test Transport.move
446
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
447
def get_transport(self):
448
from bzrlib.transport.local import LocalTransport
449
return LocalTransport('.')
452
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
456
def get_transport(self):
457
from bzrlib.transport.http import HttpTransport
458
url = self.get_remote_url('.')
459
return HttpTransport(url)
462
class TestMemoryTransport(TestCase):
464
def test_get_transport(self):
465
memory.MemoryTransport()
467
def test_clone(self):
468
transport = memory.MemoryTransport()
469
self.failUnless(transport.clone() is transport)
471
def test_abspath(self):
472
transport = memory.MemoryTransport()
473
self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
475
def test_relpath(self):
476
transport = memory.MemoryTransport()
478
def test_append_and_get(self):
479
transport = memory.MemoryTransport()
480
transport.append('path', StringIO('content'))
481
self.assertEqual(transport.get('path').read(), 'content')
482
transport.append('path', StringIO('content'))
483
self.assertEqual(transport.get('path').read(), 'contentcontent')
485
def test_put_and_get(self):
486
transport = memory.MemoryTransport()
487
transport.put('path', StringIO('content'))
488
self.assertEqual(transport.get('path').read(), 'content')
489
transport.put('path', StringIO('content'))
490
self.assertEqual(transport.get('path').read(), 'content')
492
def test_append_without_dir_fails(self):
493
transport = memory.MemoryTransport()
494
self.assertRaises(NoSuchFile,
495
transport.append, 'dir/path', StringIO('content'))
497
def test_put_without_dir_fails(self):
498
transport = memory.MemoryTransport()
499
self.assertRaises(NoSuchFile,
500
transport.put, 'dir/path', StringIO('content'))
502
def test_get_missing(self):
503
transport = memory.MemoryTransport()
504
self.assertRaises(NoSuchFile, transport.get, 'foo')
506
def test_has_missing(self):
507
transport = memory.MemoryTransport()
508
self.assertEquals(False, transport.has('foo'))
510
def test_has_present(self):
511
transport = memory.MemoryTransport()
512
transport.append('foo', StringIO('content'))
513
self.assertEquals(True, transport.has('foo'))
515
def test_mkdir(self):
516
transport = memory.MemoryTransport()
517
transport.mkdir('dir')
518
transport.append('dir/path', StringIO('content'))
519
self.assertEqual(transport.get('dir/path').read(), 'content')
521
def test_mkdir_missing_parent(self):
522
transport = memory.MemoryTransport()
523
self.assertRaises(NoSuchFile,
524
transport.mkdir, 'dir/dir')
526
def test_mkdir_twice(self):
527
transport = memory.MemoryTransport()
528
transport.mkdir('dir')
529
self.assertRaises(FileExists, transport.mkdir, 'dir')
531
def test_parameters(self):
532
transport = memory.MemoryTransport()
533
self.assertEqual(True, transport.listable())
534
self.assertEqual(False, transport.should_cache())
536
def test_iter_files_recursive(self):
537
transport = memory.MemoryTransport()
538
transport.mkdir('dir')
539
transport.put('dir/foo', StringIO('content'))
540
transport.put('dir/bar', StringIO('content'))
541
transport.put('bar', StringIO('content'))
542
paths = set(transport.iter_files_recursive())
543
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
546
transport = memory.MemoryTransport()
547
transport.put('foo', StringIO('content'))
548
transport.put('bar', StringIO('phowar'))
549
self.assertEqual(7, transport.stat('foo').st_size)
550
self.assertEqual(6, transport.stat('bar').st_size)
added
removed
bzrlib/selftest/testtransport.py
