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,
23
from bzrlib.selftest import TestCase, TestCaseInTempDir
24
from bzrlib.selftest.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
60
t = self.get_transport()
62
files = ['a', 'b', 'e', 'g', '%']
63
self.build_tree(files)
64
self.assertEqual(t.has('a'), True)
65
self.assertEqual(t.has('c'), False)
66
self.assertEqual(t.has(urlescape('%')), True)
67
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
68
[True, True, False, False, True, False, True, False])
69
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
70
[True, True, False, False, True, False, True, False])
73
t = self.get_transport()
75
files = ['a', 'b', 'e', 'g']
76
self.build_tree(files)
77
self.assertEqual(t.get('a').read(), open('a').read())
78
content_f = t.get_multi(files)
79
for path,f in zip(files, content_f):
80
self.assertEqual(open(path).read(), f.read())
82
content_f = t.get_multi(iter(files))
83
for path,f in zip(files, content_f):
84
self.assertEqual(open(path).read(), f.read())
86
self.assertRaises(NoSuchFile, t.get, 'c')
88
files = list(t.get_multi(['a', 'b', 'c']))
92
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
94
files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
98
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
101
t = self.get_transport()
104
self.assertRaises(TransportNotPossible,
105
t.put, 'a', 'some text for a\n')
106
open('a', 'wb').write('some text for a\n')
108
t.put('a', 'some text for a\n')
109
self.assert_(os.path.exists('a'))
110
self.check_file_contents('a', 'some text for a\n')
111
self.assertEqual(t.get('a').read(), 'some text for a\n')
112
# Make sure 'has' is updated
113
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
114
[True, False, False, False, False])
116
self.assertRaises(TransportNotPossible,
118
[('a', 'new\ncontents for\na\n'),
119
('d', 'contents\nfor d\n')])
120
open('a', 'wb').write('new\ncontents for\na\n')
121
open('d', 'wb').write('contents\nfor d\n')
123
# Put also replaces contents
124
self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
125
('d', 'contents\nfor d\n')]),
127
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
128
[True, False, False, True, False])
129
self.check_file_contents('a', 'new\ncontents for\na\n')
130
self.check_file_contents('d', 'contents\nfor d\n')
133
self.assertRaises(TransportNotPossible,
134
t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
135
('d', 'another contents\nfor d\n')]))
136
open('a', 'wb').write('diff\ncontents for\na\n')
137
open('d', 'wb').write('another contents\nfor d\n')
140
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
141
('d', 'another contents\nfor d\n')]))
143
self.check_file_contents('a', 'diff\ncontents for\na\n')
144
self.check_file_contents('d', 'another contents\nfor d\n')
147
self.assertRaises(TransportNotPossible,
148
t.put, 'path/doesnt/exist/c', 'contents')
150
self.assertRaises(NoSuchFile,
151
t.put, 'path/doesnt/exist/c', 'contents')
153
def test_put_file(self):
154
t = self.get_transport()
156
# Test that StringIO can be used as a file-like object with put
157
f1 = StringIO('this is a string\nand some more stuff\n')
159
open('f1', 'wb').write(f1.read())
165
self.check_file_contents('f1',
166
'this is a string\nand some more stuff\n')
168
f2 = StringIO('here is some text\nand a bit more\n')
169
f3 = StringIO('some text for the\nthird file created\n')
172
open('f2', 'wb').write(f2.read())
173
open('f3', 'wb').write(f3.read())
175
t.put_multi([('f2', f2), ('f3', f3)])
179
self.check_file_contents('f2', 'here is some text\nand a bit more\n')
180
self.check_file_contents('f3', 'some text for the\nthird file created\n')
182
# Test that an actual file object can be used with put
183
f4 = open('f1', 'rb')
185
open('f4', 'wb').write(f4.read())
191
self.check_file_contents('f4',
192
'this is a string\nand some more stuff\n')
194
f5 = open('f2', 'rb')
195
f6 = open('f3', 'rb')
197
open('f5', 'wb').write(f5.read())
198
open('f6', 'wb').write(f6.read())
200
t.put_multi([('f5', f5), ('f6', f6)])
204
self.check_file_contents('f5', 'here is some text\nand a bit more\n')
205
self.check_file_contents('f6', 'some text for the\nthird file created\n')
209
def test_mkdir(self):
210
t = self.get_transport()
214
self.assertEqual(t.has('dir_a'), True)
215
self.assertEqual(t.has('dir_b'), False)
218
self.assertRaises(TransportNotPossible,
223
self.assertEqual(t.has('dir_b'), True)
224
self.assert_(os.path.isdir('dir_b'))
227
self.assertRaises(TransportNotPossible,
228
t.mkdir_multi, ['dir_c', 'dir_d'])
232
t.mkdir_multi(['dir_c', 'dir_d'])
235
self.assertRaises(TransportNotPossible,
236
t.mkdir_multi, iter(['dir_e', 'dir_f']))
240
t.mkdir_multi(iter(['dir_e', 'dir_f']))
241
self.assertEqual(list(t.has_multi(
242
['dir_a', 'dir_b', 'dir_c', 'dir_q',
243
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
244
[True, True, True, False,
245
True, True, True, True])
246
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
247
self.assert_(os.path.isdir(d))
249
if not self.readonly:
250
self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
251
self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
253
# Make sure the transport recognizes when a
254
# directory is created by other means
255
# Caching Transports will fail, because dir_e was already seen not
256
# to exist. So instead, we will search for a new directory
258
#if not self.readonly:
259
# self.assertRaises(FileExists, t.mkdir, 'dir_e')
262
if not self.readonly:
263
self.assertRaises(FileExists, t.mkdir, 'dir_g')
265
# Test get/put in sub-directories
267
open('dir_a/a', 'wb').write('contents of dir_a/a')
268
open('dir_b/b', 'wb').write('contents of dir_b/b')
271
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
272
('dir_b/b', 'contents of dir_b/b')])
274
for f in ('dir_a/a', 'dir_b/b'):
275
self.assertEqual(t.get(f).read(), open(f).read())
277
def test_copy_to(self):
279
from bzrlib.transport.local import LocalTransport
281
t = self.get_transport()
283
files = ['a', 'b', 'c', 'd']
284
self.build_tree(files)
286
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
287
dtmp_base = os.path.basename(dtmp)
288
local_t = LocalTransport(dtmp)
290
t.copy_to(files, local_t)
292
self.assertEquals(open(f).read(),
293
open(os.path.join(dtmp_base, f)).read())
295
del dtmp, dtmp_base, local_t
297
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
298
dtmp_base = os.path.basename(dtmp)
299
local_t = LocalTransport(dtmp)
301
files = ['a', 'b', 'c', 'd']
302
t.copy_to(iter(files), local_t)
304
self.assertEquals(open(f).read(),
305
open(os.path.join(dtmp_base, f)).read())
307
del dtmp, dtmp_base, local_t
309
def test_append(self):
310
t = self.get_transport()
313
open('a', 'wb').write('diff\ncontents for\na\n')
314
open('b', 'wb').write('contents\nfor b\n')
317
('a', 'diff\ncontents for\na\n'),
318
('b', 'contents\nfor b\n')
322
self.assertRaises(TransportNotPossible,
323
t.append, 'a', 'add\nsome\nmore\ncontents\n')
324
_append('a', 'add\nsome\nmore\ncontents\n')
326
t.append('a', 'add\nsome\nmore\ncontents\n')
328
self.check_file_contents('a',
329
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
332
self.assertRaises(TransportNotPossible,
334
[('a', 'and\nthen\nsome\nmore\n'),
335
('b', 'some\nmore\nfor\nb\n')])
336
_append('a', 'and\nthen\nsome\nmore\n')
337
_append('b', 'some\nmore\nfor\nb\n')
339
t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
340
('b', 'some\nmore\nfor\nb\n')])
341
self.check_file_contents('a',
342
'diff\ncontents for\na\n'
343
'add\nsome\nmore\ncontents\n'
344
'and\nthen\nsome\nmore\n')
345
self.check_file_contents('b',
347
'some\nmore\nfor\nb\n')
350
_append('a', 'a little bit more\n')
351
_append('b', 'from an iterator\n')
353
t.append_multi(iter([('a', 'a little bit more\n'),
354
('b', 'from an iterator\n')]))
355
self.check_file_contents('a',
356
'diff\ncontents for\na\n'
357
'add\nsome\nmore\ncontents\n'
358
'and\nthen\nsome\nmore\n'
359
'a little bit more\n')
360
self.check_file_contents('b',
362
'some\nmore\nfor\nb\n'
363
'from an iterator\n')
365
def test_append_file(self):
366
t = self.get_transport()
369
('f1', 'this is a string\nand some more stuff\n'),
370
('f2', 'here is some text\nand a bit more\n'),
371
('f3', 'some text for the\nthird file created\n'),
372
('f4', 'this is a string\nand some more stuff\n'),
373
('f5', 'here is some text\nand a bit more\n'),
374
('f6', 'some text for the\nthird file created\n')
378
for f, val in contents:
379
open(f, 'wb').write(val)
381
t.put_multi(contents)
383
a1 = StringIO('appending to\none\n')
385
_append('f1', a1.read())
391
self.check_file_contents('f1',
392
'this is a string\nand some more stuff\n'
393
'appending to\none\n')
395
a2 = StringIO('adding more\ntext to two\n')
396
a3 = StringIO('some garbage\nto put in three\n')
399
_append('f2', a2.read())
400
_append('f3', a3.read())
402
t.append_multi([('f2', a2), ('f3', a3)])
406
self.check_file_contents('f2',
407
'here is some text\nand a bit more\n'
408
'adding more\ntext to two\n')
409
self.check_file_contents('f3',
410
'some text for the\nthird file created\n'
411
'some garbage\nto put in three\n')
413
# Test that an actual file object can be used with put
414
a4 = open('f1', 'rb')
416
_append('f4', a4.read())
422
self.check_file_contents('f4',
423
'this is a string\nand some more stuff\n'
424
'this is a string\nand some more stuff\n'
425
'appending to\none\n')
427
a5 = open('f2', 'rb')
428
a6 = open('f3', 'rb')
430
_append('f5', a5.read())
431
_append('f6', a6.read())
433
t.append_multi([('f5', a5), ('f6', a6)])
437
self.check_file_contents('f5',
438
'here is some text\nand a bit more\n'
439
'here is some text\nand a bit more\n'
440
'adding more\ntext to two\n')
441
self.check_file_contents('f6',
442
'some text for the\nthird file created\n'
443
'some text for the\nthird file created\n'
444
'some garbage\nto put in three\n')
446
def test_delete(self):
447
# TODO: Test Transport.delete
451
# TODO: Test Transport.move
454
def test_connection_error(self):
455
"""ConnectionError is raised when connection is impossible"""
456
if not hasattr(self, "get_bogus_transport"):
458
t = self.get_bogus_transport()
459
self.assertRaises(ConnectionError, t.get, '.bzr/branch')
462
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
463
def get_transport(self):
464
from bzrlib.transport.local import LocalTransport
465
return LocalTransport('.')
468
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
472
def get_transport(self):
473
from bzrlib.transport.http import HttpTransport
474
url = self.get_remote_url('.')
475
return HttpTransport(url)
477
def get_bogus_transport(self):
478
from bzrlib.transport.http import HttpTransport
479
return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
482
class TestMemoryTransport(TestCase):
484
def test_get_transport(self):
485
memory.MemoryTransport()
487
def test_clone(self):
488
transport = memory.MemoryTransport()
489
self.failUnless(transport.clone() is transport)
491
def test_abspath(self):
492
transport = memory.MemoryTransport()
493
self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
495
def test_relpath(self):
496
transport = memory.MemoryTransport()
498
def test_append_and_get(self):
499
transport = memory.MemoryTransport()
500
transport.append('path', StringIO('content'))
501
self.assertEqual(transport.get('path').read(), 'content')
502
transport.append('path', StringIO('content'))
503
self.assertEqual(transport.get('path').read(), 'contentcontent')
505
def test_put_and_get(self):
506
transport = memory.MemoryTransport()
507
transport.put('path', StringIO('content'))
508
self.assertEqual(transport.get('path').read(), 'content')
509
transport.put('path', StringIO('content'))
510
self.assertEqual(transport.get('path').read(), 'content')
512
def test_append_without_dir_fails(self):
513
transport = memory.MemoryTransport()
514
self.assertRaises(NoSuchFile,
515
transport.append, 'dir/path', StringIO('content'))
517
def test_put_without_dir_fails(self):
518
transport = memory.MemoryTransport()
519
self.assertRaises(NoSuchFile,
520
transport.put, 'dir/path', StringIO('content'))
522
def test_get_missing(self):
523
transport = memory.MemoryTransport()
524
self.assertRaises(NoSuchFile, transport.get, 'foo')
526
def test_has_missing(self):
527
transport = memory.MemoryTransport()
528
self.assertEquals(False, transport.has('foo'))
530
def test_has_present(self):
531
transport = memory.MemoryTransport()
532
transport.append('foo', StringIO('content'))
533
self.assertEquals(True, transport.has('foo'))
535
def test_mkdir(self):
536
transport = memory.MemoryTransport()
537
transport.mkdir('dir')
538
transport.append('dir/path', StringIO('content'))
539
self.assertEqual(transport.get('dir/path').read(), 'content')
541
def test_mkdir_missing_parent(self):
542
transport = memory.MemoryTransport()
543
self.assertRaises(NoSuchFile,
544
transport.mkdir, 'dir/dir')
546
def test_mkdir_twice(self):
547
transport = memory.MemoryTransport()
548
transport.mkdir('dir')
549
self.assertRaises(FileExists, transport.mkdir, 'dir')
551
def test_parameters(self):
552
transport = memory.MemoryTransport()
553
self.assertEqual(True, transport.listable())
554
self.assertEqual(False, transport.should_cache())
556
def test_iter_files_recursive(self):
557
transport = memory.MemoryTransport()
558
transport.mkdir('dir')
559
transport.put('dir/foo', StringIO('content'))
560
transport.put('dir/bar', StringIO('content'))
561
transport.put('bar', StringIO('content'))
562
paths = set(transport.iter_files_recursive())
563
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
566
transport = memory.MemoryTransport()
567
transport.put('foo', StringIO('content'))
568
transport.put('bar', StringIO('phowar'))
569
self.assertEqual(7, transport.stat('foo').st_size)
570
self.assertEqual(6, transport.stat('bar').st_size)