740
760
def log(self, *args):
744
"""Return as a string the log for this test"""
745
if self._log_file_name:
746
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:
748
771
return self._log_contents
749
# 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"
751
785
def capture(self, cmd, retcode=0):
752
786
"""Shortcut that splits cmd into words, runs, and returns stdout"""
753
787
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
755
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,
756
791
"""Invoke bzr and return (stdout, stderr).
758
793
Useful for code that wants to check the contents of the
883
929
:param universal_newlines: Convert CRLF => LF
885
931
env_changes = kwargs.get('env_changes', {})
932
working_dir = kwargs.get('working_dir', None)
933
process = self.start_bzr_subprocess(args, env_changes=env_changes,
934
working_dir=working_dir)
935
# We distinguish between retcode=None and retcode not passed.
936
supplied_retcode = kwargs.get('retcode', 0)
937
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
938
universal_newlines=kwargs.get('universal_newlines', False),
941
def start_bzr_subprocess(self, process_args, env_changes=None,
942
skip_if_plan_to_signal=False,
944
"""Start bzr in a subprocess for testing.
946
This starts a new Python interpreter and runs bzr in there.
947
This should only be used for tests that have a justifiable need for
948
this isolation: e.g. they are testing startup time, or signal
949
handling, or early startup code, etc. Subprocess code can't be
950
profiled or debugged so easily.
952
:param process_args: a list of arguments to pass to the bzr executable,
953
for example `['--version']`.
954
:param env_changes: A dictionary which lists changes to environment
955
variables. A value of None will unset the env variable.
956
The values must be strings. The change will only occur in the
957
child, so you don't need to fix the environment after running.
958
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
961
:returns: Popen object for the started process.
963
if skip_if_plan_to_signal:
964
if not getattr(os, 'kill', None):
965
raise TestSkipped("os.kill not available.")
967
if env_changes is None:
889
971
def cleanup_environment():
894
976
for env_var, value in old_env.iteritems():
895
977
osutils.set_or_unset_env(env_var, value)
897
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
979
bzr_path = self.get_bzr_path()
982
if working_dir is not None:
983
cwd = osutils.getcwd()
984
os.chdir(working_dir)
901
987
# win32 subprocess doesn't support preexec_fn
902
988
# so we will avoid using it on all platforms, just to
903
989
# make sure the code path is used, and we don't break on win32
904
990
cleanup_environment()
905
process = Popen([sys.executable, bzr_path]+args,
906
stdout=PIPE, stderr=PIPE)
991
process = Popen([sys.executable, bzr_path] + list(process_args),
992
stdin=PIPE, stdout=PIPE, stderr=PIPE)
908
994
restore_environment()
910
out = process.stdout.read()
911
err = process.stderr.read()
913
if kwargs.get('universal_newlines', False):
1000
def get_bzr_path(self):
1001
"""Return the path of the 'bzr' executable for this test suite."""
1002
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1003
if not os.path.isfile(bzr_path):
1004
# We are probably installed. Assume sys.argv is the right file
1005
bzr_path = sys.argv[0]
1008
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1009
universal_newlines=False, process_args=None):
1010
"""Finish the execution of process.
1012
:param process: the Popen object returned from start_bzr_subprocess.
1013
:param retcode: The status code that is expected. Defaults to 0. If
1014
None is supplied, the status code is not checked.
1015
:param send_signal: an optional signal to send to the process.
1016
:param universal_newlines: Convert CRLF => LF
1017
:returns: (stdout, stderr)
1019
if send_signal is not None:
1020
os.kill(process.pid, send_signal)
1021
out, err = process.communicate()
1023
if universal_newlines:
914
1024
out = out.replace('\r\n', '\n')
915
1025
err = err.replace('\r\n', '\n')
917
retcode = process.wait()
918
supplied_retcode = kwargs.get('retcode', 0)
919
if supplied_retcode is not None:
920
assert supplied_retcode == retcode
1027
if retcode is not None and retcode != process.returncode:
1028
if process_args is None:
1029
process_args = "(unknown args)"
1030
mutter('Output of bzr %s:\n%s', process_args, out)
1031
mutter('Error for bzr %s:\n%s', process_args, err)
1032
self.fail('Command bzr %s failed with retcode %s != %s'
1033
% (process_args, retcode, process.returncode))
921
1034
return [out, err]
923
1036
def check_inventory_shape(self, inv, shape):
993
1106
BzrTestBase = TestCase
1109
class TestCaseWithMemoryTransport(TestCase):
1110
"""Common test class for tests that do not need disk resources.
1112
Tests that need disk resources should derive from TestCaseWithTransport.
1114
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1116
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1117
a directory which does not exist. This serves to help ensure test isolation
1118
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1119
must exist. However, TestCaseWithMemoryTransport does not offer local
1120
file defaults for the transport in tests, nor does it obey the command line
1121
override, so tests that accidentally write to the common directory should
1129
def __init__(self, methodName='runTest'):
1130
# allow test parameterisation after test construction and before test
1131
# execution. Variables that the parameteriser sets need to be
1132
# ones that are not set by setUp, or setUp will trash them.
1133
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1134
self.transport_server = default_transport
1135
self.transport_readonly_server = None
1137
def failUnlessExists(self, path):
1138
"""Fail unless path, which may be abs or relative, exists."""
1139
self.failUnless(osutils.lexists(path))
1141
def failIfExists(self, path):
1142
"""Fail if path, which may be abs or relative, exists."""
1143
self.failIf(osutils.lexists(path))
1145
def get_transport(self):
1146
"""Return a writeable transport for the test scratch space"""
1147
t = get_transport(self.get_url())
1148
self.assertFalse(t.is_readonly())
1151
def get_readonly_transport(self):
1152
"""Return a readonly transport for the test scratch space
1154
This can be used to test that operations which should only need
1155
readonly access in fact do not try to write.
1157
t = get_transport(self.get_readonly_url())
1158
self.assertTrue(t.is_readonly())
1161
def get_readonly_server(self):
1162
"""Get the server instance for the readonly transport
1164
This is useful for some tests with specific servers to do diagnostics.
1166
if self.__readonly_server is None:
1167
if self.transport_readonly_server is None:
1168
# readonly decorator requested
1169
# bring up the server
1171
self.__readonly_server = ReadonlyServer()
1172
self.__readonly_server.setUp(self.__server)
1174
self.__readonly_server = self.transport_readonly_server()
1175
self.__readonly_server.setUp()
1176
self.addCleanup(self.__readonly_server.tearDown)
1177
return self.__readonly_server
1179
def get_readonly_url(self, relpath=None):
1180
"""Get a URL for the readonly transport.
1182
This will either be backed by '.' or a decorator to the transport
1183
used by self.get_url()
1184
relpath provides for clients to get a path relative to the base url.
1185
These should only be downwards relative, not upwards.
1187
base = self.get_readonly_server().get_url()
1188
if relpath is not None:
1189
if not base.endswith('/'):
1191
base = base + relpath
1194
def get_server(self):
1195
"""Get the read/write server instance.
1197
This is useful for some tests with specific servers that need
1200
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1201
is no means to override it.
1203
if self.__server is None:
1204
self.__server = MemoryServer()
1205
self.__server.setUp()
1206
self.addCleanup(self.__server.tearDown)
1207
return self.__server
1209
def get_url(self, relpath=None):
1210
"""Get a URL (or maybe a path) for the readwrite transport.
1212
This will either be backed by '.' or to an equivalent non-file based
1214
relpath provides for clients to get a path relative to the base url.
1215
These should only be downwards relative, not upwards.
1217
base = self.get_server().get_url()
1218
if relpath is not None and relpath != '.':
1219
if not base.endswith('/'):
1221
# XXX: Really base should be a url; we did after all call
1222
# get_url()! But sometimes it's just a path (from
1223
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1224
# to a non-escaped local path.
1225
if base.startswith('./') or base.startswith('/'):
1228
base += urlutils.escape(relpath)
1231
def _make_test_root(self):
1232
if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1236
root = u'test%04d.tmp' % i
1240
if e.errno == errno.EEXIST:
1245
# successfully created
1246
TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1248
# make a fake bzr directory there to prevent any tests propagating
1249
# up onto the source directory's real branch
1250
bzrdir.BzrDir.create_standalone_workingtree(
1251
TestCaseWithMemoryTransport.TEST_ROOT)
1253
def makeAndChdirToTestDir(self):
1254
"""Create a temporary directories for this one test.
1256
This must set self.test_home_dir and self.test_dir and chdir to
1259
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1261
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1262
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1263
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1265
def make_branch(self, relpath, format=None):
1266
"""Create a branch on the transport at relpath."""
1267
repo = self.make_repository(relpath, format=format)
1268
return repo.bzrdir.create_branch()
1270
def make_bzrdir(self, relpath, format=None):
1272
# might be a relative or absolute path
1273
maybe_a_url = self.get_url(relpath)
1274
segments = maybe_a_url.rsplit('/', 1)
1275
t = get_transport(maybe_a_url)
1276
if len(segments) > 1 and segments[-1] not in ('', '.'):
1279
except errors.FileExists:
1282
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1283
return format.initialize_on_transport(t)
1284
except errors.UninitializableFormat:
1285
raise TestSkipped("Format %s is not initializable." % format)
1287
def make_repository(self, relpath, shared=False, format=None):
1288
"""Create a repository on our default transport at relpath."""
1289
made_control = self.make_bzrdir(relpath, format=format)
1290
return made_control.create_repository(shared=shared)
1292
def make_branch_and_memory_tree(self, relpath):
1293
"""Create a branch on the default transport and a MemoryTree for it."""
1294
b = self.make_branch(relpath)
1295
return memorytree.MemoryTree.create_on_branch(b)
1297
def overrideEnvironmentForTesting(self):
1298
os.environ['HOME'] = self.test_home_dir
1299
os.environ['APPDATA'] = self.test_home_dir
1302
super(TestCaseWithMemoryTransport, self).setUp()
1303
self._make_test_root()
1304
_currentdir = os.getcwdu()
1305
def _leaveDirectory():
1306
os.chdir(_currentdir)
1307
self.addCleanup(_leaveDirectory)
1308
self.makeAndChdirToTestDir()
1309
self.overrideEnvironmentForTesting()
1310
self.__readonly_server = None
1311
self.__server = None
996
class TestCaseInTempDir(TestCase):
1314
class TestCaseInTempDir(TestCaseWithMemoryTransport):
997
1315
"""Derived class that runs a test within a temporary directory.
999
1317
This is useful for tests that need to create a branch, etc.
1146
1430
readwrite one must both define get_url() as resolving to os.getcwd().
1149
def __init__(self, methodName='testMethod'):
1150
super(TestCaseWithTransport, self).__init__(methodName)
1151
self.__readonly_server = None
1152
self.__server = None
1153
self.transport_server = default_transport
1154
self.transport_readonly_server = None
1156
def get_readonly_url(self, relpath=None):
1157
"""Get a URL for the readonly transport.
1159
This will either be backed by '.' or a decorator to the transport
1160
used by self.get_url()
1161
relpath provides for clients to get a path relative to the base url.
1162
These should only be downwards relative, not upwards.
1164
base = self.get_readonly_server().get_url()
1165
if relpath is not None:
1166
if not base.endswith('/'):
1168
base = base + relpath
1171
def get_readonly_server(self):
1172
"""Get the server instance for the readonly transport
1174
This is useful for some tests with specific servers to do diagnostics.
1176
if self.__readonly_server is None:
1177
if self.transport_readonly_server is None:
1178
# readonly decorator requested
1179
# bring up the server
1181
self.__readonly_server = ReadonlyServer()
1182
self.__readonly_server.setUp(self.__server)
1184
self.__readonly_server = self.transport_readonly_server()
1185
self.__readonly_server.setUp()
1186
self.addCleanup(self.__readonly_server.tearDown)
1187
return self.__readonly_server
1189
1433
def get_server(self):
1190
"""Get the read/write server instance.
1434
"""See TestCaseWithMemoryTransport.
1192
1436
This is useful for some tests with specific servers that need
1198
1442
self.addCleanup(self.__server.tearDown)
1199
1443
return self.__server
1201
def get_url(self, relpath=None):
1202
"""Get a URL for the readwrite transport.
1204
This will either be backed by '.' or to an equivalent non-file based
1206
relpath provides for clients to get a path relative to the base url.
1207
These should only be downwards relative, not upwards.
1209
base = self.get_server().get_url()
1210
if relpath is not None and relpath != '.':
1211
if not base.endswith('/'):
1213
base = base + urlutils.escape(relpath)
1216
def get_transport(self):
1217
"""Return a writeable transport for the test scratch space"""
1218
t = get_transport(self.get_url())
1219
self.assertFalse(t.is_readonly())
1222
def get_readonly_transport(self):
1223
"""Return a readonly transport for the test scratch space
1225
This can be used to test that operations which should only need
1226
readonly access in fact do not try to write.
1228
t = get_transport(self.get_readonly_url())
1229
self.assertTrue(t.is_readonly())
1232
def make_branch(self, relpath, format=None):
1233
"""Create a branch on the transport at relpath."""
1234
repo = self.make_repository(relpath, format=format)
1235
return repo.bzrdir.create_branch()
1237
def make_bzrdir(self, relpath, format=None):
1239
url = self.get_url(relpath)
1240
mutter('relpath %r => url %r', relpath, url)
1241
segments = url.split('/')
1242
if segments and segments[-1] not in ('', '.'):
1243
parent = '/'.join(segments[:-1])
1244
t = get_transport(parent)
1246
t.mkdir(segments[-1])
1247
except errors.FileExists:
1250
format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
1251
# FIXME: make this use a single transport someday. RBC 20060418
1252
return format.initialize_on_transport(get_transport(relpath))
1253
except errors.UninitializableFormat:
1254
raise TestSkipped("Format %s is not initializable." % format)
1256
def make_repository(self, relpath, shared=False, format=None):
1257
"""Create a repository on our default transport at relpath."""
1258
made_control = self.make_bzrdir(relpath, format=format)
1259
return made_control.create_repository(shared=shared)
1261
1445
def make_branch_and_tree(self, relpath, format=None):
1262
1446
"""Create a branch on the transport and a tree locally.
1448
If the transport is not a LocalTransport, the Tree can't be created on
1449
the transport. In that case the working tree is created in the local
1450
directory, and the returned tree's branch and repository will also be
1453
This will fail if the original default transport for this test
1454
case wasn't backed by the working directory, as the branch won't
1455
be on disk for us to open it.
1457
:param format: The BzrDirFormat.
1458
:returns: the WorkingTree.
1266
1460
# TODO: always use the local disk path for the working tree,
1267
1461
# this obviously requires a format that supports branch references