655
675
def _finishLogFile(self):
656
676
"""Finished with the log file.
658
Read contents into memory, close, and delete.
678
Close the file and delete it, unless setKeepLogfile was called.
660
680
if self._log_file is None:
662
682
bzrlib.trace.disable_test_log(self._log_nonce)
663
self._log_file.seek(0)
664
self._log_contents = self._log_file.read()
665
683
self._log_file.close()
666
os.remove(self._log_file_name)
667
self._log_file = self._log_file_name = None
684
self._log_file = None
685
if not self._keep_log_file:
686
os.remove(self._log_file_name)
687
self._log_file_name = None
689
def setKeepLogfile(self):
690
"""Make the logfile not be deleted when _finishLogFile is called."""
691
self._keep_log_file = True
669
693
def addCleanup(self, callable):
670
694
"""Arrange to run a callable when this case is torn down.
684
708
'BZR_EMAIL': None,
685
709
'BZREMAIL': None, # may still be present in the environment
711
'BZR_PROGRESS_BAR': None,
688
713
self.__old_env = {}
689
714
self.addCleanup(self._restoreEnvironment)
690
715
for name, value in new_env.iteritems():
691
716
self._captureVar(name, value)
694
718
def _captureVar(self, name, newvalue):
695
"""Set an environment variable, preparing it to be reset when finished."""
696
self.__old_env[name] = os.environ.get(name, None)
698
if name in os.environ:
701
os.environ[name] = newvalue
704
def _restoreVar(name, value):
706
if name in os.environ:
709
os.environ[name] = value
719
"""Set an environment variable, and reset it when finished."""
720
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
711
722
def _restoreEnvironment(self):
712
723
for name, value in self.__old_env.iteritems():
713
self._restoreVar(name, value)
724
osutils.set_or_unset_env(name, value)
715
726
def tearDown(self):
716
727
self._runCleanups()
751
762
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()
765
def _get_log(self, keep_log_file=False):
766
"""Return as a string the log for this test. If the file is still
767
on disk and keep_log_file=False, delete the log file and store the
768
content in self._log_contents."""
769
# flush the log file, to get all content
771
bzrlib.trace._trace_file.flush()
772
if self._log_contents:
759
773
return self._log_contents
760
# TODO: Delete the log after it's been read in
774
if self._log_file_name is not None:
775
logfile = open(self._log_file_name)
777
log_contents = logfile.read()
780
if not keep_log_file:
781
self._log_contents = log_contents
782
os.remove(self._log_file_name)
785
return "DELETED log file to reduce memory footprint"
762
787
def capture(self, cmd, retcode=0):
763
788
"""Shortcut that splits cmd into words, runs, and returns stdout"""
764
789
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
766
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None):
791
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None,
767
793
"""Invoke bzr and return (stdout, stderr).
769
795
Useful for code that wants to check the contents of the
891
928
variables. A value of None will unset the env variable.
892
929
The values must be strings. The change will only occur in the
893
930
child, so you don't need to fix the environment after running.
931
:param universal_newlines: Convert CRLF => LF
895
933
env_changes = kwargs.get('env_changes', {})
934
working_dir = kwargs.get('working_dir', None)
935
process = self.start_bzr_subprocess(args, env_changes=env_changes,
936
working_dir=working_dir)
937
# We distinguish between retcode=None and retcode not passed.
938
supplied_retcode = kwargs.get('retcode', 0)
939
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
940
universal_newlines=kwargs.get('universal_newlines', False),
943
def start_bzr_subprocess(self, process_args, env_changes=None,
944
skip_if_plan_to_signal=False,
946
"""Start bzr in a subprocess for testing.
948
This starts a new Python interpreter and runs bzr in there.
949
This should only be used for tests that have a justifiable need for
950
this isolation: e.g. they are testing startup time, or signal
951
handling, or early startup code, etc. Subprocess code can't be
952
profiled or debugged so easily.
954
:param process_args: a list of arguments to pass to the bzr executable,
955
for example `['--version']`.
956
:param env_changes: A dictionary which lists changes to environment
957
variables. A value of None will unset the env variable.
958
The values must be strings. The change will only occur in the
959
child, so you don't need to fix the environment after running.
960
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
963
:returns: Popen object for the started process.
965
if skip_if_plan_to_signal:
966
if not getattr(os, 'kill', None):
967
raise TestSkipped("os.kill not available.")
969
if env_changes is None:
896
973
def cleanup_environment():
897
974
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
975
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
977
def restore_environment():
978
for env_var, value in old_env.iteritems():
979
osutils.set_or_unset_env(env_var, value)
981
bzr_path = self.get_bzr_path()
984
if working_dir is not None:
985
cwd = osutils.getcwd()
986
os.chdir(working_dir)
989
# win32 subprocess doesn't support preexec_fn
990
# so we will avoid using it on all platforms, just to
991
# make sure the code path is used, and we don't break on win32
992
cleanup_environment()
993
process = Popen([sys.executable, bzr_path] + list(process_args),
994
stdin=PIPE, stdout=PIPE, stderr=PIPE)
996
restore_environment()
1002
def get_bzr_path(self):
1003
"""Return the path of the 'bzr' executable for this test suite."""
904
1004
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
1005
if not os.path.isfile(bzr_path):
1006
# We are probably installed. Assume sys.argv is the right file
1007
bzr_path = sys.argv[0]
1010
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1011
universal_newlines=False, process_args=None):
1012
"""Finish the execution of process.
1014
:param process: the Popen object returned from start_bzr_subprocess.
1015
:param retcode: The status code that is expected. Defaults to 0. If
1016
None is supplied, the status code is not checked.
1017
:param send_signal: an optional signal to send to the process.
1018
:param universal_newlines: Convert CRLF => LF
1019
:returns: (stdout, stderr)
1021
if send_signal is not None:
1022
os.kill(process.pid, send_signal)
1023
out, err = process.communicate()
1025
if universal_newlines:
1026
out = out.replace('\r\n', '\n')
1027
err = err.replace('\r\n', '\n')
1029
if retcode is not None and retcode != process.returncode:
1030
if process_args is None:
1031
process_args = "(unknown args)"
1032
mutter('Output of bzr %s:\n%s', process_args, out)
1033
mutter('Error for bzr %s:\n%s', process_args, err)
1034
self.fail('Command bzr %s failed with retcode %s != %s'
1035
% (process_args, retcode, process.returncode))
915
1036
return [out, err]
917
1038
def check_inventory_shape(self, inv, shape):
1228
1359
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)
1361
# might be a relative or absolute path
1362
maybe_a_url = self.get_url(relpath)
1363
segments = maybe_a_url.rsplit('/', 1)
1364
t = get_transport(maybe_a_url)
1365
if len(segments) > 1 and segments[-1] not in ('', '.'):
1237
t.mkdir(segments[-1])
1238
1368
except errors.FileExists:
1240
1370
if format is None:
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))
1371
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1372
return format.initialize_on_transport(t)
1244
1373
except errors.UninitializableFormat:
1245
1374
raise TestSkipped("Format %s is not initializable." % format)
1249
1378
made_control = self.make_bzrdir(relpath, format=format)
1250
1379
return made_control.create_repository(shared=shared)
1381
def make_branch_and_memory_tree(self, relpath):
1382
"""Create a branch on the default transport and a MemoryTree for it."""
1383
b = self.make_branch(relpath)
1384
return memorytree.MemoryTree.create_on_branch(b)
1252
1386
def make_branch_and_tree(self, relpath, format=None):
1253
1387
"""Create a branch on the transport and a tree locally.
1389
If the transport is not a LocalTransport, the Tree can't be created on
1390
the transport. In that case the working tree is created in the local
1391
directory, and the returned tree's branch and repository will also be
1394
This will fail if the original default transport for this test
1395
case wasn't backed by the working directory, as the branch won't
1396
be on disk for us to open it.
1398
:param format: The BzrDirFormat.
1399
:returns: the WorkingTree.
1257
1401
# TODO: always use the local disk path for the working tree,
1258
1402
# this obviously requires a format that supports branch references
added
removed
bzrlib/tests/__init__.py
