877
821
ui.ui_factory = ui.SilentUIFactory()
878
822
self.addCleanup(_restore)
880
def _check_locks(self):
881
"""Check that all lock take/release actions have been paired."""
882
# We always check for mismatched locks. If a mismatch is found, we
883
# fail unless -Edisable_lock_checks is supplied to selftest, in which
884
# case we just print a warning.
886
acquired_locks = [lock for action, lock in self._lock_actions
887
if action == 'acquired']
888
released_locks = [lock for action, lock in self._lock_actions
889
if action == 'released']
890
broken_locks = [lock for action, lock in self._lock_actions
891
if action == 'broken']
892
# trivially, given the tests for lock acquistion and release, if we
893
# have as many in each list, it should be ok. Some lock tests also
894
# break some locks on purpose and should be taken into account by
895
# considering that breaking a lock is just a dirty way of releasing it.
896
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
897
message = ('Different number of acquired and '
898
'released or broken locks. (%s, %s + %s)' %
899
(acquired_locks, released_locks, broken_locks))
900
if not self._lock_check_thorough:
901
# Rather than fail, just warn
902
print "Broken test %s: %s" % (self, message)
906
def _track_locks(self):
907
"""Track lock activity during tests."""
908
self._lock_actions = []
909
if 'disable_lock_checks' in selftest_debug_flags:
910
self._lock_check_thorough = False
912
self._lock_check_thorough = True
914
self.addCleanup(self._check_locks)
915
_mod_lock.Lock.hooks.install_named_hook('lock_acquired',
916
self._lock_acquired, None)
917
_mod_lock.Lock.hooks.install_named_hook('lock_released',
918
self._lock_released, None)
919
_mod_lock.Lock.hooks.install_named_hook('lock_broken',
920
self._lock_broken, None)
922
def _lock_acquired(self, result):
923
self._lock_actions.append(('acquired', result))
925
def _lock_released(self, result):
926
self._lock_actions.append(('released', result))
928
def _lock_broken(self, result):
929
self._lock_actions.append(('broken', result))
931
824
def _ndiff_strings(self, a, b):
932
825
"""Return ndiff between two strings containing lines.
934
827
A trailing newline is added if missing to make the strings
935
828
print properly."""
936
829
if b and b[-1] != '\n':
2759
2464
list_only=False,
2760
2465
random_seed=None,
2761
2466
exclude_pattern=None,
2764
suite_decorators=None,
2766
"""Run a test suite for bzr selftest.
2768
:param runner_class: The class of runner to use. Must support the
2769
constructor arguments passed by run_suite which are more than standard
2771
:return: A boolean indicating success.
2773
2468
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
2778
if runner_class is None:
2779
runner_class = TextTestRunner
2782
runner = runner_class(stream=stream,
2473
runner = TextTestRunner(stream=sys.stdout,
2783
2474
descriptions=0,
2784
2475
verbosity=verbosity,
2785
2476
bench_history=bench_history,
2786
2477
list_only=list_only,
2789
2479
runner.stop_on_failure=stop_on_failure
2790
# built in decorator factories:
2792
random_order(random_seed, runner),
2793
exclude_tests(exclude_pattern),
2795
if matching_tests_first:
2796
decorators.append(tests_first(pattern))
2480
# Initialise the random number generator and display the seed used.
2481
# We convert the seed to a long to make it reuseable across invocations.
2482
random_order = False
2483
if random_seed is not None:
2485
if random_seed == "now":
2486
random_seed = long(time.time())
2488
# Convert the seed to a long if we can
2490
random_seed = long(random_seed)
2493
runner.stream.writeln("Randomizing test order using seed %s\n" %
2495
random.seed(random_seed)
2496
# Customise the list of tests if requested
2497
if exclude_pattern is not None:
2498
suite = exclude_tests_by_re(suite, exclude_pattern)
2500
order_changer = randomize_suite
2798
decorators.append(filter_tests(pattern))
2799
if suite_decorators:
2800
decorators.extend(suite_decorators)
2801
# tell the result object how many tests will be running: (except if
2802
# --parallel=fork is being used. Robert said he will provide a better
2803
# progress design later -- vila 20090817)
2804
if fork_decorator not in decorators:
2805
decorators.append(CountingDecorator)
2806
for decorator in decorators:
2807
suite = decorator(suite)
2502
order_changer = preserve_input
2503
if pattern != '.*' or random_order:
2504
if matching_tests_first:
2505
suites = map(order_changer, split_suite_by_re(suite, pattern))
2506
suite = TestUtil.TestSuite(suites)
2508
suite = order_changer(filter_suite_by_re(suite, pattern))
2808
2510
result = runner.run(suite)
2813
2513
return result.wasStrictlySuccessful()
2815
return result.wasSuccessful()
2818
# A registry where get() returns a suite decorator.
2819
parallel_registry = registry.Registry()
2822
def fork_decorator(suite):
2823
concurrency = osutils.local_concurrency()
2824
if concurrency == 1:
2826
from testtools import ConcurrentTestSuite
2827
return ConcurrentTestSuite(suite, fork_for_tests)
2828
parallel_registry.register('fork', fork_decorator)
2831
def subprocess_decorator(suite):
2832
concurrency = osutils.local_concurrency()
2833
if concurrency == 1:
2835
from testtools import ConcurrentTestSuite
2836
return ConcurrentTestSuite(suite, reinvoke_for_tests)
2837
parallel_registry.register('subprocess', subprocess_decorator)
2840
def exclude_tests(exclude_pattern):
2841
"""Return a test suite decorator that excludes tests."""
2842
if exclude_pattern is None:
2843
return identity_decorator
2844
def decorator(suite):
2845
return ExcludeDecorator(suite, exclude_pattern)
2849
def filter_tests(pattern):
2851
return identity_decorator
2852
def decorator(suite):
2853
return FilterTestsDecorator(suite, pattern)
2857
def random_order(random_seed, runner):
2858
"""Return a test suite decorator factory for randomising tests order.
2860
:param random_seed: now, a string which casts to a long, or a long.
2861
:param runner: A test runner with a stream attribute to report on.
2863
if random_seed is None:
2864
return identity_decorator
2865
def decorator(suite):
2866
return RandomDecorator(suite, random_seed, runner.stream)
2870
def tests_first(pattern):
2872
return identity_decorator
2873
def decorator(suite):
2874
return TestFirstDecorator(suite, pattern)
2878
def identity_decorator(suite):
2883
class TestDecorator(TestSuite):
2884
"""A decorator for TestCase/TestSuite objects.
2886
Usually, subclasses should override __iter__(used when flattening test
2887
suites), which we do to filter, reorder, parallelise and so on, run() and
2891
def __init__(self, suite):
2892
TestSuite.__init__(self)
2895
def countTestCases(self):
2898
cases += test.countTestCases()
2905
def run(self, result):
2906
# Use iteration on self, not self._tests, to allow subclasses to hook
2909
if result.shouldStop:
2915
class CountingDecorator(TestDecorator):
2916
"""A decorator which calls result.progress(self.countTestCases)."""
2918
def run(self, result):
2919
progress_method = getattr(result, 'progress', None)
2920
if callable(progress_method):
2921
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
2922
return super(CountingDecorator, self).run(result)
2925
class ExcludeDecorator(TestDecorator):
2926
"""A decorator which excludes test matching an exclude pattern."""
2928
def __init__(self, suite, exclude_pattern):
2929
TestDecorator.__init__(self, suite)
2930
self.exclude_pattern = exclude_pattern
2931
self.excluded = False
2935
return iter(self._tests)
2936
self.excluded = True
2937
suite = exclude_tests_by_re(self, self.exclude_pattern)
2939
self.addTests(suite)
2940
return iter(self._tests)
2943
class FilterTestsDecorator(TestDecorator):
2944
"""A decorator which filters tests to those matching a pattern."""
2946
def __init__(self, suite, pattern):
2947
TestDecorator.__init__(self, suite)
2948
self.pattern = pattern
2949
self.filtered = False
2953
return iter(self._tests)
2954
self.filtered = True
2955
suite = filter_suite_by_re(self, self.pattern)
2957
self.addTests(suite)
2958
return iter(self._tests)
2961
class RandomDecorator(TestDecorator):
2962
"""A decorator which randomises the order of its tests."""
2964
def __init__(self, suite, random_seed, stream):
2965
TestDecorator.__init__(self, suite)
2966
self.random_seed = random_seed
2967
self.randomised = False
2968
self.stream = stream
2972
return iter(self._tests)
2973
self.randomised = True
2974
self.stream.writeln("Randomizing test order using seed %s\n" %
2975
(self.actual_seed()))
2976
# Initialise the random number generator.
2977
random.seed(self.actual_seed())
2978
suite = randomize_suite(self)
2980
self.addTests(suite)
2981
return iter(self._tests)
2983
def actual_seed(self):
2984
if self.random_seed == "now":
2985
# We convert the seed to a long to make it reuseable across
2986
# invocations (because the user can reenter it).
2987
self.random_seed = long(time.time())
2989
# Convert the seed to a long if we can
2991
self.random_seed = long(self.random_seed)
2994
return self.random_seed
2997
class TestFirstDecorator(TestDecorator):
2998
"""A decorator which moves named tests to the front."""
3000
def __init__(self, suite, pattern):
3001
TestDecorator.__init__(self, suite)
3002
self.pattern = pattern
3003
self.filtered = False
3007
return iter(self._tests)
3008
self.filtered = True
3009
suites = split_suite_by_re(self, self.pattern)
3011
self.addTests(suites)
3012
return iter(self._tests)
3015
def partition_tests(suite, count):
3016
"""Partition suite into count lists of tests."""
3018
tests = list(iter_suite_tests(suite))
3019
tests_per_process = int(math.ceil(float(len(tests)) / count))
3020
for block in range(count):
3021
low_test = block * tests_per_process
3022
high_test = low_test + tests_per_process
3023
process_tests = tests[low_test:high_test]
3024
result.append(process_tests)
3028
def fork_for_tests(suite):
3029
"""Take suite and start up one runner per CPU by forking()
3031
:return: An iterable of TestCase-like objects which can each have
3032
run(result) called on them to feed tests to result.
3034
concurrency = osutils.local_concurrency()
3036
from subunit import TestProtocolClient, ProtocolTestCase
3038
from subunit.test_results import AutoTimingTestResultDecorator
3040
AutoTimingTestResultDecorator = lambda x:x
3041
class TestInOtherProcess(ProtocolTestCase):
3042
# Should be in subunit, I think. RBC.
3043
def __init__(self, stream, pid):
3044
ProtocolTestCase.__init__(self, stream)
3047
def run(self, result):
3049
ProtocolTestCase.run(self, result)
3051
os.waitpid(self.pid, os.WNOHANG)
3053
test_blocks = partition_tests(suite, concurrency)
3054
for process_tests in test_blocks:
3055
process_suite = TestSuite()
3056
process_suite.addTests(process_tests)
3057
c2pread, c2pwrite = os.pipe()
3062
# Leave stderr and stdout open so we can see test noise
3063
# Close stdin so that the child goes away if it decides to
3064
# read from stdin (otherwise its a roulette to see what
3065
# child actually gets keystrokes for pdb etc).
3068
stream = os.fdopen(c2pwrite, 'wb', 1)
3069
subunit_result = AutoTimingTestResultDecorator(
3070
TestProtocolClient(stream))
3071
process_suite.run(subunit_result)
3076
stream = os.fdopen(c2pread, 'rb', 1)
3077
test = TestInOtherProcess(stream, pid)
3082
def reinvoke_for_tests(suite):
3083
"""Take suite and start up one runner per CPU using subprocess().
3085
:return: An iterable of TestCase-like objects which can each have
3086
run(result) called on them to feed tests to result.
3088
concurrency = osutils.local_concurrency()
3090
from subunit import ProtocolTestCase
3091
class TestInSubprocess(ProtocolTestCase):
3092
def __init__(self, process, name):
3093
ProtocolTestCase.__init__(self, process.stdout)
3094
self.process = process
3095
self.process.stdin.close()
3098
def run(self, result):
3100
ProtocolTestCase.run(self, result)
3103
os.unlink(self.name)
3104
# print "pid %d finished" % finished_process
3105
test_blocks = partition_tests(suite, concurrency)
3106
for process_tests in test_blocks:
3107
# ugly; currently reimplement rather than reuses TestCase methods.
3108
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3109
if not os.path.isfile(bzr_path):
3110
# We are probably installed. Assume sys.argv is the right file
3111
bzr_path = sys.argv[0]
3112
fd, test_list_file_name = tempfile.mkstemp()
3113
test_list_file = os.fdopen(fd, 'wb', 1)
3114
for test in process_tests:
3115
test_list_file.write(test.id() + '\n')
3116
test_list_file.close()
3118
argv = [bzr_path, 'selftest', '--load-list', test_list_file_name,
3120
if '--no-plugins' in sys.argv:
3121
argv.append('--no-plugins')
3122
# stderr=STDOUT would be ideal, but until we prevent noise on
3123
# stderr it can interrupt the subunit protocol.
3124
process = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE,
3126
test = TestInSubprocess(process, test_list_file_name)
3129
os.unlink(test_list_file_name)
3134
class BZRTransformingResult(unittest.TestResult):
3136
def __init__(self, target):
3137
unittest.TestResult.__init__(self)
3138
self.result = target
3140
def startTest(self, test):
3141
self.result.startTest(test)
3143
def stopTest(self, test):
3144
self.result.stopTest(test)
3146
def addError(self, test, err):
3147
feature = self._error_looks_like('UnavailableFeature: ', err)
3148
if feature is not None:
3149
self.result.addNotSupported(test, feature)
3151
self.result.addError(test, err)
3153
def addFailure(self, test, err):
3154
known = self._error_looks_like('KnownFailure: ', err)
3155
if known is not None:
3156
self.result._addKnownFailure(test, [KnownFailure,
3157
KnownFailure(known), None])
3159
self.result.addFailure(test, err)
3161
def addSkip(self, test, reason):
3162
self.result.addSkip(test, reason)
3164
def addSuccess(self, test):
3165
self.result.addSuccess(test)
3167
def _error_looks_like(self, prefix, err):
3168
"""Deserialize exception and returns the stringify value."""
3172
if isinstance(exc, subunit.RemoteException):
3173
# stringify the exception gives access to the remote traceback
3174
# We search the last line for 'prefix'
3175
lines = str(exc).split('\n')
3176
while lines and not lines[-1]:
3179
if lines[-1].startswith(prefix):
3180
value = lines[-1][len(prefix):]
2515
return result.wasSuccessful()
3184
2518
# Controlled by "bzr selftest -E=..." option
3185
# Currently supported:
3186
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3187
# preserves any flags supplied at the command line.
3188
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3189
# rather than failing tests. And no longer raise
3190
# LockContention when fctnl locks are not being used
3191
# with proper exclusion rules.
3192
2519
selftest_debug_flags = set()
3721
3062
for right_name, right_dict in scenarios_right]
3724
def multiply_tests(tests, scenarios, result):
3725
"""Multiply tests_list by scenarios into result.
3727
This is the core workhorse for test parameterisation.
3729
Typically the load_tests() method for a per-implementation test suite will
3730
call multiply_tests and return the result.
3732
:param tests: The tests to parameterise.
3733
:param scenarios: The scenarios to apply: pairs of (scenario_name,
3734
scenario_param_dict).
3735
:param result: A TestSuite to add created tests to.
3737
This returns the passed in result TestSuite with the cross product of all
3738
the tests repeated once for each scenario. Each test is adapted by adding
3739
the scenario name at the end of its id(), and updating the test object's
3740
__dict__ with the scenario_param_dict.
3742
>>> import bzrlib.tests.test_sampler
3743
>>> r = multiply_tests(
3744
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
3745
... [('one', dict(param=1)),
3746
... ('two', dict(param=2))],
3748
>>> tests = list(iter_suite_tests(r))
3752
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
3758
for test in iter_suite_tests(tests):
3759
apply_scenarios(test, scenarios, result)
3763
def apply_scenarios(test, scenarios, result):
3764
"""Apply the scenarios in scenarios to test and add to result.
3766
:param test: The test to apply scenarios to.
3767
:param scenarios: An iterable of scenarios to apply to test.
3769
:seealso: apply_scenario
3771
for scenario in scenarios:
3772
result.addTest(apply_scenario(test, scenario))
3776
def apply_scenario(test, scenario):
3777
"""Copy test and apply scenario to it.
3779
:param test: A test to adapt.
3780
:param scenario: A tuple describing the scenarion.
3781
The first element of the tuple is the new test id.
3782
The second element is a dict containing attributes to set on the
3784
:return: The adapted test.
3786
new_id = "%s(%s)" % (test.id(), scenario[0])
3787
new_test = clone_test(test, new_id)
3788
for name, value in scenario[1].items():
3789
setattr(new_test, name, value)
3793
def clone_test(test, new_id):
3794
"""Clone a test giving it a new id.
3796
:param test: The test to clone.
3797
:param new_id: The id to assign to it.
3798
:return: The new test.
3800
from copy import deepcopy
3801
new_test = deepcopy(test)
3802
new_test.id = lambda: new_id
3066
def adapt_modules(mods_list, adapter, loader, suite):
3067
"""Adapt the modules in mods_list using adapter and add to suite."""
3068
tests = loader.loadTestsFromModuleNames(mods_list)
3069
adapt_tests(tests, adapter, suite)
3072
def adapt_tests(tests_list, adapter, suite):
3073
"""Adapt the tests in tests_list using adapter and add to suite."""
3074
for test in iter_suite_tests(tests_list):
3075
suite.addTests(adapter.adapt(test))
3806
3078
def _rmtree_temp_dir(dirname):
added
removed
bzrlib/tests/__init__.py
