102
130
bzrlib.tests.bzrdir_implementations,
103
131
bzrlib.tests.interrepository_implementations,
104
132
bzrlib.tests.interversionedfile_implementations,
133
bzrlib.tests.intertree_implementations,
134
bzrlib.tests.per_lock,
105
135
bzrlib.tests.repository_implementations,
106
136
bzrlib.tests.revisionstore_implementations,
137
bzrlib.tests.tree_implementations,
107
138
bzrlib.tests.workingtree_implementations,
111
class _MyResult(unittest._TextTestResult):
112
"""Custom TestResult.
142
class ExtendedTestResult(unittest._TextTestResult):
143
"""Accepts, reports and accumulates the results of running tests.
114
Shows output in a different format, including displaying runtime for tests.
145
Compared to this unittest version this class adds support for profiling,
146
benchmarking, stopping as soon as a test fails, and skipping tests.
147
There are further-specialized subclasses for different types of display.
116
150
stop_early = False
118
def _elapsedTime(self):
119
return "%5dms" % (1000 * (time.time() - self._start_time))
152
def __init__(self, stream, descriptions, verbosity,
155
use_numbered_dirs=False,
157
"""Construct new TestResult.
159
:param bench_history: Optionally, a writable file object to accumulate
162
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
163
if bench_history is not None:
164
from bzrlib.version import _get_bzr_source_tree
165
src_tree = _get_bzr_source_tree()
168
revision_id = src_tree.get_parent_ids()[0]
170
# XXX: if this is a brand new tree, do the same as if there
174
# XXX: If there's no branch, what should we do?
176
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
177
self._bench_history = bench_history
178
self.ui = ui.ui_factory
179
self.num_tests = num_tests
181
self.failure_count = 0
182
self.known_failure_count = 0
184
self.unsupported = {}
186
self.use_numbered_dirs = use_numbered_dirs
187
self._overall_start_time = time.time()
189
def extractBenchmarkTime(self, testCase):
190
"""Add a benchmark time for the current test case."""
191
self._benchmarkTime = getattr(testCase, "_benchtime", None)
193
def _elapsedTestTimeString(self):
194
"""Return a time string for the overall time the current test has taken."""
195
return self._formatTime(time.time() - self._start_time)
197
def _testTimeString(self):
198
if self._benchmarkTime is not None:
200
self._formatTime(self._benchmarkTime),
201
self._elapsedTestTimeString())
203
return " %s" % self._elapsedTestTimeString()
205
def _formatTime(self, seconds):
206
"""Format seconds as milliseconds with leading spaces."""
207
# some benchmarks can take thousands of seconds to run, so we need 8
209
return "%8dms" % (1000 * seconds)
211
def _shortened_test_description(self, test):
213
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
121
216
def startTest(self, test):
122
217
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)
218
self.report_test_start(test)
219
test.number = self.count
220
self._recordTestStartTime()
222
def _recordTestStartTime(self):
223
"""Record that a test has started."""
145
224
self._start_time = time.time()
226
def _cleanupLogFile(self, test):
227
# We can only do this if we have one of our TestCases, not if
229
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
230
if setKeepLogfile is not None:
147
233
def addError(self, test, err):
234
self.extractBenchmarkTime(test)
235
self._cleanupLogFile(test)
148
236
if isinstance(err[1], TestSkipped):
149
return self.addSkipped(test, err)
237
return self.addSkipped(test, err)
238
elif isinstance(err[1], UnavailableFeature):
239
return self.addNotSupported(test, err[1].args[0])
150
240
unittest.TestResult.addError(self, test, err)
152
self.stream.writeln("ERROR %s" % self._elapsedTime())
154
self.stream.write('E')
241
self.error_count += 1
242
self.report_error(test, err)
156
243
if self.stop_early:
159
246
def addFailure(self, test, err):
247
self._cleanupLogFile(test)
248
self.extractBenchmarkTime(test)
249
if isinstance(err[1], KnownFailure):
250
return self.addKnownFailure(test, err)
160
251
unittest.TestResult.addFailure(self, test, err)
162
self.stream.writeln(" FAIL %s" % self._elapsedTime())
164
self.stream.write('F')
252
self.failure_count += 1
253
self.report_failure(test, err)
166
254
if self.stop_early:
257
def addKnownFailure(self, test, err):
258
self.known_failure_count += 1
259
self.report_known_failure(test, err)
261
def addNotSupported(self, test, feature):
262
self.unsupported.setdefault(str(feature), 0)
263
self.unsupported[str(feature)] += 1
264
self.report_unsupported(test, feature)
169
266
def addSuccess(self, test):
171
self.stream.writeln(' OK %s' % self._elapsedTime())
173
self.stream.write('~')
267
self.extractBenchmarkTime(test)
268
if self._bench_history is not None:
269
if self._benchmarkTime is not None:
270
self._bench_history.write("%s %s\n" % (
271
self._formatTime(self._benchmarkTime),
273
self.report_success(test)
175
274
unittest.TestResult.addSuccess(self, test)
177
276
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')
277
self.report_skip(test, skip_excinfo)
184
278
# seems best to treat this as success from point-of-view of unittest
185
279
# -- it actually does nothing so it barely matters :)
186
unittest.TestResult.addSuccess(self, test)
282
except KeyboardInterrupt:
285
self.addError(test, test.__exc_info())
287
unittest.TestResult.addSuccess(self, test)
188
289
def printErrorList(self, flavour, errors):
189
290
for test, err in errors:
190
291
self.stream.writeln(self.separator1)
191
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
292
self.stream.write("%s: " % flavour)
293
if self.use_numbered_dirs:
294
self.stream.write('#%d ' % test.number)
295
self.stream.writeln(self.getDescription(test))
192
296
if getattr(test, '_get_log', None) is not None:
193
297
print >>self.stream
194
298
print >>self.stream, \
199
303
self.stream.writeln(self.separator2)
200
304
self.stream.writeln("%s" % err)
203
class TextTestRunner(unittest.TextTestRunner):
309
def report_cleaning_up(self):
312
def report_success(self, test):
316
class TextTestResult(ExtendedTestResult):
317
"""Displays progress and results of tests in text form"""
319
def __init__(self, stream, descriptions, verbosity,
323
use_numbered_dirs=False,
325
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
326
bench_history, num_tests, use_numbered_dirs)
328
self.pb = self.ui.nested_progress_bar()
329
self._supplied_pb = False
332
self._supplied_pb = True
333
self.pb.show_pct = False
334
self.pb.show_spinner = False
335
self.pb.show_eta = False,
336
self.pb.show_count = False
337
self.pb.show_bar = False
339
def report_starting(self):
340
self.pb.update('[test 0/%d] starting...' % (self.num_tests))
342
def _progress_prefix_text(self):
343
a = '[%d' % self.count
344
if self.num_tests is not None:
345
a +='/%d' % self.num_tests
346
a += ' in %ds' % (time.time() - self._overall_start_time)
348
a += ', %d errors' % self.error_count
349
if self.failure_count:
350
a += ', %d failed' % self.failure_count
351
if self.known_failure_count:
352
a += ', %d known failures' % self.known_failure_count
354
a += ', %d skipped' % self.skip_count
356
a += ', %d missing features' % len(self.unsupported)
360
def report_test_start(self, test):
363
self._progress_prefix_text()
365
+ self._shortened_test_description(test))
367
def _test_description(self, test):
368
if self.use_numbered_dirs:
369
return '#%d %s' % (self.count,
370
self._shortened_test_description(test))
372
return self._shortened_test_description(test)
374
def report_error(self, test, err):
375
self.pb.note('ERROR: %s\n %s\n',
376
self._test_description(test),
380
def report_failure(self, test, err):
381
self.pb.note('FAIL: %s\n %s\n',
382
self._test_description(test),
386
def report_known_failure(self, test, err):
387
self.pb.note('XFAIL: %s\n%s\n',
388
self._test_description(test), err[1])
390
def report_skip(self, test, skip_excinfo):
393
# at the moment these are mostly not things we can fix
394
# and so they just produce stipple; use the verbose reporter
397
# show test and reason for skip
398
self.pb.note('SKIP: %s\n %s\n',
399
self._shortened_test_description(test),
402
# since the class name was left behind in the still-visible
404
self.pb.note('SKIP: %s', skip_excinfo[1])
406
def report_unsupported(self, test, feature):
407
"""test cannot be run because feature is missing."""
409
def report_cleaning_up(self):
410
self.pb.update('cleaning up...')
413
if not self._supplied_pb:
417
class VerboseTestResult(ExtendedTestResult):
418
"""Produce long output, with one line per test run plus times"""
420
def _ellipsize_to_right(self, a_string, final_width):
421
"""Truncate and pad a string, keeping the right hand side"""
422
if len(a_string) > final_width:
423
result = '...' + a_string[3-final_width:]
426
return result.ljust(final_width)
428
def report_starting(self):
429
self.stream.write('running %d tests...\n' % self.num_tests)
431
def report_test_start(self, test):
433
name = self._shortened_test_description(test)
434
# width needs space for 6 char status, plus 1 for slash, plus 2 10-char
435
# numbers, plus a trailing blank
436
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
437
if self.use_numbered_dirs:
438
self.stream.write('%5d ' % self.count)
439
self.stream.write(self._ellipsize_to_right(name,
440
osutils.terminal_width()-36))
442
self.stream.write(self._ellipsize_to_right(name,
443
osutils.terminal_width()-30))
446
def _error_summary(self, err):
448
if self.use_numbered_dirs:
450
return '%s%s' % (indent, err[1])
452
def report_error(self, test, err):
453
self.stream.writeln('ERROR %s\n%s'
454
% (self._testTimeString(),
455
self._error_summary(err)))
457
def report_failure(self, test, err):
458
self.stream.writeln(' FAIL %s\n%s'
459
% (self._testTimeString(),
460
self._error_summary(err)))
462
def report_known_failure(self, test, err):
463
self.stream.writeln('XFAIL %s\n%s'
464
% (self._testTimeString(),
465
self._error_summary(err)))
467
def report_success(self, test):
468
self.stream.writeln(' OK %s' % self._testTimeString())
469
for bench_called, stats in getattr(test, '_benchcalls', []):
470
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
471
stats.pprint(file=self.stream)
472
# flush the stream so that we get smooth output. This verbose mode is
473
# used to show the output in PQM.
476
def report_skip(self, test, skip_excinfo):
478
self.stream.writeln(' SKIP %s\n%s'
479
% (self._testTimeString(),
480
self._error_summary(skip_excinfo)))
482
def report_unsupported(self, test, feature):
483
"""test cannot be run because feature is missing."""
484
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
485
%(self._testTimeString(), feature))
489
class TextTestRunner(object):
204
490
stop_on_failure = False
206
def _makeResult(self):
207
result = _MyResult(self.stream, self.descriptions, self.verbosity)
498
use_numbered_dirs=False,
500
self.stream = unittest._WritelnDecorator(stream)
501
self.descriptions = descriptions
502
self.verbosity = verbosity
503
self.keep_output = keep_output
504
self._bench_history = bench_history
505
self.use_numbered_dirs = use_numbered_dirs
508
"Run the given test case or test suite."
509
startTime = time.time()
510
if self.verbosity == 1:
511
result_class = TextTestResult
512
elif self.verbosity >= 2:
513
result_class = VerboseTestResult
514
result = result_class(self.stream,
517
bench_history=self._bench_history,
518
num_tests=test.countTestCases(),
519
use_numbered_dirs=self.use_numbered_dirs,
208
521
result.stop_early = self.stop_on_failure
522
result.report_starting()
524
stopTime = time.time()
525
timeTaken = stopTime - startTime
527
self.stream.writeln(result.separator2)
528
run = result.testsRun
529
self.stream.writeln("Ran %d test%s in %.3fs" %
530
(run, run != 1 and "s" or "", timeTaken))
531
self.stream.writeln()
532
if not result.wasSuccessful():
533
self.stream.write("FAILED (")
534
failed, errored = map(len, (result.failures, result.errors))
536
self.stream.write("failures=%d" % failed)
538
if failed: self.stream.write(", ")
539
self.stream.write("errors=%d" % errored)
540
if result.known_failure_count:
541
if failed or errored: self.stream.write(", ")
542
self.stream.write("known_failure_count=%d" %
543
result.known_failure_count)
544
self.stream.writeln(")")
546
if result.known_failure_count:
547
self.stream.writeln("OK (known_failures=%d)" %
548
result.known_failure_count)
550
self.stream.writeln("OK")
551
if result.skip_count > 0:
552
skipped = result.skip_count
553
self.stream.writeln('%d test%s skipped' %
554
(skipped, skipped != 1 and "s" or ""))
555
if result.unsupported:
556
for feature, count in sorted(result.unsupported.items()):
557
self.stream.writeln("Missing feature '%s' skipped %d tests." %
559
result.report_cleaning_up()
560
# This is still a little bogus,
561
# but only a little. Folk not using our testrunner will
562
# have to delete their temp directories themselves.
563
test_root = TestCaseWithMemoryTransport.TEST_ROOT
564
if result.wasSuccessful() or not self.keep_output:
565
if test_root is not None:
566
# If LANG=C we probably have created some bogus paths
567
# which rmtree(unicode) will fail to delete
568
# so make sure we are using rmtree(str) to delete everything
569
# except on win32, where rmtree(str) will fail
570
# since it doesn't have the property of byte-stream paths
571
# (they are either ascii or mbcs)
572
if sys.platform == 'win32':
573
# make sure we are using the unicode win32 api
574
test_root = unicode(test_root)
576
test_root = test_root.encode(
577
sys.getfilesystemencoding())
578
_rmtree_temp_dir(test_root)
580
note("Failed tests working directories are in '%s'\n", test_root)
581
TestCaseWithMemoryTransport.TEST_ROOT = None
225
599
class TestSkipped(Exception):
226
600
"""Indicates that a test was intentionally skipped, rather than failing."""
603
class KnownFailure(AssertionError):
604
"""Indicates that a test failed in a precisely expected manner.
606
Such failures dont block the whole test suite from passing because they are
607
indicators of partially completed code or of future work. We have an
608
explicit error for them so that we can ensure that they are always visible:
609
KnownFailures are always shown in the output of bzr selftest.
613
class UnavailableFeature(Exception):
614
"""A feature required for this test was not available.
616
The feature should be used to construct the exception.
230
620
class CommandFailed(Exception):
624
class StringIOWrapper(object):
625
"""A wrapper around cStringIO which just adds an encoding attribute.
627
Internally we can check sys.stdout to see what the output encoding
628
should be. However, cStringIO has no encoding attribute that we can
629
set. So we wrap it instead.
634
def __init__(self, s=None):
636
self.__dict__['_cstring'] = StringIO(s)
638
self.__dict__['_cstring'] = StringIO()
640
def __getattr__(self, name, getattr=getattr):
641
return getattr(self.__dict__['_cstring'], name)
643
def __setattr__(self, name, val):
644
if name == 'encoding':
645
self.__dict__['encoding'] = val
647
return setattr(self._cstring, name, val)
650
class TestUIFactory(ui.CLIUIFactory):
651
"""A UI Factory for testing.
653
Hide the progress bar but emit note()s.
655
Allows get_password to be tested without real tty attached.
662
super(TestUIFactory, self).__init__()
663
if stdin is not None:
664
# We use a StringIOWrapper to be able to test various
665
# encodings, but the user is still responsible to
666
# encode the string and to set the encoding attribute
667
# of StringIOWrapper.
668
self.stdin = StringIOWrapper(stdin)
670
self.stdout = sys.stdout
674
self.stderr = sys.stderr
679
"""See progress.ProgressBar.clear()."""
681
def clear_term(self):
682
"""See progress.ProgressBar.clear_term()."""
684
def clear_term(self):
685
"""See progress.ProgressBar.clear_term()."""
688
"""See progress.ProgressBar.finished()."""
690
def note(self, fmt_string, *args, **kwargs):
691
"""See progress.ProgressBar.note()."""
692
self.stdout.write((fmt_string + "\n") % args)
694
def progress_bar(self):
697
def nested_progress_bar(self):
700
def update(self, message, count=None, total=None):
701
"""See progress.ProgressBar.update()."""
703
def get_non_echoed_password(self, prompt):
704
"""Get password from stdin without trying to handle the echo mode"""
706
self.stdout.write(prompt)
707
password = self.stdin.readline()
710
if password[-1] == '\n':
711
password = password[:-1]
233
715
class TestCase(unittest.TestCase):
234
716
"""Base class for bzr unit tests.
341
924
def assertIsInstance(self, obj, kls):
342
925
"""Fail if obj is not an instance of kls"""
343
926
if not isinstance(obj, kls):
344
self.fail("%r is not an instance of %s" % (obj, kls))
927
self.fail("%r is an instance of %s rather than %s" % (
928
obj, obj.__class__, kls))
930
def expectFailure(self, reason, assertion, *args, **kwargs):
931
"""Invoke a test, expecting it to fail for the given reason.
933
This is for assertions that ought to succeed, but currently fail.
934
(The failure is *expected* but not *wanted*.) Please be very precise
935
about the failure you're expecting. If a new bug is introduced,
936
AssertionError should be raised, not KnownFailure.
938
Frequently, expectFailure should be followed by an opposite assertion.
941
Intended to be used with a callable that raises AssertionError as the
942
'assertion' parameter. args and kwargs are passed to the 'assertion'.
944
Raises KnownFailure if the test fails. Raises AssertionError if the
949
self.expectFailure('Math is broken', self.assertNotEqual, 54,
951
self.assertEqual(42, dynamic_val)
953
This means that a dynamic_val of 54 will cause the test to raise
954
a KnownFailure. Once math is fixed and the expectFailure is removed,
955
only a dynamic_val of 42 will allow the test to pass. Anything other
956
than 54 or 42 will cause an AssertionError.
959
assertion(*args, **kwargs)
960
except AssertionError:
961
raise KnownFailure(reason)
963
self.fail('Unexpected success. Should have failed: %s' % reason)
965
def _capture_warnings(self, a_callable, *args, **kwargs):
966
"""A helper for callDeprecated and applyDeprecated.
968
:param a_callable: A callable to call.
969
:param args: The positional arguments for the callable
970
:param kwargs: The keyword arguments for the callable
971
:return: A tuple (warnings, result). result is the result of calling
972
a_callable(*args, **kwargs).
975
def capture_warnings(msg, cls=None, stacklevel=None):
976
# we've hooked into a deprecation specific callpath,
977
# only deprecations should getting sent via it.
978
self.assertEqual(cls, DeprecationWarning)
979
local_warnings.append(msg)
980
original_warning_method = symbol_versioning.warn
981
symbol_versioning.set_warning_method(capture_warnings)
983
result = a_callable(*args, **kwargs)
985
symbol_versioning.set_warning_method(original_warning_method)
986
return (local_warnings, result)
988
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
989
"""Call a deprecated callable without warning the user.
991
:param deprecation_format: The deprecation format that the callable
992
should have been deprecated with. This is the same type as the
993
parameter to deprecated_method/deprecated_function. If the
994
callable is not deprecated with this format, an assertion error
996
:param a_callable: A callable to call. This may be a bound method or
997
a regular function. It will be called with *args and **kwargs.
998
:param args: The positional arguments for the callable
999
:param kwargs: The keyword arguments for the callable
1000
:return: The result of a_callable(*args, **kwargs)
1002
call_warnings, result = self._capture_warnings(a_callable,
1004
expected_first_warning = symbol_versioning.deprecation_string(
1005
a_callable, deprecation_format)
1006
if len(call_warnings) == 0:
1007
self.fail("No deprecation warning generated by call to %s" %
1009
self.assertEqual(expected_first_warning, call_warnings[0])
1012
def callDeprecated(self, expected, callable, *args, **kwargs):
1013
"""Assert that a callable is deprecated in a particular way.
1015
This is a very precise test for unusual requirements. The
1016
applyDeprecated helper function is probably more suited for most tests
1017
as it allows you to simply specify the deprecation format being used
1018
and will ensure that that is issued for the function being called.
1020
:param expected: a list of the deprecation warnings expected, in order
1021
:param callable: The callable to call
1022
:param args: The positional arguments for the callable
1023
:param kwargs: The keyword arguments for the callable
1025
call_warnings, result = self._capture_warnings(callable,
1027
self.assertEqual(expected, call_warnings)
346
1030
def _startLogFile(self):
347
1031
"""Send bzr and test log messages to a temporary file.
381
1070
def _cleanEnvironment(self):
1072
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
383
1073
'HOME': os.getcwd(),
384
'APPDATA': os.getcwd(),
1074
'APPDATA': None, # bzr now use Win32 API and don't rely on APPDATA
1076
'BZREMAIL': None, # may still be present in the environment
1078
'BZR_PROGRESS_BAR': None,
1082
'https_proxy': None,
1083
'HTTPS_PROXY': None,
1088
# Nobody cares about these ones AFAIK. So far at
1089
# least. If you do (care), please update this comment
388
1094
self.__old_env = {}
389
1095
self.addCleanup(self._restoreEnvironment)
390
1096
for name, value in new_env.iteritems():
391
1097
self._captureVar(name, value)
394
1099
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
1100
"""Set an environment variable, and reset it when finished."""
1101
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
411
1103
def _restoreEnvironment(self):
412
1104
for name, value in self.__old_env.iteritems():
413
self._restoreVar(name, value)
1105
osutils.set_or_unset_env(name, value)
1107
def _restoreHooks(self):
1108
for klass, hooks in self._preserved_hooks.items():
1109
setattr(klass, 'hooks', hooks)
1111
def knownFailure(self, reason):
1112
"""This test has failed for some known reason."""
1113
raise KnownFailure(reason)
1115
def run(self, result=None):
1116
if result is None: result = self.defaultTestResult()
1117
for feature in getattr(self, '_test_needs_features', []):
1118
if not feature.available():
1119
result.startTest(self)
1120
if getattr(result, 'addNotSupported', None):
1121
result.addNotSupported(self, feature)
1123
result.addSuccess(self)
1124
result.stopTest(self)
1126
return unittest.TestCase.run(self, result)
415
1128
def tearDown(self):
416
1129
self._runCleanups()
417
1130
unittest.TestCase.tearDown(self)
1132
def time(self, callable, *args, **kwargs):
1133
"""Run callable and accrue the time it takes to the benchmark time.
1135
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1136
this will cause lsprofile statistics to be gathered and stored in
1139
if self._benchtime is None:
1143
if not self._gather_lsprof_in_benchmarks:
1144
return callable(*args, **kwargs)
1146
# record this benchmark
1147
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
1149
self._benchcalls.append(((callable, args, kwargs), stats))
1152
self._benchtime += time.time() - start
419
1154
def _runCleanups(self):
420
1155
"""Run registered cleanup functions.
496
1289
This sends the stdout/stderr results into the test's log,
497
1290
where it may be useful for debugging. See also run_captured.
1292
:param stdin: A string to be used as stdin for the command.
1293
:param retcode: The status code the command should return
1294
:param working_dir: The directory to run the command in
499
1296
retcode = kwargs.pop('retcode', 0)
500
return self.run_bzr_captured(args, retcode)
1297
encoding = kwargs.pop('encoding', None)
1298
stdin = kwargs.pop('stdin', None)
1299
working_dir = kwargs.pop('working_dir', None)
1300
return self.run_bzr_captured(args, retcode=retcode, encoding=encoding,
1301
stdin=stdin, working_dir=working_dir)
1303
def run_bzr_decode(self, *args, **kwargs):
1304
if 'encoding' in kwargs:
1305
encoding = kwargs['encoding']
1307
encoding = bzrlib.user_encoding
1308
return self.run_bzr(*args, **kwargs)[0].decode(encoding)
1310
def run_bzr_error(self, error_regexes, *args, **kwargs):
1311
"""Run bzr, and check that stderr contains the supplied regexes
1313
:param error_regexes: Sequence of regular expressions which
1314
must each be found in the error output. The relative ordering
1316
:param args: command-line arguments for bzr
1317
:param kwargs: Keyword arguments which are interpreted by run_bzr
1318
This function changes the default value of retcode to be 3,
1319
since in most cases this is run when you expect bzr to fail.
1320
:return: (out, err) The actual output of running the command (in case you
1321
want to do more inspection)
1324
# Make sure that commit is failing because there is nothing to do
1325
self.run_bzr_error(['no changes to commit'],
1326
'commit', '-m', 'my commit comment')
1327
# Make sure --strict is handling an unknown file, rather than
1328
# giving us the 'nothing to do' error
1329
self.build_tree(['unknown'])
1330
self.run_bzr_error(['Commit refused because there are unknown files'],
1331
'commit', '--strict', '-m', 'my commit comment')
1333
kwargs.setdefault('retcode', 3)
1334
out, err = self.run_bzr(*args, **kwargs)
1335
for regex in error_regexes:
1336
self.assertContainsRe(err, regex)
1339
def run_bzr_subprocess(self, *args, **kwargs):
1340
"""Run bzr in a subprocess for testing.
1342
This starts a new Python interpreter and runs bzr in there.
1343
This should only be used for tests that have a justifiable need for
1344
this isolation: e.g. they are testing startup time, or signal
1345
handling, or early startup code, etc. Subprocess code can't be
1346
profiled or debugged so easily.
1348
:param retcode: The status code that is expected. Defaults to 0. If
1349
None is supplied, the status code is not checked.
1350
:param env_changes: A dictionary which lists changes to environment
1351
variables. A value of None will unset the env variable.
1352
The values must be strings. The change will only occur in the
1353
child, so you don't need to fix the environment after running.
1354
:param universal_newlines: Convert CRLF => LF
1355
:param allow_plugins: By default the subprocess is run with
1356
--no-plugins to ensure test reproducibility. Also, it is possible
1357
for system-wide plugins to create unexpected output on stderr,
1358
which can cause unnecessary test failures.
1360
env_changes = kwargs.get('env_changes', {})
1361
working_dir = kwargs.get('working_dir', None)
1362
allow_plugins = kwargs.get('allow_plugins', False)
1363
process = self.start_bzr_subprocess(args, env_changes=env_changes,
1364
working_dir=working_dir,
1365
allow_plugins=allow_plugins)
1366
# We distinguish between retcode=None and retcode not passed.
1367
supplied_retcode = kwargs.get('retcode', 0)
1368
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1369
universal_newlines=kwargs.get('universal_newlines', False),
1372
def start_bzr_subprocess(self, process_args, env_changes=None,
1373
skip_if_plan_to_signal=False,
1375
allow_plugins=False):
1376
"""Start bzr in a subprocess for testing.
1378
This starts a new Python interpreter and runs bzr in there.
1379
This should only be used for tests that have a justifiable need for
1380
this isolation: e.g. they are testing startup time, or signal
1381
handling, or early startup code, etc. Subprocess code can't be
1382
profiled or debugged so easily.
1384
:param process_args: a list of arguments to pass to the bzr executable,
1385
for example `['--version']`.
1386
:param env_changes: A dictionary which lists changes to environment
1387
variables. A value of None will unset the env variable.
1388
The values must be strings. The change will only occur in the
1389
child, so you don't need to fix the environment after running.
1390
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1392
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1394
:returns: Popen object for the started process.
1396
if skip_if_plan_to_signal:
1397
if not getattr(os, 'kill', None):
1398
raise TestSkipped("os.kill not available.")
1400
if env_changes is None:
1404
def cleanup_environment():
1405
for env_var, value in env_changes.iteritems():
1406
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1408
def restore_environment():
1409
for env_var, value in old_env.iteritems():
1410
osutils.set_or_unset_env(env_var, value)
1412
bzr_path = self.get_bzr_path()
1415
if working_dir is not None:
1416
cwd = osutils.getcwd()
1417
os.chdir(working_dir)
1420
# win32 subprocess doesn't support preexec_fn
1421
# so we will avoid using it on all platforms, just to
1422
# make sure the code path is used, and we don't break on win32
1423
cleanup_environment()
1424
command = [sys.executable, bzr_path]
1425
if not allow_plugins:
1426
command.append('--no-plugins')
1427
command.extend(process_args)
1428
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1430
restore_environment()
1436
def _popen(self, *args, **kwargs):
1437
"""Place a call to Popen.
1439
Allows tests to override this method to intercept the calls made to
1440
Popen for introspection.
1442
return Popen(*args, **kwargs)
1444
def get_bzr_path(self):
1445
"""Return the path of the 'bzr' executable for this test suite."""
1446
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1447
if not os.path.isfile(bzr_path):
1448
# We are probably installed. Assume sys.argv is the right file
1449
bzr_path = sys.argv[0]
1452
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1453
universal_newlines=False, process_args=None):
1454
"""Finish the execution of process.
1456
:param process: the Popen object returned from start_bzr_subprocess.
1457
:param retcode: The status code that is expected. Defaults to 0. If
1458
None is supplied, the status code is not checked.
1459
:param send_signal: an optional signal to send to the process.
1460
:param universal_newlines: Convert CRLF => LF
1461
:returns: (stdout, stderr)
1463
if send_signal is not None:
1464
os.kill(process.pid, send_signal)
1465
out, err = process.communicate()
1467
if universal_newlines:
1468
out = out.replace('\r\n', '\n')
1469
err = err.replace('\r\n', '\n')
1471
if retcode is not None and retcode != process.returncode:
1472
if process_args is None:
1473
process_args = "(unknown args)"
1474
mutter('Output of bzr %s:\n%s', process_args, out)
1475
mutter('Error for bzr %s:\n%s', process_args, err)
1476
self.fail('Command bzr %s failed with retcode %s != %s'
1477
% (process_args, retcode, process.returncode))
502
1480
def check_inventory_shape(self, inv, shape):
503
1481
"""Compare an inventory to a list of expected names.
562
1541
base_rev = common_ancestor(branch_from.last_revision(),
563
1542
wt_to.branch.last_revision(),
564
1543
wt_to.branch.repository)
565
merge_inner(wt_to.branch, branch_from.basis_tree(),
1544
merge_inner(wt_to.branch, branch_from.basis_tree(),
566
1545
wt_to.branch.repository.revision_tree(base_rev),
567
1546
this_tree=wt_to)
568
wt_to.add_pending_merge(branch_from.last_revision())
1547
wt_to.add_parent_tree_id(branch_from.last_revision())
1549
def reduceLockdirTimeout(self):
1550
"""Reduce the default lock timeout for the duration of the test, so that
1551
if LockContention occurs during a test, it does so quickly.
1553
Tests that expect to provoke LockContention errors should call this.
1555
orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
1557
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
1558
self.addCleanup(resetTimeout)
1559
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
571
1561
BzrTestBase = TestCase
1564
class TestCaseWithMemoryTransport(TestCase):
1565
"""Common test class for tests that do not need disk resources.
1567
Tests that need disk resources should derive from TestCaseWithTransport.
1569
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1571
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1572
a directory which does not exist. This serves to help ensure test isolation
1573
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1574
must exist. However, TestCaseWithMemoryTransport does not offer local
1575
file defaults for the transport in tests, nor does it obey the command line
1576
override, so tests that accidentally write to the common directory should
1584
def __init__(self, methodName='runTest'):
1585
# allow test parameterisation after test construction and before test
1586
# execution. Variables that the parameteriser sets need to be
1587
# ones that are not set by setUp, or setUp will trash them.
1588
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1589
self.vfs_transport_factory = default_transport
1590
self.transport_server = None
1591
self.transport_readonly_server = None
1592
self.__vfs_server = None
1594
def get_transport(self):
1595
"""Return a writeable transport for the test scratch space"""
1596
t = get_transport(self.get_url())
1597
self.assertFalse(t.is_readonly())
1600
def get_readonly_transport(self):
1601
"""Return a readonly transport for the test scratch space
1603
This can be used to test that operations which should only need
1604
readonly access in fact do not try to write.
1606
t = get_transport(self.get_readonly_url())
1607
self.assertTrue(t.is_readonly())
1610
def create_transport_readonly_server(self):
1611
"""Create a transport server from class defined at init.
1613
This is mostly a hook for daughter classes.
1615
return self.transport_readonly_server()
1617
def get_readonly_server(self):
1618
"""Get the server instance for the readonly transport
1620
This is useful for some tests with specific servers to do diagnostics.
1622
if self.__readonly_server is None:
1623
if self.transport_readonly_server is None:
1624
# readonly decorator requested
1625
# bring up the server
1626
self.__readonly_server = ReadonlyServer()
1627
self.__readonly_server.setUp(self.get_vfs_only_server())
1629
self.__readonly_server = self.create_transport_readonly_server()
1630
self.__readonly_server.setUp(self.get_vfs_only_server())
1631
self.addCleanup(self.__readonly_server.tearDown)
1632
return self.__readonly_server
1634
def get_readonly_url(self, relpath=None):
1635
"""Get a URL for the readonly transport.
1637
This will either be backed by '.' or a decorator to the transport
1638
used by self.get_url()
1639
relpath provides for clients to get a path relative to the base url.
1640
These should only be downwards relative, not upwards.
1642
base = self.get_readonly_server().get_url()
1643
if relpath is not None:
1644
if not base.endswith('/'):
1646
base = base + relpath
1649
def get_vfs_only_server(self):
1650
"""Get the vfs only read/write server instance.
1652
This is useful for some tests with specific servers that need
1655
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1656
is no means to override it.
1658
if self.__vfs_server is None:
1659
self.__vfs_server = MemoryServer()
1660
self.__vfs_server.setUp()
1661
self.addCleanup(self.__vfs_server.tearDown)
1662
return self.__vfs_server
1664
def get_server(self):
1665
"""Get the read/write server instance.
1667
This is useful for some tests with specific servers that need
1670
This is built from the self.transport_server factory. If that is None,
1671
then the self.get_vfs_server is returned.
1673
if self.__server is None:
1674
if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
1675
return self.get_vfs_only_server()
1677
# bring up a decorated means of access to the vfs only server.
1678
self.__server = self.transport_server()
1680
self.__server.setUp(self.get_vfs_only_server())
1681
except TypeError, e:
1682
# This should never happen; the try:Except here is to assist
1683
# developers having to update code rather than seeing an
1684
# uninformative TypeError.
1685
raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
1686
self.addCleanup(self.__server.tearDown)
1687
return self.__server
1689
def _adjust_url(self, base, relpath):
1690
"""Get a URL (or maybe a path) for the readwrite transport.
1692
This will either be backed by '.' or to an equivalent non-file based
1694
relpath provides for clients to get a path relative to the base url.
1695
These should only be downwards relative, not upwards.
1697
if relpath is not None and relpath != '.':
1698
if not base.endswith('/'):
1700
# XXX: Really base should be a url; we did after all call
1701
# get_url()! But sometimes it's just a path (from
1702
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1703
# to a non-escaped local path.
1704
if base.startswith('./') or base.startswith('/'):
1707
base += urlutils.escape(relpath)
1710
def get_url(self, relpath=None):
1711
"""Get a URL (or maybe a path) for the readwrite transport.
1713
This will either be backed by '.' or to an equivalent non-file based
1715
relpath provides for clients to get a path relative to the base url.
1716
These should only be downwards relative, not upwards.
1718
base = self.get_server().get_url()
1719
return self._adjust_url(base, relpath)
1721
def get_vfs_only_url(self, relpath=None):
1722
"""Get a URL (or maybe a path for the plain old vfs transport.
1724
This will never be a smart protocol.
1725
:param relpath: provides for clients to get a path relative to the base
1726
url. These should only be downwards relative, not upwards.
1728
base = self.get_vfs_only_server().get_url()
1729
return self._adjust_url(base, relpath)
1731
def _make_test_root(self):
1732
if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1736
root = u'test%04d.tmp' % i
1740
if e.errno == errno.EEXIST:
1745
# successfully created
1746
TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1748
# make a fake bzr directory there to prevent any tests propagating
1749
# up onto the source directory's real branch
1750
bzrdir.BzrDir.create_standalone_workingtree(
1751
TestCaseWithMemoryTransport.TEST_ROOT)
1753
def makeAndChdirToTestDir(self):
1754
"""Create a temporary directories for this one test.
1756
This must set self.test_home_dir and self.test_dir and chdir to
1759
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1761
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1762
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1763
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1765
def make_branch(self, relpath, format=None):
1766
"""Create a branch on the transport at relpath."""
1767
repo = self.make_repository(relpath, format=format)
1768
return repo.bzrdir.create_branch()
1770
def make_bzrdir(self, relpath, format=None):
1772
# might be a relative or absolute path
1773
maybe_a_url = self.get_url(relpath)
1774
segments = maybe_a_url.rsplit('/', 1)
1775
t = get_transport(maybe_a_url)
1776
if len(segments) > 1 and segments[-1] not in ('', '.'):
1779
except errors.FileExists:
1783
if isinstance(format, basestring):
1784
format = bzrdir.format_registry.make_bzrdir(format)
1785
return format.initialize_on_transport(t)
1786
except errors.UninitializableFormat:
1787
raise TestSkipped("Format %s is not initializable." % format)
1789
def make_repository(self, relpath, shared=False, format=None):
1790
"""Create a repository on our default transport at relpath."""
1791
made_control = self.make_bzrdir(relpath, format=format)
1792
return made_control.create_repository(shared=shared)
1794
def make_branch_and_memory_tree(self, relpath, format=None):
1795
"""Create a branch on the default transport and a MemoryTree for it."""
1796
b = self.make_branch(relpath, format=format)
1797
return memorytree.MemoryTree.create_on_branch(b)
1799
def overrideEnvironmentForTesting(self):
1800
os.environ['HOME'] = self.test_home_dir
1801
os.environ['BZR_HOME'] = self.test_home_dir
1804
super(TestCaseWithMemoryTransport, self).setUp()
1805
self._make_test_root()
1806
_currentdir = os.getcwdu()
1807
def _leaveDirectory():
1808
os.chdir(_currentdir)
1809
self.addCleanup(_leaveDirectory)
1810
self.makeAndChdirToTestDir()
1811
self.overrideEnvironmentForTesting()
1812
self.__readonly_server = None
1813
self.__server = None
1814
self.reduceLockdirTimeout()
574
class TestCaseInTempDir(TestCase):
1817
class TestCaseInTempDir(TestCaseWithMemoryTransport):
575
1818
"""Derived class that runs a test within a temporary directory.
577
1820
This is useful for tests that need to create a branch, etc.
710
1957
readwrite one must both define get_url() as resolving to os.getcwd().
713
def __init__(self, methodName='testMethod'):
714
super(TestCaseWithTransport, self).__init__(methodName)
715
self.__readonly_server = None
717
self.transport_server = default_transport
718
self.transport_readonly_server = None
720
def get_readonly_url(self, relpath=None):
721
"""Get a URL for the readonly transport.
723
This will either be backed by '.' or a decorator to the transport
724
used by self.get_url()
725
relpath provides for clients to get a path relative to the base url.
726
These should only be downwards relative, not upwards.
728
base = self.get_readonly_server().get_url()
729
if relpath is not None:
730
if not base.endswith('/'):
732
base = base + relpath
735
def get_readonly_server(self):
736
"""Get the server instance for the readonly transport
738
This is useful for some tests with specific servers to do diagnostics.
740
if self.__readonly_server is None:
741
if self.transport_readonly_server is None:
742
# readonly decorator requested
743
# bring up the server
745
self.__readonly_server = ReadonlyServer()
746
self.__readonly_server.setUp(self.__server)
748
self.__readonly_server = self.transport_readonly_server()
749
self.__readonly_server.setUp()
750
self.addCleanup(self.__readonly_server.tearDown)
751
return self.__readonly_server
753
def get_server(self):
754
"""Get the read/write server instance.
1960
def get_vfs_only_server(self):
1961
"""See TestCaseWithMemoryTransport.
756
1963
This is useful for some tests with specific servers that need
759
if self.__server is None:
760
self.__server = self.transport_server()
761
self.__server.setUp()
762
self.addCleanup(self.__server.tearDown)
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):
1966
if self.__vfs_server is None:
1967
self.__vfs_server = self.vfs_transport_factory()
1968
self.__vfs_server.setUp()
1969
self.addCleanup(self.__vfs_server.tearDown)
1970
return self.__vfs_server
1972
def make_branch_and_tree(self, relpath, format=None):
822
1973
"""Create a branch on the transport and a tree locally.
1975
If the transport is not a LocalTransport, the Tree can't be created on
1976
the transport. In that case if the vfs_transport_factory is
1977
LocalURLServer the working tree is created in the local
1978
directory backing the transport, and the returned tree's branch and
1979
repository will also be accessed locally. Otherwise a lightweight
1980
checkout is created and returned.
1982
:param format: The BzrDirFormat.
1983
:returns: the WorkingTree.
826
1985
# TODO: always use the local disk path for the working tree,
827
1986
# this obviously requires a format that supports branch references
828
1987
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
830
b = self.make_branch(relpath)
1989
b = self.make_branch(relpath, format=format)
832
1991
return b.bzrdir.create_workingtree()
833
1992
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))
1993
# We can only make working trees locally at the moment. If the
1994
# transport can't support them, then we keep the non-disk-backed
1995
# branch and create a local checkout.
1996
if self.vfs_transport_factory is LocalURLServer:
1997
# the branch is colocated on disk, we cannot create a checkout.
1998
# hopefully callers will expect this.
1999
local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
2000
return local_controldir.create_workingtree()
2002
return b.create_checkout(relpath, lightweight=True)
840
2004
def assertIsDirectory(self, relpath, transport):
841
2005
"""Assert that relpath within transport is a directory.
2060
def sort_suite_by_re(suite, pattern):
2063
filter_re = re.compile(pattern)
2064
for test in iter_suite_tests(suite):
2065
if filter_re.search(test.id()):
2069
return TestUtil.TestSuite(first + second)
883
2072
def run_suite(suite, name='test', verbose=False, pattern=".*",
884
2073
stop_on_failure=False, keep_output=False,
886
TestCaseInTempDir._TEST_NAME = name
2074
transport=None, lsprof_timed=None, bench_history=None,
2075
matching_tests_first=None,
2076
numbered_dirs=None):
2077
use_numbered_dirs = bool(numbered_dirs)
2079
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
2080
if numbered_dirs is not None:
2081
TestCaseInTempDir.use_numbered_dirs = use_numbered_dirs
891
2086
runner = TextTestRunner(stream=sys.stdout,
2088
verbosity=verbosity,
2089
keep_output=keep_output,
2090
bench_history=bench_history,
2091
use_numbered_dirs=use_numbered_dirs,
894
2093
runner.stop_on_failure=stop_on_failure
895
2094
if pattern != '.*':
896
suite = filter_suite_by_re(suite, pattern)
2095
if matching_tests_first:
2096
suite = sort_suite_by_re(suite, pattern)
2098
suite = filter_suite_by_re(suite, pattern)
897
2099
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
2100
return result.wasSuccessful()
914
2103
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
915
2104
keep_output=False,
2106
test_suite_factory=None,
2109
matching_tests_first=None,
2110
numbered_dirs=None):
917
2111
"""Run the whole test suite under the enhanced runner"""
2112
# XXX: Very ugly way to do this...
2113
# Disable warning about old formats because we don't want it to disturb
2114
# any blackbox tests.
2115
from bzrlib import repository
2116
repository._deprecation_warning_done = True
918
2118
global default_transport
919
2119
if transport is None:
920
2120
transport = default_transport
921
2121
old_transport = default_transport
922
2122
default_transport = transport
2124
if test_suite_factory is None:
2125
suite = test_suite()
2127
suite = test_suite_factory()
925
2128
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
926
2129
stop_on_failure=stop_on_failure, keep_output=keep_output,
2130
transport=transport,
2131
lsprof_timed=lsprof_timed,
2132
bench_history=bench_history,
2133
matching_tests_first=matching_tests_first,
2134
numbered_dirs=numbered_dirs)
929
2136
default_transport = old_transport
933
2139
def test_suite():
934
"""Build and return TestSuite for the whole program."""
935
from doctest import DocTestSuite
937
global MODULES_TO_DOCTEST
2140
"""Build and return TestSuite for the whole of bzrlib.
2142
This function can be replaced if you need to change the default test
2143
suite on a global basis, but it is not encouraged.
940
2146
'bzrlib.tests.test_ancestry',
941
2147
'bzrlib.tests.test_annotate',
942
2148
'bzrlib.tests.test_api',
2149
'bzrlib.tests.test_atomicfile',
943
2150
'bzrlib.tests.test_bad_files',
944
'bzrlib.tests.test_basis_inventory',
945
2151
'bzrlib.tests.test_branch',
2152
'bzrlib.tests.test_bundle',
946
2153
'bzrlib.tests.test_bzrdir',
947
'bzrlib.tests.test_command',
2154
'bzrlib.tests.test_cache_utf8',
2155
'bzrlib.tests.test_commands',
948
2156
'bzrlib.tests.test_commit',
949
2157
'bzrlib.tests.test_commit_merge',
950
2158
'bzrlib.tests.test_config',
951
2159
'bzrlib.tests.test_conflicts',
952
2160
'bzrlib.tests.test_decorators',
2161
'bzrlib.tests.test_delta',
953
2162
'bzrlib.tests.test_diff',
2163
'bzrlib.tests.test_dirstate',
954
2164
'bzrlib.tests.test_doc_generate',
955
2165
'bzrlib.tests.test_errors',
956
2166
'bzrlib.tests.test_escaped_store',
2167
'bzrlib.tests.test_extract',
957
2168
'bzrlib.tests.test_fetch',
2169
'bzrlib.tests.test_ftp_transport',
2170
'bzrlib.tests.test_generate_docs',
2171
'bzrlib.tests.test_generate_ids',
2172
'bzrlib.tests.test_globbing',
958
2173
'bzrlib.tests.test_gpg',
959
2174
'bzrlib.tests.test_graph',
960
2175
'bzrlib.tests.test_hashcache',
2176
'bzrlib.tests.test_help',
961
2177
'bzrlib.tests.test_http',
2178
'bzrlib.tests.test_http_response',
2179
'bzrlib.tests.test_https_ca_bundle',
962
2180
'bzrlib.tests.test_identitymap',
2181
'bzrlib.tests.test_ignores',
963
2182
'bzrlib.tests.test_inv',
964
2183
'bzrlib.tests.test_knit',
2184
'bzrlib.tests.test_lazy_import',
2185
'bzrlib.tests.test_lazy_regex',
965
2186
'bzrlib.tests.test_lockdir',
966
2187
'bzrlib.tests.test_lockable_files',
967
2188
'bzrlib.tests.test_log',
2189
'bzrlib.tests.test_memorytree',
968
2190
'bzrlib.tests.test_merge',
969
2191
'bzrlib.tests.test_merge3',
970
2192
'bzrlib.tests.test_merge_core',
2193
'bzrlib.tests.test_merge_directive',
971
2194
'bzrlib.tests.test_missing',
972
2195
'bzrlib.tests.test_msgeditor',
973
2196
'bzrlib.tests.test_nonascii',
974
2197
'bzrlib.tests.test_options',
975
2198
'bzrlib.tests.test_osutils',
2199
'bzrlib.tests.test_osutils_encodings',
2200
'bzrlib.tests.test_patch',
2201
'bzrlib.tests.test_patches',
976
2202
'bzrlib.tests.test_permissions',
977
2203
'bzrlib.tests.test_plugins',
978
2204
'bzrlib.tests.test_progress',
979
2205
'bzrlib.tests.test_reconcile',
2206
'bzrlib.tests.test_registry',
980
2207
'bzrlib.tests.test_repository',
2208
'bzrlib.tests.test_revert',
981
2209
'bzrlib.tests.test_revision',
982
2210
'bzrlib.tests.test_revisionnamespaces',
983
'bzrlib.tests.test_revprops',
2211
'bzrlib.tests.test_revisiontree',
984
2212
'bzrlib.tests.test_rio',
985
2213
'bzrlib.tests.test_sampler',
986
2214
'bzrlib.tests.test_selftest',
987
2215
'bzrlib.tests.test_setup',
988
2216
'bzrlib.tests.test_sftp_transport',
989
2217
'bzrlib.tests.test_smart_add',
2218
'bzrlib.tests.test_smart_transport',
990
2219
'bzrlib.tests.test_source',
2220
'bzrlib.tests.test_ssh_transport',
2221
'bzrlib.tests.test_status',
991
2222
'bzrlib.tests.test_store',
2223
'bzrlib.tests.test_strace',
2224
'bzrlib.tests.test_subsume',
992
2225
'bzrlib.tests.test_symbol_versioning',
2226
'bzrlib.tests.test_tag',
993
2227
'bzrlib.tests.test_testament',
2228
'bzrlib.tests.test_textfile',
2229
'bzrlib.tests.test_textmerge',
2230
'bzrlib.tests.test_timestamp',
994
2231
'bzrlib.tests.test_trace',
995
2232
'bzrlib.tests.test_transactions',
996
2233
'bzrlib.tests.test_transform',
997
2234
'bzrlib.tests.test_transport',
2235
'bzrlib.tests.test_tree',
2236
'bzrlib.tests.test_treebuilder',
998
2237
'bzrlib.tests.test_tsort',
2238
'bzrlib.tests.test_tuned_gzip',
999
2239
'bzrlib.tests.test_ui',
1000
2240
'bzrlib.tests.test_upgrade',
2241
'bzrlib.tests.test_urlutils',
1001
2242
'bzrlib.tests.test_versionedfile',
2243
'bzrlib.tests.test_version',
2244
'bzrlib.tests.test_version_info',
1002
2245
'bzrlib.tests.test_weave',
1003
2246
'bzrlib.tests.test_whitebox',
1004
2247
'bzrlib.tests.test_workingtree',
2248
'bzrlib.tests.test_workingtree_4',
2249
'bzrlib.tests.test_wsgi',
1005
2250
'bzrlib.tests.test_xml',
1007
2252
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()
2253
'bzrlib.tests.test_transport_implementations',
2254
'bzrlib.tests.test_read_bundle',
2256
suite = TestUtil.TestSuite()
2257
loader = TestUtil.TestLoader()
2258
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1021
2259
from bzrlib.transport import TransportTestProviderAdapter
1022
2260
adapter = TransportTestProviderAdapter()
1023
2261
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
2262
for package in packages_to_test():
1028
2263
suite.addTest(package.test_suite())
1029
2264
for m in MODULES_TO_TEST:
1030
2265
suite.addTest(loader.loadTestsFromModule(m))
1031
for m in (MODULES_TO_DOCTEST):
1032
suite.addTest(DocTestSuite(m))
2266
for m in MODULES_TO_DOCTEST:
2268
suite.addTest(doctest.DocTestSuite(m))
2269
except ValueError, e:
2270
print '**failed to get doctest for: %s\n%s' %(m,e)
1033
2272
for name, plugin in bzrlib.plugin.all_plugins().items():
1034
2273
if getattr(plugin, 'test_suite', None) is not None:
1035
suite.addTest(plugin.test_suite())
2274
default_encoding = sys.getdefaultencoding()
2276
plugin_suite = plugin.test_suite()
2277
except ImportError, e:
2278
bzrlib.trace.warning(
2279
'Unable to test plugin "%s": %s', name, e)
2281
suite.addTest(plugin_suite)
2282
if default_encoding != sys.getdefaultencoding():
2283
bzrlib.trace.warning(
2284
'Plugin "%s" tried to reset default encoding to: %s', name,
2285
sys.getdefaultencoding())
2287
sys.setdefaultencoding(default_encoding)
1039
2291
def adapt_modules(mods_list, adapter, loader, suite):
1040
2292
"""Adapt the modules in mods_list using adapter and add to suite."""
1041
for mod_name in mods_list:
1042
mod = _load_module_by_name(mod_name)
1043
for test in iter_suite_tests(loader.loadTestsFromModule(mod)):
1044
suite.addTests(adapter.adapt(test))
1047
def _load_module_by_name(mod_name):
1048
parts = mod_name.split('.')
1049
module = __import__(mod_name)
1051
# for historical reasons python returns the top-level module even though
1052
# it loads the submodule; we need to walk down to get the one we want.
1054
module = getattr(module, parts.pop(0))
2293
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
2294
suite.addTests(adapter.adapt(test))
2297
def _rmtree_temp_dir(dirname):
2299
osutils.rmtree(dirname)
2301
if sys.platform == 'win32' and e.errno == errno.EACCES:
2302
print >>sys.stderr, ('Permission denied: '
2303
'unable to remove testing dir '
2304
'%s' % os.path.basename(dirname))
2309
def clean_selftest_output(root=None, quiet=False):
2310
"""Remove all selftest output directories from root directory.
2312
:param root: root directory for clean
2313
(if ommitted or None then clean current directory).
2314
:param quiet: suppress report about deleting directories
2319
re_dir = re.compile(r'''test\d\d\d\d\.tmp''')
2322
for i in os.listdir(root):
2323
if os.path.isdir(i) and re_dir.match(i):
2325
print 'delete directory:', i
2329
class Feature(object):
2330
"""An operating system Feature."""
2333
self._available = None
2335
def available(self):
2336
"""Is the feature available?
2338
:return: True if the feature is available.
2340
if self._available is None:
2341
self._available = self._probe()
2342
return self._available
2345
"""Implement this method in concrete features.
2347
:return: True if the feature is available.
2349
raise NotImplementedError
2352
if getattr(self, 'feature_name', None):
2353
return self.feature_name()
2354
return self.__class__.__name__
added
removed
bzrlib/tests/__init__.py
