1
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
1
# Copyright (C) 2004, 2005 by Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
11
# GNU General Public License for more details.
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
19
from cStringIO import StringIO
26
from bzrlib.errors import (DependencyNotPresent,
21
from bzrlib.errors import (NoSuchFile, FileExists, TransportNotPossible,
34
23
from bzrlib.tests import TestCase, TestCaseInTempDir
35
from bzrlib.transport import (_clear_protocol_handlers,
38
_get_protocol_handlers,
39
_set_protocol_handlers,
40
_get_transport_modules,
43
register_lazy_transport,
44
register_transport_proto,
47
from bzrlib.transport.chroot import ChrootServer
48
from bzrlib.transport.memory import MemoryTransport
49
from bzrlib.transport.local import (LocalTransport,
50
EmulatedWin32LocalTransport)
53
# TODO: Should possibly split transport-specific tests into their own files.
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."""
56
36
class TestTransport(TestCase):
57
37
"""Test the non transport-concrete class functionality."""
59
def test__get_set_protocol_handlers(self):
60
handlers = _get_protocol_handlers()
61
self.assertNotEqual([], handlers.keys( ))
63
_clear_protocol_handlers()
64
self.assertEqual([], _get_protocol_handlers().keys())
66
_set_protocol_handlers(handlers)
68
def test_get_transport_modules(self):
69
handlers = _get_protocol_handlers()
70
# don't pollute the current handlers
71
_clear_protocol_handlers()
72
class SampleHandler(object):
73
"""I exist, isnt that enough?"""
75
_clear_protocol_handlers()
76
register_transport_proto('foo')
77
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
78
'TestTransport.SampleHandler')
79
register_transport_proto('bar')
80
register_lazy_transport('bar', 'bzrlib.tests.test_transport',
81
'TestTransport.SampleHandler')
82
self.assertEqual([SampleHandler.__module__,
83
'bzrlib.transport.chroot'],
84
_get_transport_modules())
86
_set_protocol_handlers(handlers)
88
def test_transport_dependency(self):
89
"""Transport with missing dependency causes no error"""
90
saved_handlers = _get_protocol_handlers()
91
# don't pollute the current handlers
92
_clear_protocol_handlers()
94
register_transport_proto('foo')
95
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
96
'BadTransportHandler')
98
get_transport('foo://fooserver/foo')
99
except UnsupportedProtocol, e:
101
self.assertEquals('Unsupported protocol'
102
' for url "foo://fooserver/foo":'
103
' Unable to import library "some_lib":'
104
' testing missing dependency', str(e))
106
self.fail('Did not raise UnsupportedProtocol')
108
# restore original values
109
_set_protocol_handlers(saved_handlers)
111
def test_transport_fallback(self):
112
"""Transport with missing dependency causes no error"""
113
saved_handlers = _get_protocol_handlers()
115
_clear_protocol_handlers()
116
register_transport_proto('foo')
117
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
118
'BackupTransportHandler')
119
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
120
'BadTransportHandler')
121
t = get_transport('foo://fooserver/foo')
122
self.assertTrue(isinstance(t, BackupTransportHandler))
124
_set_protocol_handlers(saved_handlers)
126
def test_LateReadError(self):
127
"""The LateReadError helper should raise on read()."""
128
a_file = LateReadError('a path')
131
except ReadError, error:
132
self.assertEqual('a path', error.path)
133
self.assertRaises(ReadError, a_file.read, 40)
136
def test__combine_paths(self):
138
self.assertEqual('/home/sarah/project/foo',
139
t._combine_paths('/home/sarah', 'project/foo'))
140
self.assertEqual('/etc',
141
t._combine_paths('/home/sarah', '../../etc'))
142
self.assertEqual('/etc',
143
t._combine_paths('/home/sarah', '../../../etc'))
144
self.assertEqual('/etc',
145
t._combine_paths('/home/sarah', '/etc'))
147
def test_local_abspath_non_local_transport(self):
148
# the base implementation should throw
149
t = MemoryTransport()
150
e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
151
self.assertEqual('memory:///t is not a local path.', str(e))
154
class TestCoalesceOffsets(TestCase):
156
def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
157
coalesce = Transport._coalesce_offsets
158
exp = [_CoalescedOffset(*x) for x in expected]
159
out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
161
self.assertEqual(exp, out)
163
def test_coalesce_empty(self):
166
def test_coalesce_simple(self):
167
self.check([(0, 10, [(0, 10)])], [(0, 10)])
169
def test_coalesce_unrelated(self):
170
self.check([(0, 10, [(0, 10)]),
172
], [(0, 10), (20, 10)])
174
def test_coalesce_unsorted(self):
175
self.check([(20, 10, [(0, 10)]),
177
], [(20, 10), (0, 10)])
179
def test_coalesce_nearby(self):
180
self.check([(0, 20, [(0, 10), (10, 10)])],
183
def test_coalesce_overlapped(self):
184
self.assertRaises(ValueError,
185
self.check, [(0, 15, [(0, 10), (5, 10)])],
188
def test_coalesce_limit(self):
189
self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
190
(30, 10), (40, 10)]),
191
(60, 50, [(0, 10), (10, 10), (20, 10),
192
(30, 10), (40, 10)]),
193
], [(10, 10), (20, 10), (30, 10), (40, 10),
194
(50, 10), (60, 10), (70, 10), (80, 10),
195
(90, 10), (100, 10)],
198
def test_coalesce_no_limit(self):
199
self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
200
(30, 10), (40, 10), (50, 10),
201
(60, 10), (70, 10), (80, 10),
203
], [(10, 10), (20, 10), (30, 10), (40, 10),
204
(50, 10), (60, 10), (70, 10), (80, 10),
205
(90, 10), (100, 10)])
207
def test_coalesce_fudge(self):
208
self.check([(10, 30, [(0, 10), (20, 10)]),
209
(100, 10, [(0, 10),]),
210
], [(10, 10), (30, 10), (100, 10)],
213
def test_coalesce_max_size(self):
214
self.check([(10, 20, [(0, 10), (10, 10)]),
216
# If one range is above max_size, it gets its own coalesced
218
(100, 80, [(0, 80),]),],
219
[(10, 10), (20, 10), (30, 50), (100, 80)],
223
def test_coalesce_no_max_size(self):
224
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
225
[(10, 10), (20, 10), (30, 50), (80, 100)],
228
def test_coalesce_default_limit(self):
229
# By default we use a 100MB max size.
230
ten_mb = 10*1024*1024
231
self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
232
(10*ten_mb, ten_mb, [(0, ten_mb)])],
233
[(i*ten_mb, ten_mb) for i in range(11)])
234
self.check([(0, 11*ten_mb, [(i*ten_mb, ten_mb) for i in range(11)]),],
235
[(i*ten_mb, ten_mb) for i in range(11)],
236
max_size=1*1024*1024*1024)
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(t.has_any(['a', 'b', 'c']), True)
70
self.assertEqual(t.has_any(['c', 'd', 'f', urlescape('%%')]), False)
71
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
72
[True, True, False, False, True, False, True, False])
73
self.assertEqual(t.has_any(['c', 'c', 'c']), False)
74
self.assertEqual(t.has_any(['b', 'b', 'b']), True)
77
t = self.get_transport()
79
files = ['a', 'b', 'e', 'g']
80
self.build_tree(files)
81
self.assertEqual(t.get('a').read(), open('a').read())
82
content_f = t.get_multi(files)
83
for path,f in zip(files, content_f):
84
self.assertEqual(open(path).read(), f.read())
86
content_f = t.get_multi(iter(files))
87
for path,f in zip(files, content_f):
88
self.assertEqual(open(path).read(), f.read())
90
self.assertRaises(NoSuchFile, t.get, 'c')
92
files = list(t.get_multi(['a', 'b', 'c']))
96
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
98
files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
102
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
105
t = self.get_transport()
108
self.assertRaises(TransportNotPossible,
109
t.put, 'a', 'some text for a\n')
110
open('a', 'wb').write('some text for a\n')
112
t.put('a', 'some text for a\n')
113
self.assert_(os.path.exists('a'))
114
self.check_file_contents('a', 'some text for a\n')
115
self.assertEqual(t.get('a').read(), 'some text for a\n')
116
# Make sure 'has' is updated
117
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
118
[True, False, False, False, False])
120
self.assertRaises(TransportNotPossible,
122
[('a', 'new\ncontents for\na\n'),
123
('d', 'contents\nfor d\n')])
124
open('a', 'wb').write('new\ncontents for\na\n')
125
open('d', 'wb').write('contents\nfor d\n')
127
# Put also replaces contents
128
self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
129
('d', 'contents\nfor d\n')]),
131
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
132
[True, False, False, True, False])
133
self.check_file_contents('a', 'new\ncontents for\na\n')
134
self.check_file_contents('d', 'contents\nfor d\n')
137
self.assertRaises(TransportNotPossible,
138
t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
139
('d', 'another contents\nfor d\n')]))
140
open('a', 'wb').write('diff\ncontents for\na\n')
141
open('d', 'wb').write('another contents\nfor d\n')
144
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
145
('d', 'another contents\nfor d\n')]))
147
self.check_file_contents('a', 'diff\ncontents for\na\n')
148
self.check_file_contents('d', 'another contents\nfor d\n')
151
self.assertRaises(TransportNotPossible,
152
t.put, 'path/doesnt/exist/c', 'contents')
154
self.assertRaises(NoSuchFile,
155
t.put, 'path/doesnt/exist/c', 'contents')
157
def test_put_file(self):
158
t = self.get_transport()
160
# Test that StringIO can be used as a file-like object with put
161
f1 = StringIO('this is a string\nand some more stuff\n')
163
open('f1', 'wb').write(f1.read())
169
self.check_file_contents('f1',
170
'this is a string\nand some more stuff\n')
172
f2 = StringIO('here is some text\nand a bit more\n')
173
f3 = StringIO('some text for the\nthird file created\n')
176
open('f2', 'wb').write(f2.read())
177
open('f3', 'wb').write(f3.read())
179
t.put_multi([('f2', f2), ('f3', f3)])
183
self.check_file_contents('f2', 'here is some text\nand a bit more\n')
184
self.check_file_contents('f3', 'some text for the\nthird file created\n')
186
# Test that an actual file object can be used with put
187
f4 = open('f1', 'rb')
189
open('f4', 'wb').write(f4.read())
195
self.check_file_contents('f4',
196
'this is a string\nand some more stuff\n')
198
f5 = open('f2', 'rb')
199
f6 = open('f3', 'rb')
201
open('f5', 'wb').write(f5.read())
202
open('f6', 'wb').write(f6.read())
204
t.put_multi([('f5', f5), ('f6', f6)])
208
self.check_file_contents('f5', 'here is some text\nand a bit more\n')
209
self.check_file_contents('f6', 'some text for the\nthird file created\n')
213
def test_mkdir(self):
214
t = self.get_transport()
218
self.assertEqual(t.has('dir_a'), True)
219
self.assertEqual(t.has('dir_b'), False)
222
self.assertRaises(TransportNotPossible,
227
self.assertEqual(t.has('dir_b'), True)
228
self.assert_(os.path.isdir('dir_b'))
231
self.assertRaises(TransportNotPossible,
232
t.mkdir_multi, ['dir_c', 'dir_d'])
236
t.mkdir_multi(['dir_c', 'dir_d'])
239
self.assertRaises(TransportNotPossible,
240
t.mkdir_multi, iter(['dir_e', 'dir_f']))
244
t.mkdir_multi(iter(['dir_e', 'dir_f']))
245
self.assertEqual(list(t.has_multi(
246
['dir_a', 'dir_b', 'dir_c', 'dir_q',
247
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
248
[True, True, True, False,
249
True, True, True, True])
250
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
251
self.assert_(os.path.isdir(d))
253
if not self.readonly:
254
self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
255
self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
257
# Make sure the transport recognizes when a
258
# directory is created by other means
259
# Caching Transports will fail, because dir_e was already seen not
260
# to exist. So instead, we will search for a new directory
262
#if not self.readonly:
263
# self.assertRaises(FileExists, t.mkdir, 'dir_e')
266
if not self.readonly:
267
self.assertRaises(FileExists, t.mkdir, 'dir_g')
269
# Test get/put in sub-directories
271
open('dir_a/a', 'wb').write('contents of dir_a/a')
272
open('dir_b/b', 'wb').write('contents of dir_b/b')
275
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
276
('dir_b/b', 'contents of dir_b/b')])
278
for f in ('dir_a/a', 'dir_b/b'):
279
self.assertEqual(t.get(f).read(), open(f).read())
281
def test_copy_to(self):
283
from bzrlib.transport.local import LocalTransport
285
t = self.get_transport()
287
files = ['a', 'b', 'c', 'd']
288
self.build_tree(files)
290
dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
291
dtmp_base = os.path.basename(dtmp)
292
local_t = LocalTransport(dtmp)
294
t.copy_to(files, local_t)
296
self.assertEquals(open(f).read(),
297
open(os.path.join(dtmp_base, f)).read())
299
# Test that copying into a missing directory raises
302
open('e/f', 'wb').write('contents of e')
303
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
305
os.mkdir(os.path.join(dtmp_base, 'e'))
306
t.copy_to(['e/f'], local_t)
308
del dtmp, dtmp_base, local_t
310
dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
311
dtmp_base = os.path.basename(dtmp)
312
local_t = LocalTransport(dtmp)
314
files = ['a', 'b', 'c', 'd']
315
t.copy_to(iter(files), local_t)
317
self.assertEquals(open(f).read(),
318
open(os.path.join(dtmp_base, f)).read())
320
del dtmp, dtmp_base, local_t
322
def test_append(self):
323
t = self.get_transport()
326
open('a', 'wb').write('diff\ncontents for\na\n')
327
open('b', 'wb').write('contents\nfor b\n')
330
('a', 'diff\ncontents for\na\n'),
331
('b', 'contents\nfor b\n')
335
self.assertRaises(TransportNotPossible,
336
t.append, 'a', 'add\nsome\nmore\ncontents\n')
337
_append('a', 'add\nsome\nmore\ncontents\n')
339
t.append('a', 'add\nsome\nmore\ncontents\n')
341
self.check_file_contents('a',
342
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
345
self.assertRaises(TransportNotPossible,
347
[('a', 'and\nthen\nsome\nmore\n'),
348
('b', 'some\nmore\nfor\nb\n')])
349
_append('a', 'and\nthen\nsome\nmore\n')
350
_append('b', 'some\nmore\nfor\nb\n')
352
t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
353
('b', 'some\nmore\nfor\nb\n')])
354
self.check_file_contents('a',
355
'diff\ncontents for\na\n'
356
'add\nsome\nmore\ncontents\n'
357
'and\nthen\nsome\nmore\n')
358
self.check_file_contents('b',
360
'some\nmore\nfor\nb\n')
363
_append('a', 'a little bit more\n')
364
_append('b', 'from an iterator\n')
366
t.append_multi(iter([('a', 'a little bit more\n'),
367
('b', 'from an iterator\n')]))
368
self.check_file_contents('a',
369
'diff\ncontents for\na\n'
370
'add\nsome\nmore\ncontents\n'
371
'and\nthen\nsome\nmore\n'
372
'a little bit more\n')
373
self.check_file_contents('b',
375
'some\nmore\nfor\nb\n'
376
'from an iterator\n')
378
def test_append_file(self):
379
t = self.get_transport()
382
('f1', 'this is a string\nand some more stuff\n'),
383
('f2', 'here is some text\nand a bit more\n'),
384
('f3', 'some text for the\nthird file created\n'),
385
('f4', 'this is a string\nand some more stuff\n'),
386
('f5', 'here is some text\nand a bit more\n'),
387
('f6', 'some text for the\nthird file created\n')
391
for f, val in contents:
392
open(f, 'wb').write(val)
394
t.put_multi(contents)
396
a1 = StringIO('appending to\none\n')
398
_append('f1', a1.read())
404
self.check_file_contents('f1',
405
'this is a string\nand some more stuff\n'
406
'appending to\none\n')
408
a2 = StringIO('adding more\ntext to two\n')
409
a3 = StringIO('some garbage\nto put in three\n')
412
_append('f2', a2.read())
413
_append('f3', a3.read())
415
t.append_multi([('f2', a2), ('f3', a3)])
419
self.check_file_contents('f2',
420
'here is some text\nand a bit more\n'
421
'adding more\ntext to two\n')
422
self.check_file_contents('f3',
423
'some text for the\nthird file created\n'
424
'some garbage\nto put in three\n')
426
# Test that an actual file object can be used with put
427
a4 = open('f1', 'rb')
429
_append('f4', a4.read())
435
self.check_file_contents('f4',
436
'this is a string\nand some more stuff\n'
437
'this is a string\nand some more stuff\n'
438
'appending to\none\n')
440
a5 = open('f2', 'rb')
441
a6 = open('f3', 'rb')
443
_append('f5', a5.read())
444
_append('f6', a6.read())
446
t.append_multi([('f5', a5), ('f6', a6)])
450
self.check_file_contents('f5',
451
'here is some text\nand a bit more\n'
452
'here is some text\nand a bit more\n'
453
'adding more\ntext to two\n')
454
self.check_file_contents('f6',
455
'some text for the\nthird file created\n'
456
'some text for the\nthird file created\n'
457
'some garbage\nto put in three\n')
459
def test_delete(self):
460
# TODO: Test Transport.delete
461
t = self.get_transport()
464
# TODO: Test Transport.move
465
t = self.get_transport()
468
# TODO: Test Transport.move
469
t = self.get_transport()
471
def test_connection_error(self):
472
"""ConnectionError is raised when connection is impossible"""
473
if not hasattr(self, "get_bogus_transport"):
475
t = self.get_bogus_transport()
478
except (ConnectionError, NoSuchFile), e:
480
except (Exception), e:
481
self.failIf(True, 'Wrong exception thrown: %s' % e)
483
self.failIf(True, 'Did not get the expected exception.')
486
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
487
def get_transport(self):
488
from bzrlib.transport.local import LocalTransport
489
return LocalTransport(u'.')
492
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
496
def get_transport(self):
497
from bzrlib.transport.http import HttpTransport
498
url = self.get_remote_url(u'.')
499
return HttpTransport(url)
501
def get_bogus_transport(self):
502
from bzrlib.transport.http import HttpTransport
503
return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
239
506
class TestMemoryTransport(TestCase):
241
508
def test_get_transport(self):
509
memory.MemoryTransport()
244
511
def test_clone(self):
245
transport = MemoryTransport()
246
self.assertTrue(isinstance(transport, MemoryTransport))
247
self.assertEqual("memory:///", transport.clone("/").base)
512
transport = memory.MemoryTransport()
513
self.failUnless(transport.clone() is transport)
249
515
def test_abspath(self):
250
transport = MemoryTransport()
251
self.assertEqual("memory:///relpath", transport.abspath('relpath'))
253
def test_abspath_of_root(self):
254
transport = MemoryTransport()
255
self.assertEqual("memory:///", transport.base)
256
self.assertEqual("memory:///", transport.abspath('/'))
258
def test_abspath_of_relpath_starting_at_root(self):
259
transport = MemoryTransport()
260
self.assertEqual("memory:///foo", transport.abspath('/foo'))
516
transport = memory.MemoryTransport()
517
self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
519
def test_relpath(self):
520
transport = memory.MemoryTransport()
262
522
def test_append_and_get(self):
263
transport = MemoryTransport()
264
transport.append_bytes('path', 'content')
523
transport = memory.MemoryTransport()
524
transport.append('path', StringIO('content'))
265
525
self.assertEqual(transport.get('path').read(), 'content')
266
transport.append_file('path', StringIO('content'))
526
transport.append('path', StringIO('content'))
267
527
self.assertEqual(transport.get('path').read(), 'contentcontent')
269
529
def test_put_and_get(self):
270
transport = MemoryTransport()
271
transport.put_file('path', StringIO('content'))
530
transport = memory.MemoryTransport()
531
transport.put('path', StringIO('content'))
272
532
self.assertEqual(transport.get('path').read(), 'content')
273
transport.put_bytes('path', 'content')
533
transport.put('path', StringIO('content'))
274
534
self.assertEqual(transport.get('path').read(), 'content')
276
536
def test_append_without_dir_fails(self):
277
transport = MemoryTransport()
537
transport = memory.MemoryTransport()
278
538
self.assertRaises(NoSuchFile,
279
transport.append_bytes, 'dir/path', 'content')
539
transport.append, 'dir/path', StringIO('content'))
281
541
def test_put_without_dir_fails(self):
282
transport = MemoryTransport()
542
transport = memory.MemoryTransport()
283
543
self.assertRaises(NoSuchFile,
284
transport.put_file, 'dir/path', StringIO('content'))
544
transport.put, 'dir/path', StringIO('content'))
286
546
def test_get_missing(self):
287
transport = MemoryTransport()
547
transport = memory.MemoryTransport()
288
548
self.assertRaises(NoSuchFile, transport.get, 'foo')
290
550
def test_has_missing(self):
291
transport = MemoryTransport()
551
transport = memory.MemoryTransport()
292
552
self.assertEquals(False, transport.has('foo'))
294
554
def test_has_present(self):
295
transport = MemoryTransport()
296
transport.append_bytes('foo', 'content')
555
transport = memory.MemoryTransport()
556
transport.append('foo', StringIO('content'))
297
557
self.assertEquals(True, transport.has('foo'))
299
def test_list_dir(self):
300
transport = MemoryTransport()
301
transport.put_bytes('foo', 'content')
302
transport.mkdir('dir')
303
transport.put_bytes('dir/subfoo', 'content')
304
transport.put_bytes('dirlike', 'content')
306
self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
307
self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
309
559
def test_mkdir(self):
310
transport = MemoryTransport()
560
transport = memory.MemoryTransport()
311
561
transport.mkdir('dir')
312
transport.append_bytes('dir/path', 'content')
562
transport.append('dir/path', StringIO('content'))
313
563
self.assertEqual(transport.get('dir/path').read(), 'content')
315
565
def test_mkdir_missing_parent(self):
316
transport = MemoryTransport()
566
transport = memory.MemoryTransport()
317
567
self.assertRaises(NoSuchFile,
318
568
transport.mkdir, 'dir/dir')
320
570
def test_mkdir_twice(self):
321
transport = MemoryTransport()
571
transport = memory.MemoryTransport()
322
572
transport.mkdir('dir')
323
573
self.assertRaises(FileExists, transport.mkdir, 'dir')
325
575
def test_parameters(self):
326
transport = MemoryTransport()
576
transport = memory.MemoryTransport()
327
577
self.assertEqual(True, transport.listable())
328
self.assertEqual(False, transport.is_readonly())
578
self.assertEqual(False, transport.should_cache())
330
580
def test_iter_files_recursive(self):
331
transport = MemoryTransport()
581
transport = memory.MemoryTransport()
332
582
transport.mkdir('dir')
333
transport.put_bytes('dir/foo', 'content')
334
transport.put_bytes('dir/bar', 'content')
335
transport.put_bytes('bar', 'content')
583
transport.put('dir/foo', StringIO('content'))
584
transport.put('dir/bar', StringIO('content'))
585
transport.put('bar', StringIO('content'))
336
586
paths = set(transport.iter_files_recursive())
337
587
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
339
589
def test_stat(self):
340
transport = MemoryTransport()
341
transport.put_bytes('foo', 'content')
342
transport.put_bytes('bar', 'phowar')
590
transport = memory.MemoryTransport()
591
transport.put('foo', StringIO('content'))
592
transport.put('bar', StringIO('phowar'))
343
593
self.assertEqual(7, transport.stat('foo').st_size)
344
594
self.assertEqual(6, transport.stat('bar').st_size)
347
class ChrootDecoratorTransportTest(TestCase):
348
"""Chroot decoration specific tests."""
350
def test_abspath(self):
351
# The abspath is always relative to the chroot_url.
352
server = ChrootServer(get_transport('memory:///foo/bar/'))
354
transport = get_transport(server.get_url())
355
self.assertEqual(server.get_url(), transport.abspath('/'))
357
subdir_transport = transport.clone('subdir')
358
self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
361
def test_clone(self):
362
server = ChrootServer(get_transport('memory:///foo/bar/'))
364
transport = get_transport(server.get_url())
365
# relpath from root and root path are the same
366
relpath_cloned = transport.clone('foo')
367
abspath_cloned = transport.clone('/foo')
368
self.assertEqual(server, relpath_cloned.server)
369
self.assertEqual(server, abspath_cloned.server)
372
def test_chroot_url_preserves_chroot(self):
373
"""Calling get_transport on a chroot transport's base should produce a
374
transport with exactly the same behaviour as the original chroot
377
This is so that it is not possible to escape a chroot by doing::
378
url = chroot_transport.base
379
parent_url = urlutils.join(url, '..')
380
new_transport = get_transport(parent_url)
382
server = ChrootServer(get_transport('memory:///path/subpath'))
384
transport = get_transport(server.get_url())
385
new_transport = get_transport(transport.base)
386
self.assertEqual(transport.server, new_transport.server)
387
self.assertEqual(transport.base, new_transport.base)
390
def test_urljoin_preserves_chroot(self):
391
"""Using urlutils.join(url, '..') on a chroot URL should not produce a
392
URL that escapes the intended chroot.
394
This is so that it is not possible to escape a chroot by doing::
395
url = chroot_transport.base
396
parent_url = urlutils.join(url, '..')
397
new_transport = get_transport(parent_url)
399
server = ChrootServer(get_transport('memory:///path/'))
401
transport = get_transport(server.get_url())
403
InvalidURLJoin, urlutils.join, transport.base, '..')
407
class ChrootServerTest(TestCase):
409
def test_construct(self):
410
backing_transport = MemoryTransport()
411
server = ChrootServer(backing_transport)
412
self.assertEqual(backing_transport, server.backing_transport)
414
def test_setUp(self):
415
backing_transport = MemoryTransport()
416
server = ChrootServer(backing_transport)
418
self.assertTrue(server.scheme in _get_protocol_handlers().keys())
420
def test_tearDown(self):
421
backing_transport = MemoryTransport()
422
server = ChrootServer(backing_transport)
425
self.assertFalse(server.scheme in _get_protocol_handlers().keys())
427
def test_get_url(self):
428
backing_transport = MemoryTransport()
429
server = ChrootServer(backing_transport)
431
self.assertEqual('chroot-%d:///' % id(server), server.get_url())
435
class ReadonlyDecoratorTransportTest(TestCase):
436
"""Readonly decoration specific tests."""
438
def test_local_parameters(self):
439
import bzrlib.transport.readonly as readonly
440
# connect to . in readonly mode
441
transport = readonly.ReadonlyTransportDecorator('readonly+.')
442
self.assertEqual(True, transport.listable())
443
self.assertEqual(True, transport.is_readonly())
445
def test_http_parameters(self):
446
from bzrlib.tests.http_server import HttpServer
447
import bzrlib.transport.readonly as readonly
448
# connect to '.' via http which is not listable
449
server = HttpServer()
452
transport = get_transport('readonly+' + server.get_url())
453
self.failUnless(isinstance(transport,
454
readonly.ReadonlyTransportDecorator))
455
self.assertEqual(False, transport.listable())
456
self.assertEqual(True, transport.is_readonly())
461
class FakeNFSDecoratorTests(TestCaseInTempDir):
462
"""NFS decorator specific tests."""
464
def get_nfs_transport(self, url):
465
import bzrlib.transport.fakenfs as fakenfs
466
# connect to url with nfs decoration
467
return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
469
def test_local_parameters(self):
470
# the listable and is_readonly parameters
471
# are not changed by the fakenfs decorator
472
transport = self.get_nfs_transport('.')
473
self.assertEqual(True, transport.listable())
474
self.assertEqual(False, transport.is_readonly())
476
def test_http_parameters(self):
477
# the listable and is_readonly parameters
478
# are not changed by the fakenfs decorator
479
from bzrlib.tests.http_server import HttpServer
480
# connect to '.' via http which is not listable
481
server = HttpServer()
484
transport = self.get_nfs_transport(server.get_url())
485
self.assertIsInstance(
486
transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
487
self.assertEqual(False, transport.listable())
488
self.assertEqual(True, transport.is_readonly())
492
def test_fakenfs_server_default(self):
493
# a FakeNFSServer() should bring up a local relpath server for itself
494
import bzrlib.transport.fakenfs as fakenfs
495
server = fakenfs.FakeNFSServer()
498
# the url should be decorated appropriately
499
self.assertStartsWith(server.get_url(), 'fakenfs+')
500
# and we should be able to get a transport for it
501
transport = get_transport(server.get_url())
502
# which must be a FakeNFSTransportDecorator instance.
503
self.assertIsInstance(
504
transport, fakenfs.FakeNFSTransportDecorator)
508
def test_fakenfs_rename_semantics(self):
509
# a FakeNFS transport must mangle the way rename errors occur to
510
# look like NFS problems.
511
transport = self.get_nfs_transport('.')
512
self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
514
self.assertRaises(errors.ResourceBusy,
515
transport.rename, 'from', 'to')
518
class FakeVFATDecoratorTests(TestCaseInTempDir):
519
"""Tests for simulation of VFAT restrictions"""
521
def get_vfat_transport(self, url):
522
"""Return vfat-backed transport for test directory"""
523
from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
524
return FakeVFATTransportDecorator('vfat+' + url)
526
def test_transport_creation(self):
527
from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
528
transport = self.get_vfat_transport('.')
529
self.assertIsInstance(transport, FakeVFATTransportDecorator)
531
def test_transport_mkdir(self):
532
transport = self.get_vfat_transport('.')
533
transport.mkdir('HELLO')
534
self.assertTrue(transport.has('hello'))
535
self.assertTrue(transport.has('Hello'))
537
def test_forbidden_chars(self):
538
transport = self.get_vfat_transport('.')
539
self.assertRaises(ValueError, transport.has, "<NU>")
542
class BadTransportHandler(Transport):
543
def __init__(self, base_url):
544
raise DependencyNotPresent('some_lib', 'testing missing dependency')
547
class BackupTransportHandler(Transport):
548
"""Test transport that works as a backup for the BadTransportHandler"""
552
class TestTransportImplementation(TestCaseInTempDir):
553
"""Implementation verification for transports.
555
To verify a transport we need a server factory, which is a callable
556
that accepts no parameters and returns an implementation of
557
bzrlib.transport.Server.
559
That Server is then used to construct transport instances and test
560
the transport via loopback activity.
562
Currently this assumes that the Transport object is connected to the
563
current working directory. So that whatever is done
564
through the transport, should show up in the working
565
directory, and vice-versa. This is a bug, because its possible to have
566
URL schemes which provide access to something that may not be
567
result in storage on the local disk, i.e. due to file system limits, or
568
due to it being a database or some other non-filesystem tool.
570
This also tests to make sure that the functions work with both
571
generators and lists (assuming iter(list) is effectively a generator)
575
super(TestTransportImplementation, self).setUp()
576
self._server = self.transport_server()
578
self.addCleanup(self._server.tearDown)
580
def get_transport(self, relpath=None):
581
"""Return a connected transport to the local directory.
583
:param relpath: a path relative to the base url.
585
base_url = self._server.get_url()
586
url = self._adjust_url(base_url, relpath)
587
# try getting the transport via the regular interface:
588
t = get_transport(url)
589
# vila--20070607 if the following are commented out the test suite
590
# still pass. Is this really still needed or was it a forgotten
592
if not isinstance(t, self.transport_class):
593
# we did not get the correct transport class type. Override the
594
# regular connection behaviour by direct construction.
595
t = self.transport_class(url)
599
class TestLocalTransports(TestCase):
601
def test_get_transport_from_abspath(self):
602
here = osutils.abspath('.')
603
t = get_transport(here)
604
self.assertIsInstance(t, LocalTransport)
605
self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
607
def test_get_transport_from_relpath(self):
608
here = osutils.abspath('.')
609
t = get_transport('.')
610
self.assertIsInstance(t, LocalTransport)
611
self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
613
def test_get_transport_from_local_url(self):
614
here = osutils.abspath('.')
615
here_url = urlutils.local_path_to_url(here) + '/'
616
t = get_transport(here_url)
617
self.assertIsInstance(t, LocalTransport)
618
self.assertEquals(t.base, here_url)
620
def test_local_abspath(self):
621
here = osutils.abspath('.')
622
t = get_transport(here)
623
self.assertEquals(t.local_abspath(''), here)
626
class TestWin32LocalTransport(TestCase):
628
def test_unc_clone_to_root(self):
629
# Win32 UNC path like \\HOST\path
630
# clone to root should stop at least at \\HOST part
632
t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
635
self.assertEquals(t.base, 'file://HOST/')
636
# make sure we reach the root
638
self.assertEquals(t.base, 'file://HOST/')
641
class TestConnectedTransport(TestCase):
642
"""Tests for connected to remote server transports"""
644
def test_parse_url(self):
645
t = ConnectedTransport('http://simple.example.com/home/source')
646
self.assertEquals(t._host, 'simple.example.com')
647
self.assertEquals(t._port, None)
648
self.assertEquals(t._path, '/home/source/')
649
self.failUnless(t._user is None)
650
self.failUnless(t._password is None)
652
self.assertEquals(t.base, 'http://simple.example.com/home/source/')
654
def test_parse_url_with_at_in_user(self):
656
t = ConnectedTransport('ftp://user@host.com@www.host.com/')
657
self.assertEquals(t._user, 'user@host.com')
659
def test_parse_quoted_url(self):
660
t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
661
self.assertEquals(t._host, 'exAmple.com')
662
self.assertEquals(t._port, 2222)
663
self.assertEquals(t._user, 'robey')
664
self.assertEquals(t._password, 'h@t')
665
self.assertEquals(t._path, '/path/')
667
# Base should not keep track of the password
668
self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
670
def test_parse_invalid_url(self):
671
self.assertRaises(errors.InvalidURL,
673
'sftp://lily.org:~janneke/public/bzr/gub')
675
def test_relpath(self):
676
t = ConnectedTransport('sftp://user@host.com/abs/path')
678
self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
679
self.assertRaises(errors.PathNotChild, t.relpath,
680
'http://user@host.com/abs/path/sub')
681
self.assertRaises(errors.PathNotChild, t.relpath,
682
'sftp://user2@host.com/abs/path/sub')
683
self.assertRaises(errors.PathNotChild, t.relpath,
684
'sftp://user@otherhost.com/abs/path/sub')
685
self.assertRaises(errors.PathNotChild, t.relpath,
686
'sftp://user@host.com:33/abs/path/sub')
687
# Make sure it works when we don't supply a username
688
t = ConnectedTransport('sftp://host.com/abs/path')
689
self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
691
# Make sure it works when parts of the path will be url encoded
692
t = ConnectedTransport('sftp://host.com/dev/%path')
693
self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
695
def test_connection_sharing_propagate_credentials(self):
696
t = ConnectedTransport('ftp://user@host.com/abs/path')
697
self.assertEquals('user', t._user)
698
self.assertEquals('host.com', t._host)
699
self.assertIs(None, t._get_connection())
700
self.assertIs(None, t._password)
701
c = t.clone('subdir')
702
self.assertIs(None, c._get_connection())
703
self.assertIs(None, t._password)
705
# Simulate the user entering a password
707
connection = object()
708
t._set_connection(connection, password)
709
self.assertIs(connection, t._get_connection())
710
self.assertIs(password, t._get_credentials())
711
self.assertIs(connection, c._get_connection())
712
self.assertIs(password, c._get_credentials())
714
# credentials can be updated
715
new_password = 'even more secret'
716
c._update_credentials(new_password)
717
self.assertIs(connection, t._get_connection())
718
self.assertIs(new_password, t._get_credentials())
719
self.assertIs(connection, c._get_connection())
720
self.assertIs(new_password, c._get_credentials())
723
class TestReusedTransports(TestCase):
724
"""Tests for transport reuse"""
726
def test_reuse_same_transport(self):
727
possible_transports = []
728
t1 = get_transport('http://foo/',
729
possible_transports=possible_transports)
730
self.assertEqual([t1], possible_transports)
731
t2 = get_transport('http://foo/', possible_transports=[t1])
732
self.assertIs(t1, t2)
734
# Also check that final '/' are handled correctly
735
t3 = get_transport('http://foo/path/')
736
t4 = get_transport('http://foo/path', possible_transports=[t3])
737
self.assertIs(t3, t4)
739
t5 = get_transport('http://foo/path')
740
t6 = get_transport('http://foo/path/', possible_transports=[t5])
741
self.assertIs(t5, t6)
743
def test_don_t_reuse_different_transport(self):
744
t1 = get_transport('http://foo/path')
745
t2 = get_transport('http://bar/path', possible_transports=[t1])
746
self.assertIsNot(t1, t2)
749
class TestTransportTrace(TestCase):
752
transport = get_transport('trace+memory://')
753
self.assertIsInstance(
754
transport, bzrlib.transport.trace.TransportTraceDecorator)
756
def test_clone_preserves_activity(self):
757
transport = get_transport('trace+memory://')
758
transport2 = transport.clone('.')
759
self.assertTrue(transport is not transport2)
760
self.assertTrue(transport._activity is transport2._activity)
762
# the following specific tests are for the operations that have made use of
763
# logging in tests; we could test every single operation but doing that
764
# still won't cause a test failure when the top level Transport API
765
# changes; so there is little return doing that.
767
transport = get_transport('trace+memory:///')
768
transport.put_bytes('foo', 'barish')
771
# put_bytes records the bytes, not the content to avoid memory
773
expected_result.append(('put_bytes', 'foo', 6, None))
774
# get records the file name only.
775
expected_result.append(('get', 'foo'))
776
self.assertEqual(expected_result, transport._activity)
778
def test_readv(self):
779
transport = get_transport('trace+memory:///')
780
transport.put_bytes('foo', 'barish')
781
list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
784
# put_bytes records the bytes, not the content to avoid memory
786
expected_result.append(('put_bytes', 'foo', 6, None))
787
# readv records the supplied offset request
788
expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
789
self.assertEqual(expected_result, transport._activity)
added
removed
bzrlib/tests/test_transport.py
