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:
236
215
class ExtendedTestResult(testtools.TextTestResult):
237
216
"""Accepts, reports and accumulates the results of running tests.
406
385
getDetails = getattr(test, "getDetails", None)
407
386
if getDetails is not None:
408
387
getDetails().clear()
409
_clear__type_equality_funcs(test)
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:
410
400
self._traceback_from_test = None
412
402
def startTests(self):
513
503
self.not_applicable_count += 1
514
504
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
520
506
def _post_mortem(self, tb=None):
521
507
"""Start a PDB post mortem session."""
522
508
if os.environ.get('BZR_TEST_PDB', None):
2583
2566
root = TestCaseWithMemoryTransport.TEST_ROOT
2584
# Make sure we get a readable and accessible home for .bzr.log
2585
# and/or config files, and not fallback to weird defaults (see
2586
# http://pad.lv/825027).
2587
self.assertIs(None, os.environ.get('BZR_HOME', None))
2588
os.environ['BZR_HOME'] = root
2589
wt = bzrdir.BzrDir.create_standalone_workingtree(root)
2590
del os.environ['BZR_HOME']
2568
# Make sure we get a readable and accessible home for .bzr.log
2569
# and/or config files, and not fallback to weird defaults (see
2570
# http://pad.lv/825027).
2571
self.assertIs(None, os.environ.get('BZR_HOME', None))
2572
os.environ['BZR_HOME'] = root
2573
wt = bzrdir.BzrDir.create_standalone_workingtree(root)
2574
del os.environ['BZR_HOME']
2575
except Exception, e:
2576
self.fail("Fail to initialize the safety net: %r\nExiting\n" % (e,))
2591
2577
# Hack for speed: remember the raw bytes of the dirstate file so that
2592
2578
# we don't need to re-open the wt to check it hasn't changed.
2593
2579
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE = (
2646
2632
repo = self.make_repository(relpath, format=format)
2647
2633
return repo.bzrdir.create_branch(append_revisions_only=False)
2649
def get_default_format(self):
2652
def resolve_format(self, format):
2653
"""Resolve an object to a ControlDir format object.
2655
The initial format object can either already be
2656
a ControlDirFormat, None (for the default format),
2657
or a string with the name of the control dir format.
2659
:param format: Object to resolve
2660
:return A ControlDirFormat instance
2663
format = self.get_default_format()
2635
def resolve_format(self, format):
2636
"""Resolve an object to a ControlDir format object.
2638
The initial format object can either already be
2639
a ControlDirFormat, None (for the default format),
2640
or a string with the name of the control dir format.
2642
:param format: Object to resolve
2643
:return A ControlDirFormat instance
2647
if isinstance(format, basestring):
2648
format = bzrdir.format_registry.make_bzrdir(format)
2651
def resolve_format(self, format):
2652
"""Resolve an object to a ControlDir format object.
2654
The initial format object can either already be
2655
a ControlDirFormat, None (for the default format),
2656
or a string with the name of the control dir format.
2658
:param format: Object to resolve
2659
:return A ControlDirFormat instance
2664
2663
if isinstance(format, basestring):
2665
2664
format = bzrdir.format_registry.make_bzrdir(format)
2958
2957
# this obviously requires a format that supports branch references
2959
2958
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
2961
format = self.resolve_format(format=format)
2962
if not format.supports_workingtrees:
2963
b = self.make_branch(relpath+'.branch', format=format)
2964
return b.create_checkout(relpath, lightweight=True)
2965
2960
b = self.make_branch(relpath, format=format)
2967
2962
return b.bzrdir.create_workingtree()
3266
3261
result_decorators=result_decorators,
3268
3263
runner.stop_on_failure=stop_on_failure
3269
if isinstance(suite, unittest.TestSuite):
3270
# Empty out _tests list of passed suite and populate new TestSuite
3271
suite._tests[:], suite = [], TestSuite(suite)
3272
3264
# built in decorator factories:
3274
3266
random_order(random_seed, runner),
3373
3365
class TestDecorator(TestUtil.TestSuite):
3374
3366
"""A decorator for TestCase/TestSuite objects.
3376
Contains rather than flattening suite passed on construction
3368
Usually, subclasses should override __iter__(used when flattening test
3369
suites), which we do to filter, reorder, parallelise and so on, run() and
3379
def __init__(self, suite=None):
3380
super(TestDecorator, self).__init__()
3381
if suite is not None:
3384
# Don't need subclass run method with suite emptying
3385
run = unittest.TestSuite.run
3373
def __init__(self, suite):
3374
TestUtil.TestSuite.__init__(self)
3377
def countTestCases(self):
3380
cases += test.countTestCases()
3387
def run(self, result):
3388
# Use iteration on self, not self._tests, to allow subclasses to hook
3391
if result.shouldStop:
3388
3397
class CountingDecorator(TestDecorator):
3399
3408
"""A decorator which excludes test matching an exclude pattern."""
3401
3410
def __init__(self, suite, exclude_pattern):
3402
super(ExcludeDecorator, self).__init__(
3403
exclude_tests_by_re(suite, exclude_pattern))
3411
TestDecorator.__init__(self, suite)
3412
self.exclude_pattern = exclude_pattern
3413
self.excluded = False
3417
return iter(self._tests)
3418
self.excluded = True
3419
suite = exclude_tests_by_re(self, self.exclude_pattern)
3421
self.addTests(suite)
3422
return iter(self._tests)
3406
3425
class FilterTestsDecorator(TestDecorator):
3407
3426
"""A decorator which filters tests to those matching a pattern."""
3409
3428
def __init__(self, suite, pattern):
3410
super(FilterTestsDecorator, self).__init__(
3411
filter_suite_by_re(suite, pattern))
3429
TestDecorator.__init__(self, suite)
3430
self.pattern = pattern
3431
self.filtered = False
3435
return iter(self._tests)
3436
self.filtered = True
3437
suite = filter_suite_by_re(self, self.pattern)
3439
self.addTests(suite)
3440
return iter(self._tests)
3414
3443
class RandomDecorator(TestDecorator):
3415
3444
"""A decorator which randomises the order of its tests."""
3417
3446
def __init__(self, suite, random_seed, stream):
3418
random_seed = self.actual_seed(random_seed)
3419
stream.write("Randomizing test order using seed %s\n\n" %
3447
TestDecorator.__init__(self, suite)
3448
self.random_seed = random_seed
3449
self.randomised = False
3450
self.stream = stream
3454
return iter(self._tests)
3455
self.randomised = True
3456
self.stream.write("Randomizing test order using seed %s\n\n" %
3457
(self.actual_seed()))
3421
3458
# Initialise the random number generator.
3422
random.seed(random_seed)
3423
super(RandomDecorator, self).__init__(randomize_suite(suite))
3459
random.seed(self.actual_seed())
3460
suite = randomize_suite(self)
3462
self.addTests(suite)
3463
return iter(self._tests)
3426
def actual_seed(seed):
3465
def actual_seed(self):
3466
if self.random_seed == "now":
3428
3467
# We convert the seed to a long to make it reuseable across
3429
3468
# invocations (because the user can reenter it).
3430
return long(time.time())
3469
self.random_seed = long(time.time())
3432
3471
# Convert the seed to a long if we can
3435
except (TypeError, ValueError):
3473
self.random_seed = long(self.random_seed)
3476
return self.random_seed
3440
3479
class TestFirstDecorator(TestDecorator):
3441
3480
"""A decorator which moves named tests to the front."""
3443
3482
def __init__(self, suite, pattern):
3444
super(TestFirstDecorator, self).__init__()
3445
self.addTests(split_suite_by_re(suite, pattern))
3483
TestDecorator.__init__(self, suite)
3484
self.pattern = pattern
3485
self.filtered = False
3489
return iter(self._tests)
3490
self.filtered = True
3491
suites = split_suite_by_re(self, self.pattern)
3493
self.addTests(suites)
3494
return iter(self._tests)
3448
3497
def partition_tests(suite, count):
3495
3544
os.waitpid(self.pid, 0)
3497
3546
test_blocks = partition_tests(suite, concurrency)
3498
# Clear the tests from the original suite so it doesn't keep them alive
3499
suite._tests[:] = []
3500
3547
for process_tests in test_blocks:
3501
process_suite = TestUtil.TestSuite(process_tests)
3502
# Also clear each split list so new suite has only reference
3503
process_tests[:] = []
3548
process_suite = TestUtil.TestSuite()
3549
process_suite.addTests(process_tests)
3504
3550
c2pread, c2pwrite = os.pipe()
3505
3551
pid = os.fork()
3512
3558
# read from stdin (otherwise its a roulette to see what
3513
3559
# child actually gets keystrokes for pdb etc).
3514
3560
sys.stdin.close()
3515
# GZ 2011-06-16: Why set stdin to None? Breaks multi fork.
3517
3562
stream = os.fdopen(c2pwrite, 'wb', 1)
3518
3563
subunit_result = AutoTimingTestResultDecorator(
3519
SubUnitBzrProtocolClient(stream))
3564
TestProtocolClient(stream))
3520
3565
process_suite.run(subunit_result)
3522
# GZ 2011-06-16: Is always exiting with silent success
3523
# really the right thing? Hurts debugging.
3526
3569
os.close(c2pwrite)
3633
3676
# with proper exclusion rules.
3634
3677
# -Ethreads Will display thread ident at creation/join time to
3635
3678
# help track thread leaks
3636
# -Euncollected_cases Display the identity of any test cases that weren't
3637
# deallocated after being completed.
3638
3680
# -Econfig_stats Will collect statistics using addDetail
3639
3681
selftest_debug_flags = set()
4438
4480
from subunit.test_results import AutoTimingTestResultDecorator
4439
4481
class SubUnitBzrProtocolClient(TestProtocolClient):
4441
def stopTest(self, test):
4442
super(SubUnitBzrProtocolClient, self).stopTest(test)
4443
_clear__type_equality_funcs(test)
4445
4483
def addSuccess(self, test, details=None):
4446
4484
# The subunit client always includes the details in the subunit
4447
4485
# stream, but we don't want to include it in ours.