751
760
def log(self, *args):
755
"""Return as a string the log for this test"""
756
if self._log_file_name:
757
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:
759
771
return self._log_contents
760
# 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"
762
785
def capture(self, cmd, retcode=0):
763
786
"""Shortcut that splits cmd into words, runs, and returns stdout"""
764
787
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
766
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,
767
791
"""Invoke bzr and return (stdout, stderr).
769
793
Useful for code that wants to check the contents of the
891
926
variables. A value of None will unset the env variable.
892
927
The values must be strings. The change will only occur in the
893
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.
895
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:
896
979
def cleanup_environment():
897
980
for env_var, value in env_changes.iteritems():
899
if env_var in os.environ:
900
del os.environ[env_var]
902
os.environ[env_var] = value
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."""
904
1021
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
906
process = Popen([sys.executable, bzr_path]+args,
907
stdout=PIPE, stderr=PIPE,
908
preexec_fn=cleanup_environment)
909
out = process.stdout.read()
910
err = process.stderr.read()
911
retcode = process.wait()
912
supplied_retcode = kwargs.get('retcode', 0)
913
if supplied_retcode is not None:
914
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))
915
1053
return [out, err]
917
1055
def check_inventory_shape(self, inv, shape):
987
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
990
class TestCaseInTempDir(TestCase):
1333
class TestCaseInTempDir(TestCaseWithMemoryTransport):
991
1334
"""Derived class that runs a test within a temporary directory.
993
1336
This is useful for tests that need to create a branch, etc.
1137
1449
readwrite one must both define get_url() as resolving to os.getcwd().
1140
def __init__(self, methodName='testMethod'):
1141
super(TestCaseWithTransport, self).__init__(methodName)
1142
self.__readonly_server = None
1143
self.__server = None
1144
self.transport_server = default_transport
1145
self.transport_readonly_server = None
1147
def get_readonly_url(self, relpath=None):
1148
"""Get a URL for the readonly transport.
1150
This will either be backed by '.' or a decorator to the transport
1151
used by self.get_url()
1152
relpath provides for clients to get a path relative to the base url.
1153
These should only be downwards relative, not upwards.
1155
base = self.get_readonly_server().get_url()
1156
if relpath is not None:
1157
if not base.endswith('/'):
1159
base = base + relpath
1162
def get_readonly_server(self):
1163
"""Get the server instance for the readonly transport
1165
This is useful for some tests with specific servers to do diagnostics.
1167
if self.__readonly_server is None:
1168
if self.transport_readonly_server is None:
1169
# readonly decorator requested
1170
# bring up the server
1172
self.__readonly_server = ReadonlyServer()
1173
self.__readonly_server.setUp(self.__server)
1175
self.__readonly_server = self.transport_readonly_server()
1176
self.__readonly_server.setUp()
1177
self.addCleanup(self.__readonly_server.tearDown)
1178
return self.__readonly_server
1180
1452
def get_server(self):
1181
"""Get the read/write server instance.
1453
"""See TestCaseWithMemoryTransport.
1183
1455
This is useful for some tests with specific servers that need
1189
1461
self.addCleanup(self.__server.tearDown)
1190
1462
return self.__server
1192
def get_url(self, relpath=None):
1193
"""Get a URL for the readwrite transport.
1195
This will either be backed by '.' or to an equivalent non-file based
1197
relpath provides for clients to get a path relative to the base url.
1198
These should only be downwards relative, not upwards.
1200
base = self.get_server().get_url()
1201
if relpath is not None and relpath != '.':
1202
if not base.endswith('/'):
1204
base = base + urlutils.escape(relpath)
1207
def get_transport(self):
1208
"""Return a writeable transport for the test scratch space"""
1209
t = get_transport(self.get_url())
1210
self.assertFalse(t.is_readonly())
1213
def get_readonly_transport(self):
1214
"""Return a readonly transport for the test scratch space
1216
This can be used to test that operations which should only need
1217
readonly access in fact do not try to write.
1219
t = get_transport(self.get_readonly_url())
1220
self.assertTrue(t.is_readonly())
1223
def make_branch(self, relpath, format=None):
1224
"""Create a branch on the transport at relpath."""
1225
repo = self.make_repository(relpath, format=format)
1226
return repo.bzrdir.create_branch()
1228
def make_bzrdir(self, relpath, format=None):
1230
url = self.get_url(relpath)
1231
mutter('relpath %r => url %r', relpath, url)
1232
segments = url.split('/')
1233
if segments and segments[-1] not in ('', '.'):
1234
parent = '/'.join(segments[:-1])
1235
t = get_transport(parent)
1237
t.mkdir(segments[-1])
1238
except errors.FileExists:
1241
format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
1242
# FIXME: make this use a single transport someday. RBC 20060418
1243
return format.initialize_on_transport(get_transport(relpath))
1244
except errors.UninitializableFormat:
1245
raise TestSkipped("Format %s is not initializable." % format)
1247
def make_repository(self, relpath, shared=False, format=None):
1248
"""Create a repository on our default transport at relpath."""
1249
made_control = self.make_bzrdir(relpath, format=format)
1250
return made_control.create_repository(shared=shared)
1252
1464
def make_branch_and_tree(self, relpath, format=None):
1253
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.
1257
1479
# TODO: always use the local disk path for the working tree,
1258
1480
# this obviously requires a format that supports branch references
added
removed
bzrlib/tests/__init__.py
