552
593
self.fail("%r is an instance of %s rather than %s" % (
553
594
obj, obj.__class__, kls))
596
def _capture_warnings(self, a_callable, *args, **kwargs):
597
"""A helper for callDeprecated and applyDeprecated.
599
:param a_callable: A callable to call.
600
:param args: The positional arguments for the callable
601
:param kwargs: The keyword arguments for the callable
602
:return: A tuple (warnings, result). result is the result of calling
603
a_callable(*args, **kwargs).
606
def capture_warnings(msg, cls, stacklevel=None):
607
# we've hooked into a deprecation specific callpath,
608
# only deprecations should getting sent via it.
609
self.assertEqual(cls, DeprecationWarning)
610
local_warnings.append(msg)
611
original_warning_method = symbol_versioning.warn
612
symbol_versioning.set_warning_method(capture_warnings)
614
result = a_callable(*args, **kwargs)
616
symbol_versioning.set_warning_method(original_warning_method)
617
return (local_warnings, result)
619
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
620
"""Call a deprecated callable without warning the user.
622
:param deprecation_format: The deprecation format that the callable
623
should have been deprecated with. This is the same type as the
624
parameter to deprecated_method/deprecated_function. If the
625
callable is not deprecated with this format, an assertion error
627
:param a_callable: A callable to call. This may be a bound method or
628
a regular function. It will be called with *args and **kwargs.
629
:param args: The positional arguments for the callable
630
:param kwargs: The keyword arguments for the callable
631
:return: The result of a_callable(*args, **kwargs)
633
call_warnings, result = self._capture_warnings(a_callable,
635
expected_first_warning = symbol_versioning.deprecation_string(
636
a_callable, deprecation_format)
637
if len(call_warnings) == 0:
638
self.fail("No assertion generated by call to %s" %
640
self.assertEqual(expected_first_warning, call_warnings[0])
643
def callDeprecated(self, expected, callable, *args, **kwargs):
644
"""Assert that a callable is deprecated in a particular way.
646
This is a very precise test for unusual requirements. The
647
applyDeprecated helper function is probably more suited for most tests
648
as it allows you to simply specify the deprecation format being used
649
and will ensure that that is issued for the function being called.
651
:param expected: a list of the deprecation warnings expected, in order
652
:param callable: The callable to call
653
:param args: The positional arguments for the callable
654
:param kwargs: The keyword arguments for the callable
656
call_warnings, result = self._capture_warnings(callable,
658
self.assertEqual(expected, call_warnings)
555
661
def _startLogFile(self):
556
662
"""Send bzr and test log messages to a temporary file.
558
664
The file is removed as the test is torn down.
560
666
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
561
encoder, decoder, stream_reader, stream_writer = codecs.lookup('UTF-8')
562
self._log_file = stream_writer(os.fdopen(fileno, 'w+'))
667
self._log_file = os.fdopen(fileno, 'w+')
563
668
self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
564
669
self._log_file_name = name
565
670
self.addCleanup(self._finishLogFile)
663
760
def log(self, *args):
667
"""Return as a string the log for this test"""
668
if self._log_file_name:
669
return open(self._log_file_name).read()
763
def _get_log(self, keep_log_file=False):
764
"""Return as a string the log for this test. If the file is still
765
on disk and keep_log_file=False, delete the log file and store the
766
content in self._log_contents."""
767
# flush the log file, to get all content
769
bzrlib.trace._trace_file.flush()
770
if self._log_contents:
671
771
return self._log_contents
672
# TODO: Delete the log after it's been read in
772
if self._log_file_name is not None:
773
logfile = open(self._log_file_name)
775
log_contents = logfile.read()
778
if not keep_log_file:
779
self._log_contents = log_contents
780
os.remove(self._log_file_name)
783
return "DELETED log file to reduce memory footprint"
674
785
def capture(self, cmd, retcode=0):
675
786
"""Shortcut that splits cmd into words, runs, and returns stdout"""
676
787
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
678
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None):
789
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None,
679
791
"""Invoke bzr and return (stdout, stderr).
681
793
Useful for code that wants to check the contents of the
798
921
profiled or debugged so easily.
800
923
:param retcode: The status code that is expected. Defaults to 0. If
801
None is supplied, the status code is not checked.
924
None is supplied, the status code is not checked.
925
:param env_changes: A dictionary which lists changes to environment
926
variables. A value of None will unset the env variable.
927
The values must be strings. The change will only occur in the
928
child, so you don't need to fix the environment after running.
929
:param universal_newlines: Convert CRLF => LF
930
:param allow_plugins: By default the subprocess is run with
931
--no-plugins to ensure test reproducibility. Also, it is possible
932
for system-wide plugins to create unexpected output on stderr,
933
which can cause unnecessary test failures.
935
env_changes = kwargs.get('env_changes', {})
936
working_dir = kwargs.get('working_dir', None)
937
allow_plugins = kwargs.get('allow_plugins', False)
938
process = self.start_bzr_subprocess(args, env_changes=env_changes,
939
working_dir=working_dir,
940
allow_plugins=allow_plugins)
941
# We distinguish between retcode=None and retcode not passed.
942
supplied_retcode = kwargs.get('retcode', 0)
943
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
944
universal_newlines=kwargs.get('universal_newlines', False),
947
def start_bzr_subprocess(self, process_args, env_changes=None,
948
skip_if_plan_to_signal=False,
950
allow_plugins=False):
951
"""Start bzr in a subprocess for testing.
953
This starts a new Python interpreter and runs bzr in there.
954
This should only be used for tests that have a justifiable need for
955
this isolation: e.g. they are testing startup time, or signal
956
handling, or early startup code, etc. Subprocess code can't be
957
profiled or debugged so easily.
959
:param process_args: a list of arguments to pass to the bzr executable,
960
for example `['--version']`.
961
:param env_changes: A dictionary which lists changes to environment
962
variables. A value of None will unset the env variable.
963
The values must be strings. The change will only occur in the
964
child, so you don't need to fix the environment after running.
965
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
967
:param allow_plugins: If False (default) pass --no-plugins to bzr.
969
:returns: Popen object for the started process.
971
if skip_if_plan_to_signal:
972
if not getattr(os, 'kill', None):
973
raise TestSkipped("os.kill not available.")
975
if env_changes is None:
979
def cleanup_environment():
980
for env_var, value in env_changes.iteritems():
981
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
983
def restore_environment():
984
for env_var, value in old_env.iteritems():
985
osutils.set_or_unset_env(env_var, value)
987
bzr_path = self.get_bzr_path()
990
if working_dir is not None:
991
cwd = osutils.getcwd()
992
os.chdir(working_dir)
995
# win32 subprocess doesn't support preexec_fn
996
# so we will avoid using it on all platforms, just to
997
# make sure the code path is used, and we don't break on win32
998
cleanup_environment()
999
command = [sys.executable, bzr_path]
1000
if not allow_plugins:
1001
command.append('--no-plugins')
1002
command.extend(process_args)
1003
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1005
restore_environment()
1011
def _popen(self, *args, **kwargs):
1012
"""Place a call to Popen.
1014
Allows tests to override this method to intercept the calls made to
1015
Popen for introspection.
1017
return Popen(*args, **kwargs)
1019
def get_bzr_path(self):
1020
"""Return the path of the 'bzr' executable for this test suite."""
803
1021
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
805
process = Popen([sys.executable, bzr_path]+args, stdout=PIPE,
807
out = process.stdout.read()
808
err = process.stderr.read()
809
retcode = process.wait()
810
supplied_retcode = kwargs.get('retcode', 0)
811
if supplied_retcode is not None:
812
assert supplied_retcode == retcode
1022
if not os.path.isfile(bzr_path):
1023
# We are probably installed. Assume sys.argv is the right file
1024
bzr_path = sys.argv[0]
1027
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1028
universal_newlines=False, process_args=None):
1029
"""Finish the execution of process.
1031
:param process: the Popen object returned from start_bzr_subprocess.
1032
:param retcode: The status code that is expected. Defaults to 0. If
1033
None is supplied, the status code is not checked.
1034
:param send_signal: an optional signal to send to the process.
1035
:param universal_newlines: Convert CRLF => LF
1036
:returns: (stdout, stderr)
1038
if send_signal is not None:
1039
os.kill(process.pid, send_signal)
1040
out, err = process.communicate()
1042
if universal_newlines:
1043
out = out.replace('\r\n', '\n')
1044
err = err.replace('\r\n', '\n')
1046
if retcode is not None and retcode != process.returncode:
1047
if process_args is None:
1048
process_args = "(unknown args)"
1049
mutter('Output of bzr %s:\n%s', process_args, out)
1050
mutter('Error for bzr %s:\n%s', process_args, err)
1051
self.fail('Command bzr %s failed with retcode %s != %s'
1052
% (process_args, retcode, process.returncode))
813
1053
return [out, err]
815
1055
def check_inventory_shape(self, inv, shape):
875
1116
base_rev = common_ancestor(branch_from.last_revision(),
876
1117
wt_to.branch.last_revision(),
877
1118
wt_to.branch.repository)
878
merge_inner(wt_to.branch, branch_from.basis_tree(),
1119
merge_inner(wt_to.branch, branch_from.basis_tree(),
879
1120
wt_to.branch.repository.revision_tree(base_rev),
880
1121
this_tree=wt_to)
881
wt_to.add_pending_merge(branch_from.last_revision())
1122
wt_to.add_parent_tree_id(branch_from.last_revision())
884
1125
BzrTestBase = TestCase
1128
class TestCaseWithMemoryTransport(TestCase):
1129
"""Common test class for tests that do not need disk resources.
1131
Tests that need disk resources should derive from TestCaseWithTransport.
1133
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1135
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1136
a directory which does not exist. This serves to help ensure test isolation
1137
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1138
must exist. However, TestCaseWithMemoryTransport does not offer local
1139
file defaults for the transport in tests, nor does it obey the command line
1140
override, so tests that accidentally write to the common directory should
1148
def __init__(self, methodName='runTest'):
1149
# allow test parameterisation after test construction and before test
1150
# execution. Variables that the parameteriser sets need to be
1151
# ones that are not set by setUp, or setUp will trash them.
1152
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1153
self.transport_server = default_transport
1154
self.transport_readonly_server = None
1156
def failUnlessExists(self, path):
1157
"""Fail unless path, which may be abs or relative, exists."""
1158
self.failUnless(osutils.lexists(path))
1160
def failIfExists(self, path):
1161
"""Fail if path, which may be abs or relative, exists."""
1162
self.failIf(osutils.lexists(path))
1164
def get_transport(self):
1165
"""Return a writeable transport for the test scratch space"""
1166
t = get_transport(self.get_url())
1167
self.assertFalse(t.is_readonly())
1170
def get_readonly_transport(self):
1171
"""Return a readonly transport for the test scratch space
1173
This can be used to test that operations which should only need
1174
readonly access in fact do not try to write.
1176
t = get_transport(self.get_readonly_url())
1177
self.assertTrue(t.is_readonly())
1180
def get_readonly_server(self):
1181
"""Get the server instance for the readonly transport
1183
This is useful for some tests with specific servers to do diagnostics.
1185
if self.__readonly_server is None:
1186
if self.transport_readonly_server is None:
1187
# readonly decorator requested
1188
# bring up the server
1190
self.__readonly_server = ReadonlyServer()
1191
self.__readonly_server.setUp(self.__server)
1193
self.__readonly_server = self.transport_readonly_server()
1194
self.__readonly_server.setUp()
1195
self.addCleanup(self.__readonly_server.tearDown)
1196
return self.__readonly_server
1198
def get_readonly_url(self, relpath=None):
1199
"""Get a URL for the readonly transport.
1201
This will either be backed by '.' or a decorator to the transport
1202
used by self.get_url()
1203
relpath provides for clients to get a path relative to the base url.
1204
These should only be downwards relative, not upwards.
1206
base = self.get_readonly_server().get_url()
1207
if relpath is not None:
1208
if not base.endswith('/'):
1210
base = base + relpath
1213
def get_server(self):
1214
"""Get the read/write server instance.
1216
This is useful for some tests with specific servers that need
1219
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1220
is no means to override it.
1222
if self.__server is None:
1223
self.__server = MemoryServer()
1224
self.__server.setUp()
1225
self.addCleanup(self.__server.tearDown)
1226
return self.__server
1228
def get_url(self, relpath=None):
1229
"""Get a URL (or maybe a path) for the readwrite transport.
1231
This will either be backed by '.' or to an equivalent non-file based
1233
relpath provides for clients to get a path relative to the base url.
1234
These should only be downwards relative, not upwards.
1236
base = self.get_server().get_url()
1237
if relpath is not None and relpath != '.':
1238
if not base.endswith('/'):
1240
# XXX: Really base should be a url; we did after all call
1241
# get_url()! But sometimes it's just a path (from
1242
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1243
# to a non-escaped local path.
1244
if base.startswith('./') or base.startswith('/'):
1247
base += urlutils.escape(relpath)
1250
def _make_test_root(self):
1251
if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1255
root = u'test%04d.tmp' % i
1259
if e.errno == errno.EEXIST:
1264
# successfully created
1265
TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1267
# make a fake bzr directory there to prevent any tests propagating
1268
# up onto the source directory's real branch
1269
bzrdir.BzrDir.create_standalone_workingtree(
1270
TestCaseWithMemoryTransport.TEST_ROOT)
1272
def makeAndChdirToTestDir(self):
1273
"""Create a temporary directories for this one test.
1275
This must set self.test_home_dir and self.test_dir and chdir to
1278
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1280
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1281
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1282
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1284
def make_branch(self, relpath, format=None):
1285
"""Create a branch on the transport at relpath."""
1286
repo = self.make_repository(relpath, format=format)
1287
return repo.bzrdir.create_branch()
1289
def make_bzrdir(self, relpath, format=None):
1291
# might be a relative or absolute path
1292
maybe_a_url = self.get_url(relpath)
1293
segments = maybe_a_url.rsplit('/', 1)
1294
t = get_transport(maybe_a_url)
1295
if len(segments) > 1 and segments[-1] not in ('', '.'):
1298
except errors.FileExists:
1301
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1302
return format.initialize_on_transport(t)
1303
except errors.UninitializableFormat:
1304
raise TestSkipped("Format %s is not initializable." % format)
1306
def make_repository(self, relpath, shared=False, format=None):
1307
"""Create a repository on our default transport at relpath."""
1308
made_control = self.make_bzrdir(relpath, format=format)
1309
return made_control.create_repository(shared=shared)
1311
def make_branch_and_memory_tree(self, relpath, format=None):
1312
"""Create a branch on the default transport and a MemoryTree for it."""
1313
b = self.make_branch(relpath, format=format)
1314
return memorytree.MemoryTree.create_on_branch(b)
1316
def overrideEnvironmentForTesting(self):
1317
os.environ['HOME'] = self.test_home_dir
1318
os.environ['APPDATA'] = self.test_home_dir
1321
super(TestCaseWithMemoryTransport, self).setUp()
1322
self._make_test_root()
1323
_currentdir = os.getcwdu()
1324
def _leaveDirectory():
1325
os.chdir(_currentdir)
1326
self.addCleanup(_leaveDirectory)
1327
self.makeAndChdirToTestDir()
1328
self.overrideEnvironmentForTesting()
1329
self.__readonly_server = None
1330
self.__server = None
887
class TestCaseInTempDir(TestCase):
1333
class TestCaseInTempDir(TestCaseWithMemoryTransport):
888
1334
"""Derived class that runs a test within a temporary directory.
890
1336
This is useful for tests that need to create a branch, etc.
1033
1449
readwrite one must both define get_url() as resolving to os.getcwd().
1036
def __init__(self, methodName='testMethod'):
1037
super(TestCaseWithTransport, self).__init__(methodName)
1038
self.__readonly_server = None
1039
self.__server = None
1040
self.transport_server = default_transport
1041
self.transport_readonly_server = None
1043
def get_readonly_url(self, relpath=None):
1044
"""Get a URL for the readonly transport.
1046
This will either be backed by '.' or a decorator to the transport
1047
used by self.get_url()
1048
relpath provides for clients to get a path relative to the base url.
1049
These should only be downwards relative, not upwards.
1051
base = self.get_readonly_server().get_url()
1052
if relpath is not None:
1053
if not base.endswith('/'):
1055
base = base + relpath
1058
def get_readonly_server(self):
1059
"""Get the server instance for the readonly transport
1061
This is useful for some tests with specific servers to do diagnostics.
1063
if self.__readonly_server is None:
1064
if self.transport_readonly_server is None:
1065
# readonly decorator requested
1066
# bring up the server
1068
self.__readonly_server = ReadonlyServer()
1069
self.__readonly_server.setUp(self.__server)
1071
self.__readonly_server = self.transport_readonly_server()
1072
self.__readonly_server.setUp()
1073
self.addCleanup(self.__readonly_server.tearDown)
1074
return self.__readonly_server
1076
1452
def get_server(self):
1077
"""Get the read/write server instance.
1453
"""See TestCaseWithMemoryTransport.
1079
1455
This is useful for some tests with specific servers that need
1085
1461
self.addCleanup(self.__server.tearDown)
1086
1462
return self.__server
1088
def get_url(self, relpath=None):
1089
"""Get a URL for the readwrite transport.
1091
This will either be backed by '.' or to an equivalent non-file based
1093
relpath provides for clients to get a path relative to the base url.
1094
These should only be downwards relative, not upwards.
1096
base = self.get_server().get_url()
1097
if relpath is not None and relpath != '.':
1098
if not base.endswith('/'):
1100
base = base + urlutils.escape(relpath)
1103
def get_transport(self):
1104
"""Return a writeable transport for the test scratch space"""
1105
t = get_transport(self.get_url())
1106
self.assertFalse(t.is_readonly())
1109
def get_readonly_transport(self):
1110
"""Return a readonly transport for the test scratch space
1112
This can be used to test that operations which should only need
1113
readonly access in fact do not try to write.
1115
t = get_transport(self.get_readonly_url())
1116
self.assertTrue(t.is_readonly())
1119
def make_branch(self, relpath, format=None):
1120
"""Create a branch on the transport at relpath."""
1121
repo = self.make_repository(relpath, format=format)
1122
return repo.bzrdir.create_branch()
1124
def make_bzrdir(self, relpath, format=None):
1126
url = self.get_url(relpath)
1127
mutter('relpath %r => url %r', relpath, url)
1128
segments = url.split('/')
1129
if segments and segments[-1] not in ('', '.'):
1130
parent = '/'.join(segments[:-1])
1131
t = get_transport(parent)
1133
t.mkdir(segments[-1])
1134
except errors.FileExists:
1137
format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
1138
# FIXME: make this use a single transport someday. RBC 20060418
1139
return format.initialize_on_transport(get_transport(relpath))
1140
except errors.UninitializableFormat:
1141
raise TestSkipped("Format %s is not initializable." % format)
1143
def make_repository(self, relpath, shared=False, format=None):
1144
"""Create a repository on our default transport at relpath."""
1145
made_control = self.make_bzrdir(relpath, format=format)
1146
return made_control.create_repository(shared=shared)
1148
1464
def make_branch_and_tree(self, relpath, format=None):
1149
1465
"""Create a branch on the transport and a tree locally.
1467
If the transport is not a LocalTransport, the Tree can't be created on
1468
the transport. In that case the working tree is created in the local
1469
directory, and the returned tree's branch and repository will also be
1472
This will fail if the original default transport for this test
1473
case wasn't backed by the working directory, as the branch won't
1474
be on disk for us to open it.
1476
:param format: The BzrDirFormat.
1477
:returns: the WorkingTree.
1153
1479
# TODO: always use the local disk path for the working tree,
1154
1480
# this obviously requires a format that supports branch references
added
removed
bzrlib/tests/__init__.py
