24
24
from cStringIO import StringIO
25
from StringIO import StringIO as pyStringIO
28
30
from bzrlib import (
33
from bzrlib.errors import (DirectoryNotEmpty, NoSuchFile, FileExists,
34
LockError, NoSmartServer, PathError,
35
TransportNotPossible, ConnectionError,
35
from bzrlib.errors import (ConnectionError,
37
46
from bzrlib.osutils import getcwd
47
from bzrlib.smart import medium
38
48
from bzrlib.symbol_versioning import zero_eleven
39
from bzrlib.tests import TestCaseInTempDir, TestSkipped
49
from bzrlib.tests import TestCaseInTempDir, TestScenarioApplier, TestSkipped
40
50
from bzrlib.tests.test_transport import TestTransportImplementation
41
from bzrlib.transport import memory, smart
51
from bzrlib.transport import memory, remote, _get_transport_modules
42
52
import bzrlib.transport
46
"""Append the given text (file-like object) to the supplied filename."""
55
class TransportTestProviderAdapter(TestScenarioApplier):
56
"""A tool to generate a suite testing all transports for a single test.
58
This is done by copying the test once for each transport and injecting
59
the transport_class and transport_server classes into each copy. Each copy
60
is also given a new id() to make it easy to identify.
64
self.scenarios = self._test_permutations()
66
def get_transport_test_permutations(self, module):
67
"""Get the permutations module wants to have tested."""
68
if getattr(module, 'get_test_permutations', None) is None:
69
raise AssertionError("transport module %s doesn't provide get_test_permutations()"
71
##warning("transport module %s doesn't provide get_test_permutations()"
74
return module.get_test_permutations()
76
def _test_permutations(self):
77
"""Return a list of the klass, server_factory pairs to test."""
79
for module in _get_transport_modules():
81
permutations = self.get_transport_test_permutations(
82
reduce(getattr, (module).split('.')[1:], __import__(module)))
83
for (klass, server_factory) in permutations:
84
scenario = (server_factory.__name__,
85
{"transport_class":klass,
86
"transport_server":server_factory})
87
result.append(scenario)
88
except errors.DependencyNotPresent, e:
89
# Continue even if a dependency prevents us
90
# from running this test
54
96
class TransportTests(TestTransportImplementation):
99
super(TransportTests, self).setUp()
100
self._captureVar('BZR_NO_SMART_VFS', None)
56
102
def check_transport_contents(self, content, transport, relpath):
57
103
"""Check that transport.get(relpath).read() == content."""
58
104
self.assertEqualDiff(content, transport.get(relpath).read())
60
def assertListRaises(self, excClass, func, *args, **kwargs):
61
"""Fail unless excClass is raised when the iterator from func is used.
63
Many transport functions can return generators this makes sure
64
to wrap them in a list() call to make sure the whole generator
65
is run, and that the proper exception is raised.
68
list(func(*args, **kwargs))
72
if getattr(excClass,'__name__', None) is not None:
73
excName = excClass.__name__
75
excName = str(excClass)
76
raise self.failureException, "%s not raised" % excName
106
def test_ensure_base_missing(self):
107
""".ensure_base() should create the directory if it doesn't exist"""
108
t = self.get_transport()
110
if t_a.is_readonly():
111
self.assertRaises(TransportNotPossible,
114
self.assertTrue(t_a.ensure_base())
115
self.assertTrue(t.has('a'))
117
def test_ensure_base_exists(self):
118
""".ensure_base() should just be happy if it already exists"""
119
t = self.get_transport()
125
# ensure_base returns False if it didn't create the base
126
self.assertFalse(t_a.ensure_base())
128
def test_ensure_base_missing_parent(self):
129
""".ensure_base() will fail if the parent dir doesn't exist"""
130
t = self.get_transport()
136
self.assertRaises(NoSuchFile, t_b.ensure_base)
78
138
def test_has(self):
79
139
t = self.get_transport()
412
471
dir_mode=0777, create_parent_dir=True)
413
472
self.assertTransportMode(t, 'dir777', 0777)
474
def test_put_bytes_unicode(self):
475
# Expect put_bytes to raise AssertionError or UnicodeEncodeError if
476
# given unicode "bytes". UnicodeEncodeError doesn't really make sense
477
# (we don't want to encode unicode here at all, callers should be
478
# strictly passing bytes to put_bytes), but we allow it for backwards
479
# compatibility. At some point we should use a specific exception.
480
# See https://bugs.launchpad.net/bzr/+bug/106898.
481
t = self.get_transport()
484
unicode_string = u'\u1234'
486
(AssertionError, UnicodeEncodeError),
487
t.put_bytes, 'foo', unicode_string)
489
def test_put_file_unicode(self):
490
# Like put_bytes, except with a StringIO.StringIO of a unicode string.
491
# This situation can happen (and has) if code is careless about the type
492
# of "string" they initialise/write to a StringIO with. We cannot use
493
# cStringIO, because it never returns unicode from read.
494
# Like put_bytes, UnicodeEncodeError isn't quite the right exception to
495
# raise, but we raise it for hysterical raisins.
496
t = self.get_transport()
499
unicode_file = pyStringIO(u'\u1234')
500
self.assertRaises(UnicodeEncodeError, t.put_file, 'foo', unicode_file)
415
502
def test_put_multi(self):
416
503
t = self.get_transport()
982
# SftpServer creates control files in the working directory
983
# so lets move down a directory to avoid those.
984
if not t.is_readonly():
990
1089
self.assertEqual([], sorted_list('.'))
991
1090
# c2 is precisely one letter longer than c here to test that
992
1091
# suffixing is not confused.
993
1092
# a%25b checks that quoting is done consistently across transports
994
1093
tree_names = ['a', 'a%25b', 'b', 'c/', 'c/d', 'c/e', 'c2/']
995
1095
if not t.is_readonly():
996
1096
self.build_tree(tree_names, transport=t)
998
self.build_tree(['wd/' + name for name in tree_names])
1098
self.build_tree(tree_names)
1000
1100
self.assertEqual(
1001
1101
['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('.'))
1068
1168
# directory of this transport
1069
1169
root_transport = orig_transport
1070
1170
new_transport = root_transport.clone("..")
1071
# as we are walking up directories, the path must be must be
1171
# as we are walking up directories, the path must be
1072
1172
# growing less, except at the top
1073
1173
self.assertTrue(len(new_transport.base) < len(root_transport.base)
1074
1174
or new_transport.base == root_transport.base)
1075
1175
while new_transport.base != root_transport.base:
1076
1176
root_transport = new_transport
1077
1177
new_transport = root_transport.clone("..")
1078
# as we are walking up directories, the path must be must be
1178
# as we are walking up directories, the path must be
1079
1179
# growing less, except at the top
1080
1180
self.assertTrue(len(new_transport.base) < len(root_transport.base)
1081
1181
or new_transport.base == root_transport.base)
1329
1436
self.assertEqual(d[2], (0, '0'))
1330
1437
self.assertEqual(d[3], (3, '34'))
1332
def test_get_smart_client(self):
1333
"""All transports must either give a smart client, or know they can't.
1335
For some transports such as http this might depend on probing to see
1336
what's actually present on the other end. (But we can adjust for that
1439
def test_get_smart_medium(self):
1440
"""All transports must either give a smart medium, or know they can't.
1339
1442
transport = self.get_transport()
1341
client = transport.get_smart_client()
1342
# XXX: should be a more general class
1343
self.assertIsInstance(client, smart.SmartStreamClient)
1344
except NoSmartServer:
1444
client_medium = transport.get_smart_medium()
1445
self.assertIsInstance(client_medium, medium.SmartClientMedium)
1446
except errors.NoSmartMedium:
1345
1447
# as long as we got it we're fine
1355
1457
# This is intentionally reading off the end of the file
1356
1458
# since we are sure that it cannot get there
1357
self.assertListRaises((errors.ShortReadvError, AssertionError),
1459
self.assertListRaises((errors.ShortReadvError, errors.InvalidRange,
1460
# Can be raised by paramiko
1358
1462
transport.readv, 'a', [(1,1), (8,10)])
1360
1464
# This is trying to seek past the end of the file, it should
1361
1465
# also raise a special error
1362
self.assertListRaises(errors.ShortReadvError,
1466
self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1363
1467
transport.readv, 'a', [(12,2)])
added
removed
bzrlib/tests/test_transport_implementations.py
