102
126
bzrlib.tests.bzrdir_implementations,
103
127
bzrlib.tests.interrepository_implementations,
104
128
bzrlib.tests.interversionedfile_implementations,
129
bzrlib.tests.intertree_implementations,
105
130
bzrlib.tests.repository_implementations,
106
131
bzrlib.tests.revisionstore_implementations,
132
bzrlib.tests.tree_implementations,
107
133
bzrlib.tests.workingtree_implementations,
111
class _MyResult(unittest._TextTestResult):
112
"""Custom TestResult.
137
class ExtendedTestResult(unittest._TextTestResult):
138
"""Accepts, reports and accumulates the results of running tests.
114
Shows output in a different format, including displaying runtime for tests.
140
Compared to this unittest version this class adds support for profiling,
141
benchmarking, stopping as soon as a test fails, and skipping tests.
142
There are further-specialized subclasses for different types of display.
116
145
stop_early = False
118
def _elapsedTime(self):
119
return "%5dms" % (1000 * (time.time() - self._start_time))
147
def __init__(self, stream, descriptions, verbosity,
151
"""Construct new TestResult.
153
:param bench_history: Optionally, a writable file object to accumulate
156
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
157
if bench_history is not None:
158
from bzrlib.version import _get_bzr_source_tree
159
src_tree = _get_bzr_source_tree()
162
revision_id = src_tree.get_parent_ids()[0]
164
# XXX: if this is a brand new tree, do the same as if there
168
# XXX: If there's no branch, what should we do?
170
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
171
self._bench_history = bench_history
172
self.ui = bzrlib.ui.ui_factory
173
self.num_tests = num_tests
175
self.failure_count = 0
178
self._overall_start_time = time.time()
180
def extractBenchmarkTime(self, testCase):
181
"""Add a benchmark time for the current test case."""
182
self._benchmarkTime = getattr(testCase, "_benchtime", None)
184
def _elapsedTestTimeString(self):
185
"""Return a time string for the overall time the current test has taken."""
186
return self._formatTime(time.time() - self._start_time)
188
def _testTimeString(self):
189
if self._benchmarkTime is not None:
191
self._formatTime(self._benchmarkTime),
192
self._elapsedTestTimeString())
194
return " %s" % self._elapsedTestTimeString()
196
def _formatTime(self, seconds):
197
"""Format seconds as milliseconds with leading spaces."""
198
return "%5dms" % (1000 * seconds)
200
def _shortened_test_description(self, test):
202
what = re.sub(r'^bzrlib\.(tests|benchmark)\.', '', what)
121
205
def startTest(self, test):
122
206
unittest.TestResult.startTest(self, test)
123
# In a short description, the important words are in
124
# the beginning, but in an id, the important words are
126
SHOW_DESCRIPTIONS = False
128
width = osutils.terminal_width()
129
name_width = width - 15
131
if SHOW_DESCRIPTIONS:
132
what = test.shortDescription()
134
if len(what) > name_width:
135
what = what[:name_width-3] + '...'
138
if what.startswith('bzrlib.tests.'):
140
if len(what) > name_width:
141
what = '...' + what[3-name_width:]
142
what = what.ljust(name_width)
143
self.stream.write(what)
207
self.report_test_start(test)
208
self._recordTestStartTime()
210
def _recordTestStartTime(self):
211
"""Record that a test has started."""
145
212
self._start_time = time.time()
147
214
def addError(self, test, err):
148
215
if isinstance(err[1], TestSkipped):
149
216
return self.addSkipped(test, err)
150
217
unittest.TestResult.addError(self, test, err)
152
self.stream.writeln("ERROR %s" % self._elapsedTime())
154
self.stream.write('E')
218
# We can only do this if we have one of our TestCases, not if
220
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
221
if setKeepLogfile is not None:
223
self.extractBenchmarkTime(test)
224
self.report_error(test, err)
156
225
if self.stop_early:
159
228
def addFailure(self, test, err):
160
229
unittest.TestResult.addFailure(self, test, err)
162
self.stream.writeln(" FAIL %s" % self._elapsedTime())
164
self.stream.write('F')
230
# We can only do this if we have one of our TestCases, not if
232
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
233
if setKeepLogfile is not None:
235
self.extractBenchmarkTime(test)
236
self.report_failure(test, err)
166
237
if self.stop_early:
169
240
def addSuccess(self, test):
171
self.stream.writeln(' OK %s' % self._elapsedTime())
173
self.stream.write('~')
241
self.extractBenchmarkTime(test)
242
if self._bench_history is not None:
243
if self._benchmarkTime is not None:
244
self._bench_history.write("%s %s\n" % (
245
self._formatTime(self._benchmarkTime),
247
self.report_success(test)
175
248
unittest.TestResult.addSuccess(self, test)
177
250
def addSkipped(self, test, skip_excinfo):
179
print >>self.stream, ' SKIP %s' % self._elapsedTime()
180
print >>self.stream, ' %s' % skip_excinfo[1]
182
self.stream.write('S')
251
self.extractBenchmarkTime(test)
252
self.report_skip(test, skip_excinfo)
184
253
# seems best to treat this as success from point-of-view of unittest
185
254
# -- it actually does nothing so it barely matters :)
186
unittest.TestResult.addSuccess(self, test)
257
except KeyboardInterrupt:
260
self.addError(test, test.__exc_info())
262
unittest.TestResult.addSuccess(self, test)
188
264
def printErrorList(self, flavour, errors):
189
265
for test, err in errors:
199
275
self.stream.writeln(self.separator2)
200
276
self.stream.writeln("%s" % err)
203
class TextTestRunner(unittest.TextTestRunner):
281
def report_cleaning_up(self):
284
def report_success(self, test):
288
class TextTestResult(ExtendedTestResult):
289
"""Displays progress and results of tests in text form"""
291
def __init__(self, *args, **kw):
292
ExtendedTestResult.__init__(self, *args, **kw)
293
self.pb = self.ui.nested_progress_bar()
294
self.pb.show_pct = False
295
self.pb.show_spinner = False
296
self.pb.show_eta = False,
297
self.pb.show_count = False
298
self.pb.show_bar = False
300
def report_starting(self):
301
self.pb.update('[test 0/%d] starting...' % (self.num_tests))
303
def _progress_prefix_text(self):
304
a = '[%d' % self.count
305
if self.num_tests is not None:
306
a +='/%d' % self.num_tests
307
a += ' in %ds' % (time.time() - self._overall_start_time)
309
a += ', %d errors' % self.error_count
310
if self.failure_count:
311
a += ', %d failed' % self.failure_count
313
a += ', %d skipped' % self.skip_count
317
def report_test_start(self, test):
320
self._progress_prefix_text()
322
+ self._shortened_test_description(test))
324
def report_error(self, test, err):
325
self.error_count += 1
326
self.pb.note('ERROR: %s\n %s\n' % (
327
self._shortened_test_description(test),
331
def report_failure(self, test, err):
332
self.failure_count += 1
333
self.pb.note('FAIL: %s\n %s\n' % (
334
self._shortened_test_description(test),
338
def report_skip(self, test, skip_excinfo):
341
# at the moment these are mostly not things we can fix
342
# and so they just produce stipple; use the verbose reporter
345
# show test and reason for skip
346
self.pb.note('SKIP: %s\n %s\n' % (
347
self._shortened_test_description(test),
350
# since the class name was left behind in the still-visible
352
self.pb.note('SKIP: %s' % (skip_excinfo[1]))
354
def report_cleaning_up(self):
355
self.pb.update('cleaning up...')
361
class VerboseTestResult(ExtendedTestResult):
362
"""Produce long output, with one line per test run plus times"""
364
def _ellipsize_to_right(self, a_string, final_width):
365
"""Truncate and pad a string, keeping the right hand side"""
366
if len(a_string) > final_width:
367
result = '...' + a_string[3-final_width:]
370
return result.ljust(final_width)
372
def report_starting(self):
373
self.stream.write('running %d tests...\n' % self.num_tests)
375
def report_test_start(self, test):
377
name = self._shortened_test_description(test)
378
self.stream.write(self._ellipsize_to_right(name, 60))
381
def report_error(self, test, err):
382
self.error_count += 1
383
self.stream.writeln('ERROR %s\n %s'
384
% (self._testTimeString(), err[1]))
386
def report_failure(self, test, err):
387
self.failure_count += 1
388
self.stream.writeln('FAIL %s\n %s'
389
% (self._testTimeString(), err[1]))
391
def report_success(self, test):
392
self.stream.writeln(' OK %s' % self._testTimeString())
393
for bench_called, stats in getattr(test, '_benchcalls', []):
394
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
395
stats.pprint(file=self.stream)
398
def report_skip(self, test, skip_excinfo):
399
print >>self.stream, ' SKIP %s' % self._testTimeString()
400
print >>self.stream, ' %s' % skip_excinfo[1]
403
class TextTestRunner(object):
204
404
stop_on_failure = False
206
def _makeResult(self):
207
result = _MyResult(self.stream, self.descriptions, self.verbosity)
412
self.stream = unittest._WritelnDecorator(stream)
413
self.descriptions = descriptions
414
self.verbosity = verbosity
415
self.keep_output = keep_output
416
self._bench_history = bench_history
419
"Run the given test case or test suite."
420
startTime = time.time()
421
if self.verbosity == 1:
422
result_class = TextTestResult
423
elif self.verbosity >= 2:
424
result_class = VerboseTestResult
425
result = result_class(self.stream,
428
bench_history=self._bench_history,
429
num_tests=test.countTestCases(),
208
431
result.stop_early = self.stop_on_failure
432
result.report_starting()
434
stopTime = time.time()
435
timeTaken = stopTime - startTime
437
self.stream.writeln(result.separator2)
438
run = result.testsRun
439
self.stream.writeln("Ran %d test%s in %.3fs" %
440
(run, run != 1 and "s" or "", timeTaken))
441
self.stream.writeln()
442
if not result.wasSuccessful():
443
self.stream.write("FAILED (")
444
failed, errored = map(len, (result.failures, result.errors))
446
self.stream.write("failures=%d" % failed)
448
if failed: self.stream.write(", ")
449
self.stream.write("errors=%d" % errored)
450
self.stream.writeln(")")
452
self.stream.writeln("OK")
453
result.report_cleaning_up()
454
# This is still a little bogus,
455
# but only a little. Folk not using our testrunner will
456
# have to delete their temp directories themselves.
457
test_root = TestCaseWithMemoryTransport.TEST_ROOT
458
if result.wasSuccessful() or not self.keep_output:
459
if test_root is not None:
460
# If LANG=C we probably have created some bogus paths
461
# which rmtree(unicode) will fail to delete
462
# so make sure we are using rmtree(str) to delete everything
463
# except on win32, where rmtree(str) will fail
464
# since it doesn't have the property of byte-stream paths
465
# (they are either ascii or mbcs)
466
if sys.platform == 'win32':
467
# make sure we are using the unicode win32 api
468
test_root = unicode(test_root)
470
test_root = test_root.encode(
471
sys.getfilesystemencoding())
472
osutils.rmtree(test_root)
474
note("Failed tests working directories are in '%s'\n", test_root)
475
TestCaseWithMemoryTransport.TEST_ROOT = None
341
656
def assertIsInstance(self, obj, kls):
342
657
"""Fail if obj is not an instance of kls"""
343
658
if not isinstance(obj, kls):
344
self.fail("%r is not an instance of %s" % (obj, kls))
659
self.fail("%r is an instance of %s rather than %s" % (
660
obj, obj.__class__, kls))
662
def _capture_warnings(self, a_callable, *args, **kwargs):
663
"""A helper for callDeprecated and applyDeprecated.
665
:param a_callable: A callable to call.
666
:param args: The positional arguments for the callable
667
:param kwargs: The keyword arguments for the callable
668
:return: A tuple (warnings, result). result is the result of calling
669
a_callable(*args, **kwargs).
672
def capture_warnings(msg, cls=None, stacklevel=None):
673
# we've hooked into a deprecation specific callpath,
674
# only deprecations should getting sent via it.
675
self.assertEqual(cls, DeprecationWarning)
676
local_warnings.append(msg)
677
original_warning_method = symbol_versioning.warn
678
symbol_versioning.set_warning_method(capture_warnings)
680
result = a_callable(*args, **kwargs)
682
symbol_versioning.set_warning_method(original_warning_method)
683
return (local_warnings, result)
685
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
686
"""Call a deprecated callable without warning the user.
688
:param deprecation_format: The deprecation format that the callable
689
should have been deprecated with. This is the same type as the
690
parameter to deprecated_method/deprecated_function. If the
691
callable is not deprecated with this format, an assertion error
693
:param a_callable: A callable to call. This may be a bound method or
694
a regular function. It will be called with *args and **kwargs.
695
:param args: The positional arguments for the callable
696
:param kwargs: The keyword arguments for the callable
697
:return: The result of a_callable(*args, **kwargs)
699
call_warnings, result = self._capture_warnings(a_callable,
701
expected_first_warning = symbol_versioning.deprecation_string(
702
a_callable, deprecation_format)
703
if len(call_warnings) == 0:
704
self.fail("No assertion generated by call to %s" %
706
self.assertEqual(expected_first_warning, call_warnings[0])
709
def callDeprecated(self, expected, callable, *args, **kwargs):
710
"""Assert that a callable is deprecated in a particular way.
712
This is a very precise test for unusual requirements. The
713
applyDeprecated helper function is probably more suited for most tests
714
as it allows you to simply specify the deprecation format being used
715
and will ensure that that is issued for the function being called.
717
:param expected: a list of the deprecation warnings expected, in order
718
:param callable: The callable to call
719
:param args: The positional arguments for the callable
720
:param kwargs: The keyword arguments for the callable
722
call_warnings, result = self._capture_warnings(callable,
724
self.assertEqual(expected, call_warnings)
346
727
def _startLogFile(self):
347
728
"""Send bzr and test log messages to a temporary file.
381
767
def _cleanEnvironment(self):
769
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
383
770
'HOME': os.getcwd(),
384
771
'APPDATA': os.getcwd(),
773
'BZREMAIL': None, # may still be present in the environment
775
'BZR_PROGRESS_BAR': None,
388
777
self.__old_env = {}
389
778
self.addCleanup(self._restoreEnvironment)
390
779
for name, value in new_env.iteritems():
391
780
self._captureVar(name, value)
394
782
def _captureVar(self, name, newvalue):
395
"""Set an environment variable, preparing it to be reset when finished."""
396
self.__old_env[name] = os.environ.get(name, None)
398
if name in os.environ:
401
os.environ[name] = newvalue
404
def _restoreVar(name, value):
406
if name in os.environ:
409
os.environ[name] = value
783
"""Set an environment variable, and reset it when finished."""
784
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
411
786
def _restoreEnvironment(self):
412
787
for name, value in self.__old_env.iteritems():
413
self._restoreVar(name, value)
788
osutils.set_or_unset_env(name, value)
415
790
def tearDown(self):
416
791
self._runCleanups()
417
792
unittest.TestCase.tearDown(self)
794
def time(self, callable, *args, **kwargs):
795
"""Run callable and accrue the time it takes to the benchmark time.
797
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
798
this will cause lsprofile statistics to be gathered and stored in
801
if self._benchtime is None:
805
if not self._gather_lsprof_in_benchmarks:
806
return callable(*args, **kwargs)
808
# record this benchmark
809
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
811
self._benchcalls.append(((callable, args, kwargs), stats))
814
self._benchtime += time.time() - start
419
816
def _runCleanups(self):
420
817
"""Run registered cleanup functions.
458
870
errors, and with logging set to something approximating the
459
871
default, so that error reporting can be checked.
461
argv -- arguments to invoke bzr
462
retcode -- expected return code, or None for don't-care.
873
:param argv: arguments to invoke bzr
874
:param retcode: expected return code, or None for don't-care.
875
:param encoding: encoding for sys.stdout and sys.stderr
876
:param stdin: A string to be used as stdin for the command.
877
:param working_dir: Change to this directory before running
466
self.log('run bzr: %s', ' '.join(argv))
880
encoding = bzrlib.user_encoding
881
if stdin is not None:
882
stdin = StringIO(stdin)
883
stdout = StringIOWrapper()
884
stderr = StringIOWrapper()
885
stdout.encoding = encoding
886
stderr.encoding = encoding
888
self.log('run bzr: %r', argv)
467
889
# FIXME: don't call into logging here
468
890
handler = logging.StreamHandler(stderr)
469
handler.setFormatter(bzrlib.trace.QuietFormatter())
470
891
handler.setLevel(logging.INFO)
471
892
logger = logging.getLogger('')
472
893
logger.addHandler(handler)
894
old_ui_factory = bzrlib.ui.ui_factory
895
bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
898
bzrlib.ui.ui_factory.stdin = stdin
901
if working_dir is not None:
902
cwd = osutils.getcwd()
903
os.chdir(working_dir)
474
result = self.apply_redirected(None, stdout, stderr,
475
bzrlib.commands.run_bzr_catch_errors,
906
saved_debug_flags = frozenset(debug.debug_flags)
907
debug.debug_flags.clear()
909
result = self.apply_redirected(stdin, stdout, stderr,
910
bzrlib.commands.run_bzr_catch_errors,
913
debug.debug_flags.update(saved_debug_flags)
478
915
logger.removeHandler(handler)
916
bzrlib.ui.ui_factory = old_ui_factory
479
920
out = stdout.getvalue()
480
921
err = stderr.getvalue()
482
self.log('output:\n%s', out)
923
self.log('output:\n%r', out)
484
self.log('errors:\n%s', err)
925
self.log('errors:\n%r', err)
485
926
if retcode is not None:
486
self.assertEquals(result, retcode)
927
self.assertEquals(retcode, result)
489
930
def run_bzr(self, *args, **kwargs):
496
937
This sends the stdout/stderr results into the test's log,
497
938
where it may be useful for debugging. See also run_captured.
940
:param stdin: A string to be used as stdin for the command.
499
942
retcode = kwargs.pop('retcode', 0)
500
return self.run_bzr_captured(args, retcode)
943
encoding = kwargs.pop('encoding', None)
944
stdin = kwargs.pop('stdin', None)
945
working_dir = kwargs.pop('working_dir', None)
946
return self.run_bzr_captured(args, retcode=retcode, encoding=encoding,
947
stdin=stdin, working_dir=working_dir)
949
def run_bzr_decode(self, *args, **kwargs):
950
if 'encoding' in kwargs:
951
encoding = kwargs['encoding']
953
encoding = bzrlib.user_encoding
954
return self.run_bzr(*args, **kwargs)[0].decode(encoding)
956
def run_bzr_error(self, error_regexes, *args, **kwargs):
957
"""Run bzr, and check that stderr contains the supplied regexes
959
:param error_regexes: Sequence of regular expressions which
960
must each be found in the error output. The relative ordering
962
:param args: command-line arguments for bzr
963
:param kwargs: Keyword arguments which are interpreted by run_bzr
964
This function changes the default value of retcode to be 3,
965
since in most cases this is run when you expect bzr to fail.
966
:return: (out, err) The actual output of running the command (in case you
967
want to do more inspection)
970
# Make sure that commit is failing because there is nothing to do
971
self.run_bzr_error(['no changes to commit'],
972
'commit', '-m', 'my commit comment')
973
# Make sure --strict is handling an unknown file, rather than
974
# giving us the 'nothing to do' error
975
self.build_tree(['unknown'])
976
self.run_bzr_error(['Commit refused because there are unknown files'],
977
'commit', '--strict', '-m', 'my commit comment')
979
kwargs.setdefault('retcode', 3)
980
out, err = self.run_bzr(*args, **kwargs)
981
for regex in error_regexes:
982
self.assertContainsRe(err, regex)
985
def run_bzr_subprocess(self, *args, **kwargs):
986
"""Run bzr in a subprocess for testing.
988
This starts a new Python interpreter and runs bzr in there.
989
This should only be used for tests that have a justifiable need for
990
this isolation: e.g. they are testing startup time, or signal
991
handling, or early startup code, etc. Subprocess code can't be
992
profiled or debugged so easily.
994
:param retcode: The status code that is expected. Defaults to 0. If
995
None is supplied, the status code is not checked.
996
:param env_changes: A dictionary which lists changes to environment
997
variables. A value of None will unset the env variable.
998
The values must be strings. The change will only occur in the
999
child, so you don't need to fix the environment after running.
1000
:param universal_newlines: Convert CRLF => LF
1001
:param allow_plugins: By default the subprocess is run with
1002
--no-plugins to ensure test reproducibility. Also, it is possible
1003
for system-wide plugins to create unexpected output on stderr,
1004
which can cause unnecessary test failures.
1006
env_changes = kwargs.get('env_changes', {})
1007
working_dir = kwargs.get('working_dir', None)
1008
allow_plugins = kwargs.get('allow_plugins', False)
1009
process = self.start_bzr_subprocess(args, env_changes=env_changes,
1010
working_dir=working_dir,
1011
allow_plugins=allow_plugins)
1012
# We distinguish between retcode=None and retcode not passed.
1013
supplied_retcode = kwargs.get('retcode', 0)
1014
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1015
universal_newlines=kwargs.get('universal_newlines', False),
1018
def start_bzr_subprocess(self, process_args, env_changes=None,
1019
skip_if_plan_to_signal=False,
1021
allow_plugins=False):
1022
"""Start bzr in a subprocess for testing.
1024
This starts a new Python interpreter and runs bzr in there.
1025
This should only be used for tests that have a justifiable need for
1026
this isolation: e.g. they are testing startup time, or signal
1027
handling, or early startup code, etc. Subprocess code can't be
1028
profiled or debugged so easily.
1030
:param process_args: a list of arguments to pass to the bzr executable,
1031
for example `['--version']`.
1032
:param env_changes: A dictionary which lists changes to environment
1033
variables. A value of None will unset the env variable.
1034
The values must be strings. The change will only occur in the
1035
child, so you don't need to fix the environment after running.
1036
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1038
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1040
:returns: Popen object for the started process.
1042
if skip_if_plan_to_signal:
1043
if not getattr(os, 'kill', None):
1044
raise TestSkipped("os.kill not available.")
1046
if env_changes is None:
1050
def cleanup_environment():
1051
for env_var, value in env_changes.iteritems():
1052
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1054
def restore_environment():
1055
for env_var, value in old_env.iteritems():
1056
osutils.set_or_unset_env(env_var, value)
1058
bzr_path = self.get_bzr_path()
1061
if working_dir is not None:
1062
cwd = osutils.getcwd()
1063
os.chdir(working_dir)
1066
# win32 subprocess doesn't support preexec_fn
1067
# so we will avoid using it on all platforms, just to
1068
# make sure the code path is used, and we don't break on win32
1069
cleanup_environment()
1070
command = [sys.executable, bzr_path]
1071
if not allow_plugins:
1072
command.append('--no-plugins')
1073
command.extend(process_args)
1074
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1076
restore_environment()
1082
def _popen(self, *args, **kwargs):
1083
"""Place a call to Popen.
1085
Allows tests to override this method to intercept the calls made to
1086
Popen for introspection.
1088
return Popen(*args, **kwargs)
1090
def get_bzr_path(self):
1091
"""Return the path of the 'bzr' executable for this test suite."""
1092
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1093
if not os.path.isfile(bzr_path):
1094
# We are probably installed. Assume sys.argv is the right file
1095
bzr_path = sys.argv[0]
1098
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1099
universal_newlines=False, process_args=None):
1100
"""Finish the execution of process.
1102
:param process: the Popen object returned from start_bzr_subprocess.
1103
:param retcode: The status code that is expected. Defaults to 0. If
1104
None is supplied, the status code is not checked.
1105
:param send_signal: an optional signal to send to the process.
1106
:param universal_newlines: Convert CRLF => LF
1107
:returns: (stdout, stderr)
1109
if send_signal is not None:
1110
os.kill(process.pid, send_signal)
1111
out, err = process.communicate()
1113
if universal_newlines:
1114
out = out.replace('\r\n', '\n')
1115
err = err.replace('\r\n', '\n')
1117
if retcode is not None and retcode != process.returncode:
1118
if process_args is None:
1119
process_args = "(unknown args)"
1120
mutter('Output of bzr %s:\n%s', process_args, out)
1121
mutter('Error for bzr %s:\n%s', process_args, err)
1122
self.fail('Command bzr %s failed with retcode %s != %s'
1123
% (process_args, retcode, process.returncode))
502
1126
def check_inventory_shape(self, inv, shape):
503
1127
"""Compare an inventory to a list of expected names.
562
1187
base_rev = common_ancestor(branch_from.last_revision(),
563
1188
wt_to.branch.last_revision(),
564
1189
wt_to.branch.repository)
565
merge_inner(wt_to.branch, branch_from.basis_tree(),
1190
merge_inner(wt_to.branch, branch_from.basis_tree(),
566
1191
wt_to.branch.repository.revision_tree(base_rev),
567
1192
this_tree=wt_to)
568
wt_to.add_pending_merge(branch_from.last_revision())
1193
wt_to.add_parent_tree_id(branch_from.last_revision())
571
1196
BzrTestBase = TestCase
1199
class TestCaseWithMemoryTransport(TestCase):
1200
"""Common test class for tests that do not need disk resources.
1202
Tests that need disk resources should derive from TestCaseWithTransport.
1204
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1206
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1207
a directory which does not exist. This serves to help ensure test isolation
1208
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1209
must exist. However, TestCaseWithMemoryTransport does not offer local
1210
file defaults for the transport in tests, nor does it obey the command line
1211
override, so tests that accidentally write to the common directory should
1219
def __init__(self, methodName='runTest'):
1220
# allow test parameterisation after test construction and before test
1221
# execution. Variables that the parameteriser sets need to be
1222
# ones that are not set by setUp, or setUp will trash them.
1223
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1224
self.transport_server = default_transport
1225
self.transport_readonly_server = None
1227
def failUnlessExists(self, path):
1228
"""Fail unless path, which may be abs or relative, exists."""
1229
self.failUnless(osutils.lexists(path))
1231
def failIfExists(self, path):
1232
"""Fail if path, which may be abs or relative, exists."""
1233
self.failIf(osutils.lexists(path))
1235
def get_transport(self):
1236
"""Return a writeable transport for the test scratch space"""
1237
t = get_transport(self.get_url())
1238
self.assertFalse(t.is_readonly())
1241
def get_readonly_transport(self):
1242
"""Return a readonly transport for the test scratch space
1244
This can be used to test that operations which should only need
1245
readonly access in fact do not try to write.
1247
t = get_transport(self.get_readonly_url())
1248
self.assertTrue(t.is_readonly())
1251
def get_readonly_server(self):
1252
"""Get the server instance for the readonly transport
1254
This is useful for some tests with specific servers to do diagnostics.
1256
if self.__readonly_server is None:
1257
if self.transport_readonly_server is None:
1258
# readonly decorator requested
1259
# bring up the server
1261
self.__readonly_server = ReadonlyServer()
1262
self.__readonly_server.setUp(self.__server)
1264
self.__readonly_server = self.transport_readonly_server()
1265
self.__readonly_server.setUp()
1266
self.addCleanup(self.__readonly_server.tearDown)
1267
return self.__readonly_server
1269
def get_readonly_url(self, relpath=None):
1270
"""Get a URL for the readonly transport.
1272
This will either be backed by '.' or a decorator to the transport
1273
used by self.get_url()
1274
relpath provides for clients to get a path relative to the base url.
1275
These should only be downwards relative, not upwards.
1277
base = self.get_readonly_server().get_url()
1278
if relpath is not None:
1279
if not base.endswith('/'):
1281
base = base + relpath
1284
def get_server(self):
1285
"""Get the read/write server instance.
1287
This is useful for some tests with specific servers that need
1290
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1291
is no means to override it.
1293
if self.__server is None:
1294
self.__server = MemoryServer()
1295
self.__server.setUp()
1296
self.addCleanup(self.__server.tearDown)
1297
return self.__server
1299
def get_url(self, relpath=None):
1300
"""Get a URL (or maybe a path) for the readwrite transport.
1302
This will either be backed by '.' or to an equivalent non-file based
1304
relpath provides for clients to get a path relative to the base url.
1305
These should only be downwards relative, not upwards.
1307
base = self.get_server().get_url()
1308
if relpath is not None and relpath != '.':
1309
if not base.endswith('/'):
1311
# XXX: Really base should be a url; we did after all call
1312
# get_url()! But sometimes it's just a path (from
1313
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1314
# to a non-escaped local path.
1315
if base.startswith('./') or base.startswith('/'):
1318
base += urlutils.escape(relpath)
1321
def _make_test_root(self):
1322
if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1326
root = u'test%04d.tmp' % i
1330
if e.errno == errno.EEXIST:
1335
# successfully created
1336
TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1338
# make a fake bzr directory there to prevent any tests propagating
1339
# up onto the source directory's real branch
1340
bzrdir.BzrDir.create_standalone_workingtree(
1341
TestCaseWithMemoryTransport.TEST_ROOT)
1343
def makeAndChdirToTestDir(self):
1344
"""Create a temporary directories for this one test.
1346
This must set self.test_home_dir and self.test_dir and chdir to
1349
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1351
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1352
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1353
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1355
def make_branch(self, relpath, format=None):
1356
"""Create a branch on the transport at relpath."""
1357
repo = self.make_repository(relpath, format=format)
1358
return repo.bzrdir.create_branch()
1360
def make_bzrdir(self, relpath, format=None):
1362
# might be a relative or absolute path
1363
maybe_a_url = self.get_url(relpath)
1364
segments = maybe_a_url.rsplit('/', 1)
1365
t = get_transport(maybe_a_url)
1366
if len(segments) > 1 and segments[-1] not in ('', '.'):
1369
except errors.FileExists:
1372
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1373
return format.initialize_on_transport(t)
1374
except errors.UninitializableFormat:
1375
raise TestSkipped("Format %s is not initializable." % format)
1377
def make_repository(self, relpath, shared=False, format=None):
1378
"""Create a repository on our default transport at relpath."""
1379
made_control = self.make_bzrdir(relpath, format=format)
1380
return made_control.create_repository(shared=shared)
1382
def make_branch_and_memory_tree(self, relpath, format=None):
1383
"""Create a branch on the default transport and a MemoryTree for it."""
1384
b = self.make_branch(relpath, format=format)
1385
return memorytree.MemoryTree.create_on_branch(b)
1387
def overrideEnvironmentForTesting(self):
1388
os.environ['HOME'] = self.test_home_dir
1389
os.environ['APPDATA'] = self.test_home_dir
1392
super(TestCaseWithMemoryTransport, self).setUp()
1393
self._make_test_root()
1394
_currentdir = os.getcwdu()
1395
def _leaveDirectory():
1396
os.chdir(_currentdir)
1397
self.addCleanup(_leaveDirectory)
1398
self.makeAndChdirToTestDir()
1399
self.overrideEnvironmentForTesting()
1400
self.__readonly_server = None
1401
self.__server = None
574
class TestCaseInTempDir(TestCase):
1404
class TestCaseInTempDir(TestCaseWithMemoryTransport):
575
1405
"""Derived class that runs a test within a temporary directory.
577
1407
This is useful for tests that need to create a branch, etc.
676
1485
end = os.linesep
678
1487
raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
679
content = "contents of %s%s" % (name, end)
680
transport.put(urlescape(name), StringIO(content))
1488
content = "contents of %s%s" % (name.encode('utf-8'), end)
1489
# Technically 'put()' is the right command. However, put
1490
# uses an AtomicFile, which requires an extra rename into place
1491
# As long as the files didn't exist in the past, append() will
1492
# do the same thing as put()
1493
# On jam's machine, make_kernel_like_tree is:
1494
# put: 4.5-7.5s (averaging 6s)
1496
# put_non_atomic: 2.9-4.5s
1497
transport.put_bytes_non_atomic(urlutils.escape(name), content)
682
1499
def build_tree_contents(self, shape):
683
1500
build_tree_contents(shape)
685
def failUnlessExists(self, path):
686
"""Fail unless path, which may be abs or relative, exists."""
687
self.failUnless(osutils.lexists(path))
689
def failIfExists(self, path):
690
"""Fail if path, which may be abs or relative, exists."""
691
self.failIf(osutils.lexists(path))
693
1502
def assertFileEqual(self, content, path):
694
1503
"""Fail if path does not contain 'content'."""
695
1504
self.failUnless(osutils.lexists(path))
1505
# TODO: jam 20060427 Shouldn't this be 'rb'?
696
1506
self.assertEqualDiff(content, open(path, 'r').read())
762
1532
self.addCleanup(self.__server.tearDown)
763
1533
return self.__server
765
def get_url(self, relpath=None):
766
"""Get a URL for the readwrite transport.
768
This will either be backed by '.' or to an equivalent non-file based
770
relpath provides for clients to get a path relative to the base url.
771
These should only be downwards relative, not upwards.
773
base = self.get_server().get_url()
774
if relpath is not None and relpath != '.':
775
if not base.endswith('/'):
777
base = base + relpath
780
def get_transport(self):
781
"""Return a writeable transport for the test scratch space"""
782
t = get_transport(self.get_url())
783
self.assertFalse(t.is_readonly())
786
def get_readonly_transport(self):
787
"""Return a readonly transport for the test scratch space
789
This can be used to test that operations which should only need
790
readonly access in fact do not try to write.
792
t = get_transport(self.get_readonly_url())
793
self.assertTrue(t.is_readonly())
796
def make_branch(self, relpath):
797
"""Create a branch on the transport at relpath."""
798
repo = self.make_repository(relpath)
799
return repo.bzrdir.create_branch()
801
def make_bzrdir(self, relpath):
803
url = self.get_url(relpath)
804
segments = relpath.split('/')
805
if segments and segments[-1] not in ('', '.'):
806
parent = self.get_url('/'.join(segments[:-1]))
807
t = get_transport(parent)
809
t.mkdir(segments[-1])
810
except errors.FileExists:
812
return bzrlib.bzrdir.BzrDir.create(url)
813
except errors.UninitializableFormat:
814
raise TestSkipped("Format %s is not initializable.")
816
def make_repository(self, relpath, shared=False):
817
"""Create a repository on our default transport at relpath."""
818
made_control = self.make_bzrdir(relpath)
819
return made_control.create_repository(shared=shared)
821
def make_branch_and_tree(self, relpath):
1535
def make_branch_and_tree(self, relpath, format=None):
822
1536
"""Create a branch on the transport and a tree locally.
1538
If the transport is not a LocalTransport, the Tree can't be created on
1539
the transport. In that case the working tree is created in the local
1540
directory, and the returned tree's branch and repository will also be
1543
This will fail if the original default transport for this test
1544
case wasn't backed by the working directory, as the branch won't
1545
be on disk for us to open it.
1547
:param format: The BzrDirFormat.
1548
:returns: the WorkingTree.
826
1550
# TODO: always use the local disk path for the working tree,
827
1551
# this obviously requires a format that supports branch references
828
1552
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
830
b = self.make_branch(relpath)
1554
b = self.make_branch(relpath, format=format)
832
1556
return b.bzrdir.create_workingtree()
833
1557
except errors.NotLocalUrl:
834
# new formats - catch No tree error and create
835
# a branch reference and a checkout.
836
# old formats at that point - raise TestSkipped.
838
return WorkingTreeFormat2().initialize(bzrdir.BzrDir.open(relpath))
1558
# We can only make working trees locally at the moment. If the
1559
# transport can't support them, then reopen the branch on a local
1560
# transport, and create the working tree there.
1562
# Possibly we should instead keep
1563
# the non-disk-backed branch and create a local checkout?
1564
bd = bzrdir.BzrDir.open(relpath)
1565
return bd.create_workingtree()
840
1567
def assertIsDirectory(self, relpath, transport):
841
1568
"""Assert that relpath within transport is a directory.
883
1614
def run_suite(suite, name='test', verbose=False, pattern=".*",
884
1615
stop_on_failure=False, keep_output=False,
886
TestCaseInTempDir._TEST_NAME = name
1616
transport=None, lsprof_timed=None, bench_history=None):
1617
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
891
1622
runner = TextTestRunner(stream=sys.stdout,
1624
verbosity=verbosity,
1625
keep_output=keep_output,
1626
bench_history=bench_history)
894
1627
runner.stop_on_failure=stop_on_failure
895
1628
if pattern != '.*':
896
1629
suite = filter_suite_by_re(suite, pattern)
897
1630
result = runner.run(suite)
898
# This is still a little bogus,
899
# but only a little. Folk not using our testrunner will
900
# have to delete their temp directories themselves.
901
test_root = TestCaseInTempDir.TEST_ROOT
902
if result.wasSuccessful() or not keep_output:
903
if test_root is not None:
904
print 'Deleting test root %s...' % test_root
906
shutil.rmtree(test_root)
910
print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
911
1631
return result.wasSuccessful()
914
1634
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
915
1635
keep_output=False,
1637
test_suite_factory=None,
1639
bench_history=None):
917
1640
"""Run the whole test suite under the enhanced runner"""
1641
# XXX: Very ugly way to do this...
1642
# Disable warning about old formats because we don't want it to disturb
1643
# any blackbox tests.
1644
from bzrlib import repository
1645
repository._deprecation_warning_done = True
918
1647
global default_transport
919
1648
if transport is None:
920
1649
transport = default_transport
921
1650
old_transport = default_transport
922
1651
default_transport = transport
1653
if test_suite_factory is None:
1654
suite = test_suite()
1656
suite = test_suite_factory()
925
1657
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
926
1658
stop_on_failure=stop_on_failure, keep_output=keep_output,
1659
transport=transport,
1660
lsprof_timed=lsprof_timed,
1661
bench_history=bench_history)
929
1663
default_transport = old_transport
933
1666
def test_suite():
934
"""Build and return TestSuite for the whole program."""
935
from doctest import DocTestSuite
937
global MODULES_TO_DOCTEST
1667
"""Build and return TestSuite for the whole of bzrlib.
1669
This function can be replaced if you need to change the default test
1670
suite on a global basis, but it is not encouraged.
940
1673
'bzrlib.tests.test_ancestry',
941
'bzrlib.tests.test_annotate',
942
1674
'bzrlib.tests.test_api',
1675
'bzrlib.tests.test_atomicfile',
943
1676
'bzrlib.tests.test_bad_files',
944
'bzrlib.tests.test_basis_inventory',
945
1677
'bzrlib.tests.test_branch',
1678
'bzrlib.tests.test_bundle',
946
1679
'bzrlib.tests.test_bzrdir',
1680
'bzrlib.tests.test_cache_utf8',
947
1681
'bzrlib.tests.test_command',
948
1682
'bzrlib.tests.test_commit',
949
1683
'bzrlib.tests.test_commit_merge',
973
1714
'bzrlib.tests.test_nonascii',
974
1715
'bzrlib.tests.test_options',
975
1716
'bzrlib.tests.test_osutils',
1717
'bzrlib.tests.test_patch',
1718
'bzrlib.tests.test_patches',
976
1719
'bzrlib.tests.test_permissions',
977
1720
'bzrlib.tests.test_plugins',
978
1721
'bzrlib.tests.test_progress',
979
1722
'bzrlib.tests.test_reconcile',
1723
'bzrlib.tests.test_registry',
980
1724
'bzrlib.tests.test_repository',
1725
'bzrlib.tests.test_revert',
981
1726
'bzrlib.tests.test_revision',
982
1727
'bzrlib.tests.test_revisionnamespaces',
983
'bzrlib.tests.test_revprops',
1728
'bzrlib.tests.test_revisiontree',
984
1729
'bzrlib.tests.test_rio',
985
1730
'bzrlib.tests.test_sampler',
986
1731
'bzrlib.tests.test_selftest',
987
1732
'bzrlib.tests.test_setup',
988
1733
'bzrlib.tests.test_sftp_transport',
989
1734
'bzrlib.tests.test_smart_add',
1735
'bzrlib.tests.test_smart_transport',
990
1736
'bzrlib.tests.test_source',
1737
'bzrlib.tests.test_status',
991
1738
'bzrlib.tests.test_store',
992
1739
'bzrlib.tests.test_symbol_versioning',
993
1740
'bzrlib.tests.test_testament',
1741
'bzrlib.tests.test_textfile',
1742
'bzrlib.tests.test_textmerge',
994
1743
'bzrlib.tests.test_trace',
995
1744
'bzrlib.tests.test_transactions',
996
1745
'bzrlib.tests.test_transform',
997
1746
'bzrlib.tests.test_transport',
1747
'bzrlib.tests.test_tree',
1748
'bzrlib.tests.test_treebuilder',
998
1749
'bzrlib.tests.test_tsort',
1750
'bzrlib.tests.test_tuned_gzip',
999
1751
'bzrlib.tests.test_ui',
1000
1752
'bzrlib.tests.test_upgrade',
1753
'bzrlib.tests.test_urlutils',
1001
1754
'bzrlib.tests.test_versionedfile',
1755
'bzrlib.tests.test_version',
1756
'bzrlib.tests.test_version_info',
1002
1757
'bzrlib.tests.test_weave',
1003
1758
'bzrlib.tests.test_whitebox',
1004
1759
'bzrlib.tests.test_workingtree',
1005
1760
'bzrlib.tests.test_xml',
1007
1762
test_transport_implementations = [
1008
'bzrlib.tests.test_transport_implementations']
1010
TestCase.BZRPATH = osutils.pathjoin(
1011
osutils.realpath(osutils.dirname(bzrlib.__path__[0])), 'bzr')
1012
print '%10s: %s' % ('bzr', osutils.realpath(sys.argv[0]))
1013
print '%10s: %s' % ('bzrlib', bzrlib.__path__[0])
1016
# python2.4's TestLoader.loadTestsFromNames gives very poor
1017
# errors if it fails to load a named module - no indication of what's
1018
# actually wrong, just "no such module". We should probably override that
1019
# class, but for the moment just load them ourselves. (mbp 20051202)
1020
loader = TestLoader()
1763
'bzrlib.tests.test_transport_implementations',
1764
'bzrlib.tests.test_read_bundle',
1766
suite = TestUtil.TestSuite()
1767
loader = TestUtil.TestLoader()
1768
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1021
1769
from bzrlib.transport import TransportTestProviderAdapter
1022
1770
adapter = TransportTestProviderAdapter()
1023
1771
adapt_modules(test_transport_implementations, adapter, loader, suite)
1024
for mod_name in testmod_names:
1025
mod = _load_module_by_name(mod_name)
1026
suite.addTest(loader.loadTestsFromModule(mod))
1027
1772
for package in packages_to_test():
1028
1773
suite.addTest(package.test_suite())
1029
1774
for m in MODULES_TO_TEST:
1030
1775
suite.addTest(loader.loadTestsFromModule(m))
1031
for m in (MODULES_TO_DOCTEST):
1032
suite.addTest(DocTestSuite(m))
1776
for m in MODULES_TO_DOCTEST:
1778
suite.addTest(doctest.DocTestSuite(m))
1779
except ValueError, e:
1780
print '**failed to get doctest for: %s\n%s' %(m,e)
1033
1782
for name, plugin in bzrlib.plugin.all_plugins().items():
1034
1783
if getattr(plugin, 'test_suite', None) is not None:
1035
1784
suite.addTest(plugin.test_suite())
added
removed
bzrlib/tests/__init__.py
