165
167
self.unsupported = {}
167
169
self._overall_start_time = time.time()
170
self._strict = strict
174
ok = self.wasStrictlySuccessful()
176
ok = self.wasSuccessful()
178
self.stream.write('tests passed\n')
180
self.stream.write('tests failed\n')
181
if TestCase._first_thread_leaker_id:
183
'%s is leaking threads among %d leaking tests.\n' % (
184
TestCase._first_thread_leaker_id,
185
TestCase._leaking_threads_tests))
169
187
def _extractBenchmarkTime(self, testCase):
170
188
"""Add a benchmark time for the current test case."""
197
215
def startTest(self, test):
198
216
unittest.TestResult.startTest(self, test)
199
219
self.report_test_start(test)
200
220
test.number = self.count
201
221
self._recordTestStartTime()
223
def startTests(self):
225
'testing: %s\n' % (osutils.realpath(sys.argv[0]),))
227
' %s (%s python%s)\n' % (
229
bzrlib.version_string,
230
bzrlib._format_version_tuple(sys.version_info),
232
self.stream.write('\n')
203
234
def _recordTestStartTime(self):
204
235
"""Record that a test has started."""
205
236
self._start_time = time.time()
349
380
bench_history=None,
353
385
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
354
bench_history, num_tests)
386
bench_history, num_tests, strict)
356
388
self.pb = self.ui.nested_progress_bar()
357
389
self._supplied_pb = False
514
546
bench_history=None,
517
550
self.stream = unittest._WritelnDecorator(stream)
518
551
self.descriptions = descriptions
519
552
self.verbosity = verbosity
520
553
self._bench_history = bench_history
521
554
self.list_only = list_only
555
self._strict = strict
523
557
def run(self, test):
524
558
"Run the given test case or test suite."
719
def _report_leaked_threads():
720
bzrlib.trace.warning('%s is leaking threads among %d leaking tests',
721
TestCase._first_thread_leaker_id,
722
TestCase._leaking_threads_tests)
725
754
class TestCase(unittest.TestCase):
726
755
"""Base class for bzr unit tests.
787
818
TestCase._leaking_threads_tests += 1
788
819
if TestCase._first_thread_leaker_id is None:
789
820
TestCase._first_thread_leaker_id = self.id()
790
# we're not specifically told when all tests are finished.
791
# This will do. We use a function to avoid keeping a reference
792
# to a TestCase object.
793
atexit.register(_report_leaked_threads)
795
822
def _clear_debug_flags(self):
796
823
"""Prevent externally set debug flags affecting tests.
826
853
ui.ui_factory = ui.SilentUIFactory()
827
854
self.addCleanup(_restore)
856
def _check_locks(self):
857
"""Check that all lock take/release actions have been paired."""
858
# once we have fixed all the current lock problems, we can change the
859
# following code to always check for mismatched locks, but only do
860
# traceback showing with -Dlock (self._lock_check_thorough is True).
861
# For now, because the test suite will fail, we only assert that lock
862
# matching has occured with -Dlock.
864
acquired_locks = [lock for action, lock in self._lock_actions
865
if action == 'acquired']
866
released_locks = [lock for action, lock in self._lock_actions
867
if action == 'released']
868
# trivially, given the tests for lock acquistion and release, if we
869
# have as many in each list, it should be ok.
870
if len(acquired_locks) != len(released_locks):
872
("Different number of acquired and released locks. (%s, %s)" %
873
(acquired_locks, released_locks))
874
if not self._lock_check_thorough:
875
# Rather than fail, just warn
876
print "Broken test %s: %s" % (self, message)
880
def _track_locks(self):
881
"""Track lock activity during tests."""
882
self._lock_actions = []
883
self._lock_check_thorough = 'lock' in debug.debug_flags
884
self.addCleanup(self._check_locks)
885
_mod_lock.Lock.hooks.install_named_hook('lock_acquired', self._lock_acquired, None)
886
_mod_lock.Lock.hooks.install_named_hook('lock_released', self._lock_released, None)
888
def _lock_acquired(self, result):
889
self._lock_actions.append(('acquired', result))
891
def _lock_released(self, result):
892
self._lock_actions.append(('released', result))
829
894
def _ndiff_strings(self, a, b):
830
895
"""Return ndiff between two strings containing lines.
1326
1391
"test setUp did not invoke "
1327
1392
"bzrlib.tests.TestCase's setUp")
1328
1393
except KeyboardInterrupt:
1330
1396
except TestSkipped, e:
1331
1397
self._do_skip(result, e.args[0])
1332
1398
self.tearDown()
1335
1401
result.addError(self, sys.exc_info())
1359
1427
"test tearDown did not invoke "
1360
1428
"bzrlib.tests.TestCase's tearDown")
1361
1429
except KeyboardInterrupt:
1364
1433
result.addError(self, sys.exc_info())
1366
1436
if ok: result.addSuccess(self)
1368
1438
result.stopTest(self)
1370
1440
except TestNotApplicable:
1371
1441
# Not moved from the result [yet].
1373
1444
except KeyboardInterrupt:
1376
1448
saved_attrs = {}
1382
1454
self.__dict__ = saved_attrs
1384
1456
def tearDown(self):
1385
self._bzr_test_tearDown_run = True
1386
1457
self._runCleanups()
1387
1458
self._log_contents = ''
1459
self._bzr_test_tearDown_run = True
1388
1460
unittest.TestCase.tearDown(self)
1390
1462
def time(self, callable, *args, **kwargs):
2633
2706
if runner_class is None:
2634
2707
runner_class = TextTestRunner
2635
runner = runner_class(stream=sys.stdout,
2710
runner = runner_class(stream=stream,
2636
2711
descriptions=0,
2637
2712
verbosity=verbosity,
2638
2713
bench_history=bench_history,
2639
2714
list_only=list_only,
2641
2717
runner.stop_on_failure=stop_on_failure
2642
2718
# built in decorator factories:
3908
3985
from subunit import TestProtocolClient
3909
3986
class SubUnitBzrRunner(TextTestRunner):
3910
3987
def run(self, test):
3911
# undo out claim for testing which looks like a test start to subunit
3912
self.stream.write("success: %s\n" % (osutils.realpath(sys.argv[0]),))
3913
3988
result = TestProtocolClient(self.stream)
3914
3989
test.run(result)
added
removed
bzrlib/tests/__init__.py
