212
212
osutils.set_or_unset_env(var, value)
215
def _clear__type_equality_funcs(test):
216
"""Cleanup bound methods stored on TestCase instances
218
Clear the dict breaking a few (mostly) harmless cycles in the affected
219
unittests released with Python 2.6 and initial Python 2.7 versions.
221
For a few revisions between Python 2.7.1 and Python 2.7.2 that annoyingly
222
shipped in Oneiric, an object with no clear method was used, hence the
223
extra complications, see bug 809048 for details.
225
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
226
if type_equality_funcs is not None:
227
tef_clear = getattr(type_equality_funcs, "clear", None)
228
if tef_clear is None:
229
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
230
if tef_instance_dict is not None:
231
tef_clear = tef_instance_dict.clear
232
if tef_clear is not None:
215
236
class ExtendedTestResult(testtools.TextTestResult):
216
237
"""Accepts, reports and accumulates the results of running tests.
385
406
getDetails = getattr(test, "getDetails", None)
386
407
if getDetails is not None:
387
408
getDetails().clear()
388
# Clear _type_equality_funcs to try to stop TestCase instances
389
# from wasting memory. 'clear' is not available in all Python
390
# versions (bug 809048)
391
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
392
if type_equality_funcs is not None:
393
tef_clear = getattr(type_equality_funcs, "clear", None)
394
if tef_clear is None:
395
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
396
if tef_instance_dict is not None:
397
tef_clear = tef_instance_dict.clear
398
if tef_clear is not None:
409
_clear__type_equality_funcs(test)
400
410
self._traceback_from_test = None
402
412
def startTests(self):
503
513
self.not_applicable_count += 1
504
514
self.report_not_applicable(test, reason)
516
def _count_stored_tests(self):
517
"""Count of tests instances kept alive due to not succeeding"""
518
return self.error_count + self.failure_count + self.known_failure_count
506
520
def _post_mortem(self, tb=None):
507
521
"""Start a PDB post mortem session."""
508
522
if os.environ.get('BZR_TEST_PDB', None):
983
997
for feature in getattr(self, '_test_needs_features', []):
984
998
self.requireFeature(feature)
985
999
self._cleanEnvironment()
986
self.overrideAttr(bzrlib.global_state, 'cmdline_overrides',
987
config.CommandLineSection())
1000
if bzrlib.global_state is not None:
1001
self.overrideAttr(bzrlib.global_state, 'cmdline_overrides',
1002
config.CommandLineSection())
988
1003
self._silenceUI()
989
1004
self._startLogFile()
990
1005
self._benchcalls = []
1719
1731
self.addCleanup(self._finishLogFile)
1721
1733
def _finishLogFile(self):
1722
"""Finished with the log file.
1724
Close the file and delete it.
1734
"""Flush and dereference the in-memory log for this testcase"""
1726
1735
if trace._trace_file:
1727
1736
# flush the log file, to get all content
1728
1737
trace._trace_file.flush()
1729
1738
trace.pop_log_file(self._log_memento)
1739
# The logging module now tracks references for cleanup so discard ours
1740
del self._log_memento
1731
1742
def thisFailsStrictLockCheck(self):
1732
1743
"""It is known that this test would fail with -Dstrict_locks.
2568
2579
root = TestCaseWithMemoryTransport.TEST_ROOT
2570
# Make sure we get a readable and accessible home for .bzr.log
2571
# and/or config files, and not fallback to weird defaults (see
2572
# http://pad.lv/825027).
2573
self.assertIs(None, os.environ.get('BZR_HOME', None))
2574
os.environ['BZR_HOME'] = root
2575
wt = bzrdir.BzrDir.create_standalone_workingtree(root)
2576
del os.environ['BZR_HOME']
2577
except Exception, e:
2578
self.fail("Fail to initialize the safety net: %r\nExiting\n" % (e,))
2580
# Make sure we get a readable and accessible home for .bzr.log
2581
# and/or config files, and not fallback to weird defaults (see
2582
# http://pad.lv/825027).
2583
self.assertIs(None, os.environ.get('BZR_HOME', None))
2584
os.environ['BZR_HOME'] = root
2585
wt = bzrdir.BzrDir.create_standalone_workingtree(root)
2586
del os.environ['BZR_HOME']
2579
2587
# Hack for speed: remember the raw bytes of the dirstate file so that
2580
2588
# we don't need to re-open the wt to check it hasn't changed.
2581
2589
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE = (
2634
2642
repo = self.make_repository(relpath, format=format)
2635
2643
return repo.bzrdir.create_branch(append_revisions_only=False)
2637
def resolve_format(self, format):
2638
"""Resolve an object to a ControlDir format object.
2640
The initial format object can either already be
2641
a ControlDirFormat, None (for the default format),
2642
or a string with the name of the control dir format.
2644
:param format: Object to resolve
2645
:return A ControlDirFormat instance
2649
if isinstance(format, basestring):
2650
format = bzrdir.format_registry.make_bzrdir(format)
2653
def resolve_format(self, format):
2654
"""Resolve an object to a ControlDir format object.
2656
The initial format object can either already be
2657
a ControlDirFormat, None (for the default format),
2658
or a string with the name of the control dir format.
2660
:param format: Object to resolve
2661
:return A ControlDirFormat instance
2645
def get_default_format(self):
2648
def resolve_format(self, format):
2649
"""Resolve an object to a ControlDir format object.
2651
The initial format object can either already be
2652
a ControlDirFormat, None (for the default format),
2653
or a string with the name of the control dir format.
2655
:param format: Object to resolve
2656
:return A ControlDirFormat instance
2659
format = self.get_default_format()
2665
2660
if isinstance(format, basestring):
2666
2661
format = bzrdir.format_registry.make_bzrdir(format)
2959
2954
# this obviously requires a format that supports branch references
2960
2955
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
2957
format = self.resolve_format(format=format)
2958
if not format.supports_workingtrees:
2959
b = self.make_branch(relpath+'.branch', format=format)
2960
return b.create_checkout(relpath, lightweight=True)
2962
2961
b = self.make_branch(relpath, format=format)
2964
2963
return b.bzrdir.create_workingtree()
3263
3262
result_decorators=result_decorators,
3265
3264
runner.stop_on_failure=stop_on_failure
3265
if isinstance(suite, unittest.TestSuite):
3266
# Empty out _tests list of passed suite and populate new TestSuite
3267
suite._tests[:], suite = [], TestSuite(suite)
3266
3268
# built in decorator factories:
3268
3270
random_order(random_seed, runner),
3367
3369
class TestDecorator(TestUtil.TestSuite):
3368
3370
"""A decorator for TestCase/TestSuite objects.
3370
Usually, subclasses should override __iter__(used when flattening test
3371
suites), which we do to filter, reorder, parallelise and so on, run() and
3372
Contains rather than flattening suite passed on construction
3375
def __init__(self, suite):
3376
TestUtil.TestSuite.__init__(self)
3379
def countTestCases(self):
3382
cases += test.countTestCases()
3389
def run(self, result):
3390
# Use iteration on self, not self._tests, to allow subclasses to hook
3393
if result.shouldStop:
3375
def __init__(self, suite=None):
3376
super(TestDecorator, self).__init__()
3377
if suite is not None:
3380
# Don't need subclass run method with suite emptying
3381
run = unittest.TestSuite.run
3399
3384
class CountingDecorator(TestDecorator):
3410
3395
"""A decorator which excludes test matching an exclude pattern."""
3412
3397
def __init__(self, suite, exclude_pattern):
3413
TestDecorator.__init__(self, suite)
3414
self.exclude_pattern = exclude_pattern
3415
self.excluded = False
3419
return iter(self._tests)
3420
self.excluded = True
3421
suite = exclude_tests_by_re(self, self.exclude_pattern)
3423
self.addTests(suite)
3424
return iter(self._tests)
3398
super(ExcludeDecorator, self).__init__(
3399
exclude_tests_by_re(suite, exclude_pattern))
3427
3402
class FilterTestsDecorator(TestDecorator):
3428
3403
"""A decorator which filters tests to those matching a pattern."""
3430
3405
def __init__(self, suite, pattern):
3431
TestDecorator.__init__(self, suite)
3432
self.pattern = pattern
3433
self.filtered = False
3437
return iter(self._tests)
3438
self.filtered = True
3439
suite = filter_suite_by_re(self, self.pattern)
3441
self.addTests(suite)
3442
return iter(self._tests)
3406
super(FilterTestsDecorator, self).__init__(
3407
filter_suite_by_re(suite, pattern))
3445
3410
class RandomDecorator(TestDecorator):
3446
3411
"""A decorator which randomises the order of its tests."""
3448
3413
def __init__(self, suite, random_seed, stream):
3449
TestDecorator.__init__(self, suite)
3450
self.random_seed = random_seed
3451
self.randomised = False
3452
self.stream = stream
3456
return iter(self._tests)
3457
self.randomised = True
3458
self.stream.write("Randomizing test order using seed %s\n\n" %
3459
(self.actual_seed()))
3414
random_seed = self.actual_seed(random_seed)
3415
stream.write("Randomizing test order using seed %s\n\n" %
3460
3417
# Initialise the random number generator.
3461
random.seed(self.actual_seed())
3462
suite = randomize_suite(self)
3464
self.addTests(suite)
3465
return iter(self._tests)
3418
random.seed(random_seed)
3419
super(RandomDecorator, self).__init__(randomize_suite(suite))
3467
def actual_seed(self):
3468
if self.random_seed == "now":
3422
def actual_seed(seed):
3469
3424
# We convert the seed to a long to make it reuseable across
3470
3425
# invocations (because the user can reenter it).
3471
self.random_seed = long(time.time())
3426
return long(time.time())
3473
3428
# Convert the seed to a long if we can
3475
self.random_seed = long(self.random_seed)
3431
except (TypeError, ValueError):
3478
return self.random_seed
3481
3436
class TestFirstDecorator(TestDecorator):
3482
3437
"""A decorator which moves named tests to the front."""
3484
3439
def __init__(self, suite, pattern):
3485
TestDecorator.__init__(self, suite)
3486
self.pattern = pattern
3487
self.filtered = False
3491
return iter(self._tests)
3492
self.filtered = True
3493
suites = split_suite_by_re(self, self.pattern)
3495
self.addTests(suites)
3496
return iter(self._tests)
3440
super(TestFirstDecorator, self).__init__()
3441
self.addTests(split_suite_by_re(suite, pattern))
3499
3444
def partition_tests(suite, count):
3544
3489
ProtocolTestCase.run(self, result)
3546
os.waitpid(self.pid, 0)
3491
pid, status = os.waitpid(self.pid, 0)
3492
# GZ 2011-10-18: If status is nonzero, should report to the result
3493
# that something went wrong.
3548
3495
test_blocks = partition_tests(suite, concurrency)
3496
# Clear the tests from the original suite so it doesn't keep them alive
3497
suite._tests[:] = []
3549
3498
for process_tests in test_blocks:
3550
process_suite = TestUtil.TestSuite()
3551
process_suite.addTests(process_tests)
3499
process_suite = TestUtil.TestSuite(process_tests)
3500
# Also clear each split list so new suite has only reference
3501
process_tests[:] = []
3552
3502
c2pread, c2pwrite = os.pipe()
3553
3503
pid = os.fork()
3555
workaround_zealous_crypto_random()
3506
stream = os.fdopen(c2pwrite, 'wb', 1)
3507
workaround_zealous_crypto_random()
3557
3508
os.close(c2pread)
3558
3509
# Leave stderr and stdout open so we can see test noise
3559
3510
# Close stdin so that the child goes away if it decides to
3560
3511
# read from stdin (otherwise its a roulette to see what
3561
3512
# child actually gets keystrokes for pdb etc).
3562
3513
sys.stdin.close()
3564
stream = os.fdopen(c2pwrite, 'wb', 1)
3565
3514
subunit_result = AutoTimingTestResultDecorator(
3566
TestProtocolClient(stream))
3515
SubUnitBzrProtocolClient(stream))
3567
3516
process_suite.run(subunit_result)
3518
# Try and report traceback on stream, but exit with error even
3519
# if stream couldn't be created or something else goes wrong
3521
traceback.print_exc(file=stream)
3571
3526
os.close(c2pwrite)
3572
3527
stream = os.fdopen(c2pread, 'rb', 1)
3678
3633
# with proper exclusion rules.
3679
3634
# -Ethreads Will display thread ident at creation/join time to
3680
3635
# help track thread leaks
3636
# -Euncollected_cases Display the identity of any test cases that weren't
3637
# deallocated after being completed.
3682
3638
# -Econfig_stats Will collect statistics using addDetail
3683
3639
selftest_debug_flags = set()
4482
4438
from subunit.test_results import AutoTimingTestResultDecorator
4483
4439
class SubUnitBzrProtocolClient(TestProtocolClient):
4441
def stopTest(self, test):
4442
super(SubUnitBzrProtocolClient, self).stopTest(test)
4443
_clear__type_equality_funcs(test)
4485
4445
def addSuccess(self, test, details=None):
4486
4446
# The subunit client always includes the details in the subunit
4487
4447
# stream, but we don't want to include it in ours.
added
removed
bzrlib/tests/__init__.py
