20
20
TransportTestProviderAdapter.
24
25
from cStringIO import StringIO
26
from StringIO import StringIO as pyStringIO
28
from bzrlib.errors import (DirectoryNotEmpty, NoSuchFile, FileExists,
30
TransportNotPossible, ConnectionError,
37
from bzrlib.errors import (ConnectionError,
32
47
from bzrlib.osutils import getcwd
33
from bzrlib.tests import TestCaseInTempDir, TestSkipped
34
from bzrlib.transport import memory
35
import bzrlib.transport
36
import bzrlib.urlutils as urlutils
40
"""Append the given text (file-like object) to the supplied filename."""
48
class TestTransportImplementation(TestCaseInTempDir):
49
"""Implementation verification for transports.
51
To verify a transport we need a server factory, which is a callable
52
that accepts no parameters and returns an implementation of
53
bzrlib.transport.Server.
55
That Server is then used to construct transport instances and test
56
the transport via loopback activity.
58
Currently this assumes that the Transport object is connected to the
59
current working directory. So that whatever is done
60
through the transport, should show up in the working
61
directory, and vice-versa. This is a bug, because its possible to have
62
URL schemes which provide access to something that may not be
63
result in storage on the local disk, i.e. due to file system limits, or
64
due to it being a database or some other non-filesystem tool.
66
This also tests to make sure that the functions work with both
67
generators and lists (assuming iter(list) is effectively a generator)
48
from bzrlib.smart import medium
49
from bzrlib.tests import (
55
from bzrlib.tests.test_transport import TestTransportImplementation
56
from bzrlib.transport import (
59
_get_transport_modules,
61
from bzrlib.transport.memory import MemoryTransport
64
class TransportTestProviderAdapter(TestScenarioApplier):
65
"""A tool to generate a suite testing all transports for a single test.
67
This is done by copying the test once for each transport and injecting
68
the transport_class and transport_server classes into each copy. Each copy
69
is also given a new id() to make it easy to identify.
73
self.scenarios = self._test_permutations()
75
def get_transport_test_permutations(self, module):
76
"""Get the permutations module wants to have tested."""
77
if getattr(module, 'get_test_permutations', None) is None:
79
"transport module %s doesn't provide get_test_permutations()"
82
return module.get_test_permutations()
84
def _test_permutations(self):
85
"""Return a list of the klass, server_factory pairs to test."""
87
for module in _get_transport_modules():
89
permutations = self.get_transport_test_permutations(
90
reduce(getattr, (module).split('.')[1:], __import__(module)))
91
for (klass, server_factory) in permutations:
92
scenario = (server_factory.__name__,
93
{"transport_class":klass,
94
"transport_server":server_factory})
95
result.append(scenario)
96
except errors.DependencyNotPresent, e:
97
# Continue even if a dependency prevents us
98
# from adding this test
103
def load_tests(standard_tests, module, loader):
104
"""Multiply tests for tranport implementations."""
105
result = loader.suiteClass()
106
adapter = TransportTestProviderAdapter()
107
for test in tests.iter_suite_tests(standard_tests):
108
result.addTests(adapter.adapt(test))
112
class TransportTests(TestTransportImplementation):
71
super(TestTransportImplementation, self).setUp()
72
self._server = self.transport_server()
115
super(TransportTests, self).setUp()
116
self._captureVar('BZR_NO_SMART_VFS', None)
76
super(TestTransportImplementation, self).tearDown()
77
self._server.tearDown()
79
118
def check_transport_contents(self, content, transport, relpath):
80
119
"""Check that transport.get(relpath).read() == content."""
81
120
self.assertEqualDiff(content, transport.get(relpath).read())
83
def get_transport(self):
84
"""Return a connected transport to the local directory."""
85
base_url = self._server.get_url()
86
t = bzrlib.transport.get_transport(base_url)
87
if not isinstance(t, self.transport_class):
88
# we want to make sure to construct one particular class, even if
89
# there are several available implementations of this transport;
90
# therefore construct it by hand rather than through the regular
91
# get_transport method
92
t = self.transport_class(base_url)
95
def assertListRaises(self, excClass, func, *args, **kwargs):
96
"""Fail unless excClass is raised when the iterator from func is used.
98
Many transport functions can return generators this makes sure
99
to wrap them in a list() call to make sure the whole generator
100
is run, and that the proper exception is raised.
122
def test_ensure_base_missing(self):
123
""".ensure_base() should create the directory if it doesn't exist"""
124
t = self.get_transport()
126
if t_a.is_readonly():
127
self.assertRaises(TransportNotPossible,
130
self.assertTrue(t_a.ensure_base())
131
self.assertTrue(t.has('a'))
133
def test_ensure_base_exists(self):
134
""".ensure_base() should just be happy if it already exists"""
135
t = self.get_transport()
141
# ensure_base returns False if it didn't create the base
142
self.assertFalse(t_a.ensure_base())
144
def test_ensure_base_missing_parent(self):
145
""".ensure_base() will fail if the parent dir doesn't exist"""
146
t = self.get_transport()
152
self.assertRaises(NoSuchFile, t_b.ensure_base)
154
def test_external_url(self):
155
""".external_url either works or raises InProcessTransport."""
156
t = self.get_transport()
103
list(func(*args, **kwargs))
107
if hasattr(excClass,'__name__'): excName = excClass.__name__
108
else: excName = str(excClass)
109
raise self.failureException, "%s not raised" % excName
159
except errors.InProcessTransport:
111
162
def test_has(self):
112
163
t = self.get_transport()
137
199
self.build_tree(files, transport=t, line_endings='binary')
138
200
self.check_transport_contents('contents of a\n', t, 'a')
139
201
content_f = t.get_multi(files)
140
for content, f in zip(contents, content_f):
202
# Use itertools.izip() instead of use zip() or map(), since they fully
203
# evaluate their inputs, the transport requests should be issued and
204
# handled sequentially (we don't want to force transport to buffer).
205
for content, f in itertools.izip(contents, content_f):
141
206
self.assertEqual(content, f.read())
143
208
content_f = t.get_multi(iter(files))
144
for content, f in zip(contents, content_f):
209
# Use itertools.izip() for the same reason
210
for content, f in itertools.izip(contents, content_f):
145
211
self.assertEqual(content, f.read())
147
213
self.assertRaises(NoSuchFile, t.get, 'c')
148
214
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
149
215
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
152
t = self.get_transport()
155
self.assertRaises(TransportNotPossible,
156
t.put, 'a', 'some text for a\n')
159
t.put('a', StringIO('some text for a\n'))
160
self.failUnless(t.has('a'))
161
self.check_transport_contents('some text for a\n', t, 'a')
162
# Make sure 'has' is updated
163
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
164
[True, False, False, False, False])
165
# Put also replaces contents
166
self.assertEqual(t.put_multi([('a', StringIO('new\ncontents for\na\n')),
167
('d', StringIO('contents\nfor d\n'))]),
169
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
170
[True, False, False, True, False])
171
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
172
self.check_transport_contents('contents\nfor d\n', t, 'd')
175
t.put_multi(iter([('a', StringIO('diff\ncontents for\na\n')),
176
('d', StringIO('another contents\nfor d\n'))])),
178
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
179
self.check_transport_contents('another contents\nfor d\n', t, 'd')
181
self.assertRaises(NoSuchFile,
182
t.put, 'path/doesnt/exist/c', 'contents')
184
def test_put_permissions(self):
185
t = self.get_transport()
189
t.put('mode644', StringIO('test text\n'), mode=0644)
190
self.assertTransportMode(t, 'mode644', 0644)
191
t.put('mode666', StringIO('test text\n'), mode=0666)
192
self.assertTransportMode(t, 'mode666', 0666)
193
t.put('mode600', StringIO('test text\n'), mode=0600)
217
def test_get_directory_read_gives_ReadError(self):
218
"""consistent errors for read() on a file returned by get()."""
219
t = self.get_transport()
221
self.build_tree(['a directory/'])
223
t.mkdir('a%20directory')
224
# getting the file must either work or fail with a PathError
226
a_file = t.get('a%20directory')
227
except (errors.PathError, errors.RedirectRequested):
228
# early failure return immediately.
230
# having got a file, read() must either work (i.e. http reading a dir
231
# listing) or fail with ReadError
234
except errors.ReadError:
237
def test_get_bytes(self):
238
t = self.get_transport()
240
files = ['a', 'b', 'e', 'g']
241
contents = ['contents of a\n',
246
self.build_tree(files, transport=t, line_endings='binary')
247
self.check_transport_contents('contents of a\n', t, 'a')
249
for content, fname in zip(contents, files):
250
self.assertEqual(content, t.get_bytes(fname))
252
self.assertRaises(NoSuchFile, t.get_bytes, 'c')
254
def test_get_with_open_write_stream_sees_all_content(self):
255
t = self.get_transport()
258
handle = t.open_write_stream('foo')
261
self.assertEqual('b', t.get('foo').read())
265
def test_get_bytes_with_open_write_stream_sees_all_content(self):
266
t = self.get_transport()
269
handle = t.open_write_stream('foo')
272
self.assertEqual('b', t.get_bytes('foo'))
273
self.assertEqual('b', t.get('foo').read())
277
def test_put_bytes(self):
278
t = self.get_transport()
281
self.assertRaises(TransportNotPossible,
282
t.put_bytes, 'a', 'some text for a\n')
285
t.put_bytes('a', 'some text for a\n')
286
self.failUnless(t.has('a'))
287
self.check_transport_contents('some text for a\n', t, 'a')
289
# The contents should be overwritten
290
t.put_bytes('a', 'new text for a\n')
291
self.check_transport_contents('new text for a\n', t, 'a')
293
self.assertRaises(NoSuchFile,
294
t.put_bytes, 'path/doesnt/exist/c', 'contents')
296
def test_put_bytes_non_atomic(self):
297
t = self.get_transport()
300
self.assertRaises(TransportNotPossible,
301
t.put_bytes_non_atomic, 'a', 'some text for a\n')
304
self.failIf(t.has('a'))
305
t.put_bytes_non_atomic('a', 'some text for a\n')
306
self.failUnless(t.has('a'))
307
self.check_transport_contents('some text for a\n', t, 'a')
308
# Put also replaces contents
309
t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
310
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
312
# Make sure we can create another file
313
t.put_bytes_non_atomic('d', 'contents for\nd\n')
314
# And overwrite 'a' with empty contents
315
t.put_bytes_non_atomic('a', '')
316
self.check_transport_contents('contents for\nd\n', t, 'd')
317
self.check_transport_contents('', t, 'a')
319
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'no/such/path',
321
# Now test the create_parent flag
322
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
324
self.failIf(t.has('dir/a'))
325
t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
326
create_parent_dir=True)
327
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
329
# But we still get NoSuchFile if we can't make the parent dir
330
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
332
create_parent_dir=True)
334
def test_put_bytes_permissions(self):
335
t = self.get_transport()
339
if not t._can_roundtrip_unix_modebits():
340
# Can't roundtrip, so no need to run this test
342
t.put_bytes('mode644', 'test text\n', mode=0644)
343
self.assertTransportMode(t, 'mode644', 0644)
344
t.put_bytes('mode666', 'test text\n', mode=0666)
345
self.assertTransportMode(t, 'mode666', 0666)
346
t.put_bytes('mode600', 'test text\n', mode=0600)
347
self.assertTransportMode(t, 'mode600', 0600)
348
# Yes, you can put_bytes a file such that it becomes readonly
349
t.put_bytes('mode400', 'test text\n', mode=0400)
350
self.assertTransportMode(t, 'mode400', 0400)
352
# The default permissions should be based on the current umask
353
umask = osutils.get_umask()
354
t.put_bytes('nomode', 'test text\n', mode=None)
355
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
357
def test_put_bytes_non_atomic_permissions(self):
358
t = self.get_transport()
362
if not t._can_roundtrip_unix_modebits():
363
# Can't roundtrip, so no need to run this test
365
t.put_bytes_non_atomic('mode644', 'test text\n', mode=0644)
366
self.assertTransportMode(t, 'mode644', 0644)
367
t.put_bytes_non_atomic('mode666', 'test text\n', mode=0666)
368
self.assertTransportMode(t, 'mode666', 0666)
369
t.put_bytes_non_atomic('mode600', 'test text\n', mode=0600)
370
self.assertTransportMode(t, 'mode600', 0600)
371
t.put_bytes_non_atomic('mode400', 'test text\n', mode=0400)
372
self.assertTransportMode(t, 'mode400', 0400)
374
# The default permissions should be based on the current umask
375
umask = osutils.get_umask()
376
t.put_bytes_non_atomic('nomode', 'test text\n', mode=None)
377
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
379
# We should also be able to set the mode for a parent directory
381
t.put_bytes_non_atomic('dir700/mode664', 'test text\n', mode=0664,
382
dir_mode=0700, create_parent_dir=True)
383
self.assertTransportMode(t, 'dir700', 0700)
384
t.put_bytes_non_atomic('dir770/mode664', 'test text\n', mode=0664,
385
dir_mode=0770, create_parent_dir=True)
386
self.assertTransportMode(t, 'dir770', 0770)
387
t.put_bytes_non_atomic('dir777/mode664', 'test text\n', mode=0664,
388
dir_mode=0777, create_parent_dir=True)
389
self.assertTransportMode(t, 'dir777', 0777)
391
def test_put_file(self):
392
t = self.get_transport()
395
self.assertRaises(TransportNotPossible,
396
t.put_file, 'a', StringIO('some text for a\n'))
399
result = t.put_file('a', StringIO('some text for a\n'))
400
# put_file returns the length of the data written
401
self.assertEqual(16, result)
402
self.failUnless(t.has('a'))
403
self.check_transport_contents('some text for a\n', t, 'a')
404
# Put also replaces contents
405
result = t.put_file('a', StringIO('new\ncontents for\na\n'))
406
self.assertEqual(19, result)
407
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
408
self.assertRaises(NoSuchFile,
409
t.put_file, 'path/doesnt/exist/c',
410
StringIO('contents'))
412
def test_put_file_non_atomic(self):
413
t = self.get_transport()
416
self.assertRaises(TransportNotPossible,
417
t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
420
self.failIf(t.has('a'))
421
t.put_file_non_atomic('a', StringIO('some text for a\n'))
422
self.failUnless(t.has('a'))
423
self.check_transport_contents('some text for a\n', t, 'a')
424
# Put also replaces contents
425
t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
426
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
428
# Make sure we can create another file
429
t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
430
# And overwrite 'a' with empty contents
431
t.put_file_non_atomic('a', StringIO(''))
432
self.check_transport_contents('contents for\nd\n', t, 'd')
433
self.check_transport_contents('', t, 'a')
435
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
436
StringIO('contents\n'))
437
# Now test the create_parent flag
438
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
439
StringIO('contents\n'))
440
self.failIf(t.has('dir/a'))
441
t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
442
create_parent_dir=True)
443
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
445
# But we still get NoSuchFile if we can't make the parent dir
446
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
447
StringIO('contents\n'),
448
create_parent_dir=True)
450
def test_put_file_permissions(self):
452
t = self.get_transport()
456
if not t._can_roundtrip_unix_modebits():
457
# Can't roundtrip, so no need to run this test
459
t.put_file('mode644', StringIO('test text\n'), mode=0644)
460
self.assertTransportMode(t, 'mode644', 0644)
461
t.put_file('mode666', StringIO('test text\n'), mode=0666)
462
self.assertTransportMode(t, 'mode666', 0666)
463
t.put_file('mode600', StringIO('test text\n'), mode=0600)
194
464
self.assertTransportMode(t, 'mode600', 0600)
195
465
# Yes, you can put a file such that it becomes readonly
196
t.put('mode400', StringIO('test text\n'), mode=0400)
197
self.assertTransportMode(t, 'mode400', 0400)
198
t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
199
self.assertTransportMode(t, 'mmode644', 0644)
466
t.put_file('mode400', StringIO('test text\n'), mode=0400)
467
self.assertTransportMode(t, 'mode400', 0400)
468
# The default permissions should be based on the current umask
469
umask = osutils.get_umask()
470
t.put_file('nomode', StringIO('test text\n'), mode=None)
471
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
473
def test_put_file_non_atomic_permissions(self):
474
t = self.get_transport()
478
if not t._can_roundtrip_unix_modebits():
479
# Can't roundtrip, so no need to run this test
481
t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
482
self.assertTransportMode(t, 'mode644', 0644)
483
t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
484
self.assertTransportMode(t, 'mode666', 0666)
485
t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
486
self.assertTransportMode(t, 'mode600', 0600)
487
# Yes, you can put_file_non_atomic a file such that it becomes readonly
488
t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
489
self.assertTransportMode(t, 'mode400', 0400)
491
# The default permissions should be based on the current umask
492
umask = osutils.get_umask()
493
t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
494
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
496
# We should also be able to set the mode for a parent directory
499
t.put_file_non_atomic('dir700/mode664', sio, mode=0664,
500
dir_mode=0700, create_parent_dir=True)
501
self.assertTransportMode(t, 'dir700', 0700)
502
t.put_file_non_atomic('dir770/mode664', sio, mode=0664,
503
dir_mode=0770, create_parent_dir=True)
504
self.assertTransportMode(t, 'dir770', 0770)
505
t.put_file_non_atomic('dir777/mode664', sio, mode=0664,
506
dir_mode=0777, create_parent_dir=True)
507
self.assertTransportMode(t, 'dir777', 0777)
509
def test_put_bytes_unicode(self):
510
# Expect put_bytes to raise AssertionError or UnicodeEncodeError if
511
# given unicode "bytes". UnicodeEncodeError doesn't really make sense
512
# (we don't want to encode unicode here at all, callers should be
513
# strictly passing bytes to put_bytes), but we allow it for backwards
514
# compatibility. At some point we should use a specific exception.
515
# See https://bugs.launchpad.net/bzr/+bug/106898.
516
t = self.get_transport()
519
unicode_string = u'\u1234'
521
(AssertionError, UnicodeEncodeError),
522
t.put_bytes, 'foo', unicode_string)
524
def test_put_file_unicode(self):
525
# Like put_bytes, except with a StringIO.StringIO of a unicode string.
526
# This situation can happen (and has) if code is careless about the type
527
# of "string" they initialise/write to a StringIO with. We cannot use
528
# cStringIO, because it never returns unicode from read.
529
# Like put_bytes, UnicodeEncodeError isn't quite the right exception to
530
# raise, but we raise it for hysterical raisins.
531
t = self.get_transport()
534
unicode_file = pyStringIO(u'\u1234')
535
self.assertRaises(UnicodeEncodeError, t.put_file, 'foo', unicode_file)
201
537
def test_mkdir(self):
202
538
t = self.get_transport()
317
682
self.assertTransportMode(temp_transport, f, mode)
319
def test_append(self):
320
t = self.get_transport()
323
open('a', 'wb').write('diff\ncontents for\na\n')
324
open('b', 'wb').write('contents\nfor b\n')
327
('a', StringIO('diff\ncontents for\na\n')),
328
('b', StringIO('contents\nfor b\n'))
332
self.assertRaises(TransportNotPossible,
333
t.append, 'a', 'add\nsome\nmore\ncontents\n')
334
_append('a', StringIO('add\nsome\nmore\ncontents\n'))
337
t.append('a', StringIO('add\nsome\nmore\ncontents\n')))
339
self.check_transport_contents(
340
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
344
self.assertRaises(TransportNotPossible,
346
[('a', 'and\nthen\nsome\nmore\n'),
347
('b', 'some\nmore\nfor\nb\n')])
348
_append('a', StringIO('and\nthen\nsome\nmore\n'))
349
_append('b', StringIO('some\nmore\nfor\nb\n'))
351
self.assertEqual((43, 15),
352
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
353
('b', StringIO('some\nmore\nfor\nb\n'))]))
684
def test_append_file(self):
685
t = self.get_transport()
688
self.assertRaises(TransportNotPossible,
689
t.append_file, 'a', 'add\nsome\nmore\ncontents\n')
691
t.put_bytes('a', 'diff\ncontents for\na\n')
692
t.put_bytes('b', 'contents\nfor b\n')
695
t.append_file('a', StringIO('add\nsome\nmore\ncontents\n')))
697
self.check_transport_contents(
698
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
701
# a file with no parent should fail..
702
self.assertRaises(NoSuchFile,
703
t.append_file, 'missing/path', StringIO('content'))
705
# And we can create new files, too
707
t.append_file('c', StringIO('some text\nfor a missing file\n')))
708
self.check_transport_contents('some text\nfor a missing file\n',
711
def test_append_bytes(self):
712
t = self.get_transport()
715
self.assertRaises(TransportNotPossible,
716
t.append_bytes, 'a', 'add\nsome\nmore\ncontents\n')
719
self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))
720
self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))
723
t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))
725
self.check_transport_contents(
726
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
729
# a file with no parent should fail..
730
self.assertRaises(NoSuchFile,
731
t.append_bytes, 'missing/path', 'content')
733
def test_append_multi(self):
734
t = self.get_transport()
738
t.put_bytes('a', 'diff\ncontents for\na\n'
739
'add\nsome\nmore\ncontents\n')
740
t.put_bytes('b', 'contents\nfor b\n')
742
self.assertEqual((43, 15),
743
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
744
('b', StringIO('some\nmore\nfor\nb\n'))]))
354
746
self.check_transport_contents(
355
747
'diff\ncontents for\na\n'
356
748
'add\nsome\nmore\ncontents\n'
397
779
'a little bit more\n'
398
780
'some text in a\n',
400
self.check_transport_contents('some text\nfor a missing file\n',
402
782
self.check_transport_contents('missing file r\n', t, 'd')
404
# a file with no parent should fail..
405
if not t.is_readonly():
406
self.assertRaises(NoSuchFile,
407
t.append, 'missing/path',
410
def test_append_file(self):
411
t = self.get_transport()
414
('f1', StringIO('this is a string\nand some more stuff\n')),
415
('f2', StringIO('here is some text\nand a bit more\n')),
416
('f3', StringIO('some text for the\nthird file created\n')),
417
('f4', StringIO('this is a string\nand some more stuff\n')),
418
('f5', StringIO('here is some text\nand a bit more\n')),
419
('f6', StringIO('some text for the\nthird file created\n'))
423
for f, val in contents:
424
open(f, 'wb').write(val.read())
426
t.put_multi(contents)
428
a1 = StringIO('appending to\none\n')
436
self.check_transport_contents(
437
'this is a string\nand some more stuff\n'
438
'appending to\none\n',
441
a2 = StringIO('adding more\ntext to two\n')
442
a3 = StringIO('some garbage\nto put in three\n')
448
t.append_multi([('f2', a2), ('f3', a3)])
452
self.check_transport_contents(
453
'here is some text\nand a bit more\n'
454
'adding more\ntext to two\n',
456
self.check_transport_contents(
457
'some text for the\nthird file created\n'
458
'some garbage\nto put in three\n',
461
# Test that an actual file object can be used with put
470
self.check_transport_contents(
471
'this is a string\nand some more stuff\n'
472
'this is a string\nand some more stuff\n'
473
'appending to\none\n',
482
t.append_multi([('f5', a5), ('f6', a6)])
486
self.check_transport_contents(
487
'here is some text\nand a bit more\n'
488
'here is some text\nand a bit more\n'
489
'adding more\ntext to two\n',
491
self.check_transport_contents(
492
'some text for the\nthird file created\n'
493
'some text for the\nthird file created\n'
494
'some garbage\nto put in three\n',
506
t.append_multi([('a', a6), ('d', a7)])
508
self.check_transport_contents(t.get('f2').read(), t, 'c')
509
self.check_transport_contents(t.get('f3').read(), t, 'd')
511
def test_append_mode(self):
784
def test_append_file_mode(self):
785
"""Check that append accepts a mode parameter"""
512
786
# check append accepts a mode
513
787
t = self.get_transport()
514
788
if t.is_readonly():
516
t.append('f', StringIO('f'), mode=None)
789
self.assertRaises(TransportNotPossible,
790
t.append_file, 'f', StringIO('f'), mode=None)
792
t.append_file('f', StringIO('f'), mode=None)
794
def test_append_bytes_mode(self):
795
# check append_bytes accepts a mode
796
t = self.get_transport()
798
self.assertRaises(TransportNotPossible,
799
t.append_bytes, 'f', 'f', mode=None)
801
t.append_bytes('f', 'f', mode=None)
518
803
def test_delete(self):
519
804
# TODO: Test Transport.delete
752
1073
def test_list_dir(self):
753
1074
# TODO: Test list_dir, just try once, and if it throws, stop testing
754
1075
t = self.get_transport()
756
1077
if not t.listable():
757
1078
self.assertRaises(TransportNotPossible, t.list_dir, '.')
761
l = list(t.list_dir(d))
1081
def sorted_list(d, transport):
1082
l = list(transport.list_dir(d))
765
# SftpServer creates control files in the working directory
766
# so lets move down a directory to avoid those.
767
if not t.is_readonly():
773
self.assertEqual([], sorted_list(u'.'))
1086
self.assertEqual([], sorted_list('.', t))
774
1087
# c2 is precisely one letter longer than c here to test that
775
1088
# suffixing is not confused.
1089
# a%25b checks that quoting is done consistently across transports
1090
tree_names = ['a', 'a%25b', 'b', 'c/', 'c/d', 'c/e', 'c2/']
776
1092
if not t.is_readonly():
777
self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e', 'c2/'], transport=t)
1093
self.build_tree(tree_names, transport=t)
779
self.build_tree(['wd/a', 'wd/b', 'wd/c/', 'wd/c/d', 'wd/c/e', 'wd/c2/'])
781
self.assertEqual([u'a', u'b', u'c', u'c2'], sorted_list(u'.'))
782
self.assertEqual([u'd', u'e'], sorted_list(u'c'))
1095
self.build_tree(tree_names)
1098
['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('', t))
1100
['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('.', t))
1101
self.assertEqual(['d', 'e'], sorted_list('c', t))
1103
# Cloning the transport produces an equivalent listing
1104
self.assertEqual(['d', 'e'], sorted_list('', t.clone('c')))
784
1106
if not t.is_readonly():
791
self.assertEqual([u'a', u'c', u'c2'], sorted_list('.'))
792
self.assertEqual([u'e'], sorted_list(u'c'))
1113
self.assertEqual(['a', 'a%2525b', 'c', 'c2'], sorted_list('.', t))
1114
self.assertEqual(['e'], sorted_list('c', t))
794
1116
self.assertListRaises(PathError, t.list_dir, 'q')
795
1117
self.assertListRaises(PathError, t.list_dir, 'c/f')
796
1118
self.assertListRaises(PathError, t.list_dir, 'a')
1120
def test_list_dir_result_is_url_escaped(self):
1121
t = self.get_transport()
1122
if not t.listable():
1123
raise TestSkipped("transport not listable")
1125
if not t.is_readonly():
1126
self.build_tree(['a/', 'a/%'], transport=t)
1128
self.build_tree(['a/', 'a/%'])
1130
names = list(t.list_dir('a'))
1131
self.assertEqual(['%25'], names)
1132
self.assertIsInstance(names[0], str)
1134
def test_clone_preserve_info(self):
1135
t1 = self.get_transport()
1136
if not isinstance(t1, ConnectedTransport):
1137
raise TestSkipped("not a connected transport")
1139
t2 = t1.clone('subdir')
1140
self.assertEquals(t1._scheme, t2._scheme)
1141
self.assertEquals(t1._user, t2._user)
1142
self.assertEquals(t1._password, t2._password)
1143
self.assertEquals(t1._host, t2._host)
1144
self.assertEquals(t1._port, t2._port)
1146
def test__reuse_for(self):
1147
t = self.get_transport()
1148
if not isinstance(t, ConnectedTransport):
1149
raise TestSkipped("not a connected transport")
1151
def new_url(scheme=None, user=None, password=None,
1152
host=None, port=None, path=None):
1153
"""Build a new url from t.base changing only parts of it.
1155
Only the parameters different from None will be changed.
1157
if scheme is None: scheme = t._scheme
1158
if user is None: user = t._user
1159
if password is None: password = t._password
1160
if user is None: user = t._user
1161
if host is None: host = t._host
1162
if port is None: port = t._port
1163
if path is None: path = t._path
1164
return t._unsplit_url(scheme, user, password, host, port, path)
1166
if t._scheme == 'ftp':
1170
self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1175
self.assertIsNot(t, t._reuse_for(new_url(user=user)))
1176
# passwords are not taken into account because:
1177
# - it makes no sense to have two different valid passwords for the
1179
# - _password in ConnectedTransport is intended to collect what the
1180
# user specified from the command-line and there are cases where the
1181
# new url can contain no password (if the url was built from an
1182
# existing transport.base for example)
1183
# - password are considered part of the credentials provided at
1184
# connection creation time and as such may not be present in the url
1185
# (they may be typed by the user when prompted for example)
1186
self.assertIs(t, t._reuse_for(new_url(password='from space')))
1187
# We will not connect, we can use a invalid host
1188
self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))
1193
self.assertIsNot(t, t._reuse_for(new_url(port=port)))
1194
# No point in trying to reuse a transport for a local URL
1195
self.assertIs(None, t._reuse_for('/valid_but_not_existing'))
1197
def test_connection_sharing(self):
1198
t = self.get_transport()
1199
if not isinstance(t, ConnectedTransport):
1200
raise TestSkipped("not a connected transport")
1202
c = t.clone('subdir')
1203
# Some transports will create the connection only when needed
1204
t.has('surely_not') # Force connection
1205
self.assertIs(t._get_connection(), c._get_connection())
1207
# Temporary failure, we need to create a new dummy connection
1208
new_connection = object()
1209
t._set_connection(new_connection)
1210
# Check that both transports use the same connection
1211
self.assertIs(new_connection, t._get_connection())
1212
self.assertIs(new_connection, c._get_connection())
1214
def test_reuse_connection_for_various_paths(self):
1215
t = self.get_transport()
1216
if not isinstance(t, ConnectedTransport):
1217
raise TestSkipped("not a connected transport")
1219
t.has('surely_not') # Force connection
1220
self.assertIsNot(None, t._get_connection())
1222
subdir = t._reuse_for(t.base + 'whatever/but/deep/down/the/path')
1223
self.assertIsNot(t, subdir)
1224
self.assertIs(t._get_connection(), subdir._get_connection())
1226
home = subdir._reuse_for(t.base + 'home')
1227
self.assertIs(t._get_connection(), home._get_connection())
1228
self.assertIs(subdir._get_connection(), home._get_connection())
798
1230
def test_clone(self):
799
1231
# TODO: Test that clone moves up and down the filesystem
800
1232
t1 = self.get_transport()
981
1510
if transport.is_readonly():
982
1511
file('a', 'w').write('0123456789')
984
transport.put('a', StringIO('01234567890'))
1513
transport.put_bytes('a', '0123456789')
1515
d = list(transport.readv('a', ((0, 1),)))
1516
self.assertEqual(d[0], (0, '0'))
986
1518
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
987
1519
self.assertEqual(d[0], (0, '0'))
988
1520
self.assertEqual(d[1], (1, '1'))
989
1521
self.assertEqual(d[2], (3, '34'))
990
1522
self.assertEqual(d[3], (9, '9'))
1524
def test_readv_out_of_order(self):
1525
transport = self.get_transport()
1526
if transport.is_readonly():
1527
file('a', 'w').write('0123456789')
1529
transport.put_bytes('a', '01234567890')
1531
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1532
self.assertEqual(d[0], (1, '1'))
1533
self.assertEqual(d[1], (9, '9'))
1534
self.assertEqual(d[2], (0, '0'))
1535
self.assertEqual(d[3], (3, '34'))
1537
def test_readv_with_adjust_for_latency(self):
1538
transport = self.get_transport()
1539
# the adjust for latency flag expands the data region returned
1540
# according to a per-transport heuristic, so testing is a little
1541
# tricky as we need more data than the largest combining that our
1542
# transports do. To accomodate this we generate random data and cross
1543
# reference the returned data with the random data. To avoid doing
1544
# multiple large random byte look ups we do several tests on the same
1546
content = osutils.rand_bytes(200*1024)
1547
content_size = len(content)
1548
if transport.is_readonly():
1549
file('a', 'w').write(content)
1551
transport.put_bytes('a', content)
1552
def check_result_data(result_vector):
1553
for item in result_vector:
1554
data_len = len(item[1])
1555
self.assertEqual(content[item[0]:item[0] + data_len], item[1])
1558
result = list(transport.readv('a', ((0, 30),),
1559
adjust_for_latency=True, upper_limit=content_size))
1560
# we expect 1 result, from 0, to something > 30
1561
self.assertEqual(1, len(result))
1562
self.assertEqual(0, result[0][0])
1563
self.assertTrue(len(result[0][1]) >= 30)
1564
check_result_data(result)
1565
# end of file corner case
1566
result = list(transport.readv('a', ((204700, 100),),
1567
adjust_for_latency=True, upper_limit=content_size))
1568
# we expect 1 result, from 204800- its length, to the end
1569
self.assertEqual(1, len(result))
1570
data_len = len(result[0][1])
1571
self.assertEqual(204800-data_len, result[0][0])
1572
self.assertTrue(data_len >= 100)
1573
check_result_data(result)
1574
# out of order ranges are made in order
1575
result = list(transport.readv('a', ((204700, 100), (0, 50)),
1576
adjust_for_latency=True, upper_limit=content_size))
1577
# we expect 2 results, in order, start and end.
1578
self.assertEqual(2, len(result))
1580
data_len = len(result[0][1])
1581
self.assertEqual(0, result[0][0])
1582
self.assertTrue(data_len >= 30)
1584
data_len = len(result[1][1])
1585
self.assertEqual(204800-data_len, result[1][0])
1586
self.assertTrue(data_len >= 100)
1587
check_result_data(result)
1588
# close ranges get combined (even if out of order)
1589
for request_vector in [((400,50), (800, 234)), ((800, 234), (400,50))]:
1590
result = list(transport.readv('a', request_vector,
1591
adjust_for_latency=True, upper_limit=content_size))
1592
self.assertEqual(1, len(result))
1593
data_len = len(result[0][1])
1594
# minimum length is from 400 to 1034 - 634
1595
self.assertTrue(data_len >= 634)
1596
# must contain the region 400 to 1034
1597
self.assertTrue(result[0][0] <= 400)
1598
self.assertTrue(result[0][0] + data_len >= 1034)
1599
check_result_data(result)
1601
def test_readv_with_adjust_for_latency_with_big_file(self):
1602
transport = self.get_transport()
1603
# test from observed failure case.
1604
if transport.is_readonly():
1605
file('a', 'w').write('a'*1024*1024)
1607
transport.put_bytes('a', 'a'*1024*1024)
1608
broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1609
(225037, 800), (221357, 800), (437077, 800), (947670, 800),
1610
(465373, 800), (947422, 800)]
1611
results = list(transport.readv('a', broken_vector, True, 1024*1024))
1612
found_items = [False]*9
1613
for pos, (start, length) in enumerate(broken_vector):
1614
# check the range is covered by the result
1615
for offset, data in results:
1616
if offset <= start and start + length <= offset + len(data):
1617
found_items[pos] = True
1618
self.assertEqual([True]*9, found_items)
1620
def test_get_with_open_write_stream_sees_all_content(self):
1621
t = self.get_transport()
1624
handle = t.open_write_stream('foo')
1627
self.assertEqual([(0, 'b'), (2, 'd')], list(t.readv('foo', ((0,1), (2,1)))))
1631
def test_get_smart_medium(self):
1632
"""All transports must either give a smart medium, or know they can't.
1634
transport = self.get_transport()
1636
client_medium = transport.get_smart_medium()
1637
self.assertIsInstance(client_medium, medium.SmartClientMedium)
1638
except errors.NoSmartMedium:
1639
# as long as we got it we're fine
1642
def test_readv_short_read(self):
1643
transport = self.get_transport()
1644
if transport.is_readonly():
1645
file('a', 'w').write('0123456789')
1647
transport.put_bytes('a', '01234567890')
1649
# This is intentionally reading off the end of the file
1650
# since we are sure that it cannot get there
1651
self.assertListRaises((errors.ShortReadvError, errors.InvalidRange,
1652
# Can be raised by paramiko
1654
transport.readv, 'a', [(1,1), (8,10)])
1656
# This is trying to seek past the end of the file, it should
1657
# also raise a special error
1658
self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1659
transport.readv, 'a', [(12,2)])
added
removed
bzrlib/tests/test_transport_implementations.py
