2645
2661
list_only=list_only,
2647
2663
runner.stop_on_failure=stop_on_failure
2648
# Initialise the random number generator and display the seed used.
2649
# We convert the seed to a long to make it reuseable across invocations.
2650
random_order = False
2651
if random_seed is not None:
2653
if random_seed == "now":
2654
random_seed = long(time.time())
2664
# built in decorator factories:
2666
random_order(random_seed, runner),
2667
exclude_tests(exclude_pattern),
2669
if matching_tests_first:
2670
decorators.append(tests_first(pattern))
2672
decorators.append(filter_tests(pattern))
2673
if suite_decorators:
2674
decorators.extend(suite_decorators)
2675
for decorator in decorators:
2676
suite = decorator(suite)
2677
result = runner.run(suite)
2679
return result.wasStrictlySuccessful()
2681
return result.wasSuccessful()
2684
# A registry where get() returns a suite decorator.
2685
parallel_registry = registry.Registry()
2686
def fork_decorator(suite):
2687
concurrency = local_concurrency()
2688
if concurrency == 1:
2690
from testtools import ConcurrentTestSuite
2691
return ConcurrentTestSuite(suite, fork_for_tests)
2692
parallel_registry.register('fork', fork_decorator)
2693
def subprocess_decorator(suite):
2694
concurrency = local_concurrency()
2695
if concurrency == 1:
2697
from testtools import ConcurrentTestSuite
2698
return ConcurrentTestSuite(suite, reinvoke_for_tests)
2699
parallel_registry.register('subprocess', subprocess_decorator)
2702
def exclude_tests(exclude_pattern):
2703
"""Return a test suite decorator that excludes tests."""
2704
if exclude_pattern is None:
2705
return identity_decorator
2706
def decorator(suite):
2707
return ExcludeDecorator(suite, exclude_pattern)
2711
def filter_tests(pattern):
2713
return identity_decorator
2714
def decorator(suite):
2715
return FilterTestsDecorator(suite, pattern)
2719
def random_order(random_seed, runner):
2720
"""Return a test suite decorator factory for randomising tests order.
2722
:param random_seed: now, a string which casts to a long, or a long.
2723
:param runner: A test runner with a stream attribute to report on.
2725
if random_seed is None:
2726
return identity_decorator
2727
def decorator(suite):
2728
return RandomDecorator(suite, random_seed, runner.stream)
2732
def tests_first(pattern):
2734
return identity_decorator
2735
def decorator(suite):
2736
return TestFirstDecorator(suite, pattern)
2740
def identity_decorator(suite):
2745
class TestDecorator(TestSuite):
2746
"""A decorator for TestCase/TestSuite objects.
2748
Usually, subclasses should override __iter__(used when flattening test
2749
suites), which we do to filter, reorder, parallelise and so on, run() and
2753
def __init__(self, suite):
2754
TestSuite.__init__(self)
2757
def countTestCases(self):
2760
cases += test.countTestCases()
2767
def run(self, result):
2768
# Use iteration on self, not self._tests, to allow subclasses to hook
2771
if result.shouldStop:
2777
class ExcludeDecorator(TestDecorator):
2778
"""A decorator which excludes test matching an exclude pattern."""
2780
def __init__(self, suite, exclude_pattern):
2781
TestDecorator.__init__(self, suite)
2782
self.exclude_pattern = exclude_pattern
2783
self.excluded = False
2787
return iter(self._tests)
2788
self.excluded = True
2789
suite = exclude_tests_by_re(self, self.exclude_pattern)
2791
self.addTests(suite)
2792
return iter(self._tests)
2795
class FilterTestsDecorator(TestDecorator):
2796
"""A decorator which filters tests to those matching a pattern."""
2798
def __init__(self, suite, pattern):
2799
TestDecorator.__init__(self, suite)
2800
self.pattern = pattern
2801
self.filtered = False
2805
return iter(self._tests)
2806
self.filtered = True
2807
suite = filter_suite_by_re(self, self.pattern)
2809
self.addTests(suite)
2810
return iter(self._tests)
2813
class RandomDecorator(TestDecorator):
2814
"""A decorator which randomises the order of its tests."""
2816
def __init__(self, suite, random_seed, stream):
2817
TestDecorator.__init__(self, suite)
2818
self.random_seed = random_seed
2819
self.randomised = False
2820
self.stream = stream
2824
return iter(self._tests)
2825
self.randomised = True
2826
self.stream.writeln("Randomizing test order using seed %s\n" %
2827
(self.actual_seed()))
2828
# Initialise the random number generator.
2829
random.seed(self.actual_seed())
2830
suite = randomize_suite(self)
2832
self.addTests(suite)
2833
return iter(self._tests)
2835
def actual_seed(self):
2836
if self.random_seed == "now":
2837
# We convert the seed to a long to make it reuseable across
2838
# invocations (because the user can reenter it).
2839
self.random_seed = long(time.time())
2656
2841
# Convert the seed to a long if we can
2658
random_seed = long(random_seed)
2843
self.random_seed = long(self.random_seed)
2661
runner.stream.writeln("Randomizing test order using seed %s\n" %
2663
random.seed(random_seed)
2664
# Customise the list of tests if requested
2665
if exclude_pattern is not None:
2666
suite = exclude_tests_by_re(suite, exclude_pattern)
2668
order_changer = randomize_suite
2670
order_changer = preserve_input
2671
if pattern != '.*' or random_order:
2672
if matching_tests_first:
2673
suites = map(order_changer, split_suite_by_re(suite, pattern))
2674
suite = TestUtil.TestSuite(suites)
2676
suite = order_changer(filter_suite_by_re(suite, pattern))
2678
result = runner.run(suite)
2681
return result.wasStrictlySuccessful()
2683
return result.wasSuccessful()
2846
return self.random_seed
2849
class TestFirstDecorator(TestDecorator):
2850
"""A decorator which moves named tests to the front."""
2852
def __init__(self, suite, pattern):
2853
TestDecorator.__init__(self, suite)
2854
self.pattern = pattern
2855
self.filtered = False
2859
return iter(self._tests)
2860
self.filtered = True
2861
suites = split_suite_by_re(self, self.pattern)
2863
self.addTests(suites)
2864
return iter(self._tests)
2867
def partition_tests(suite, count):
2868
"""Partition suite into count lists of tests."""
2870
tests = list(iter_suite_tests(suite))
2871
tests_per_process = int(math.ceil(float(len(tests)) / count))
2872
for block in range(count):
2873
low_test = block * tests_per_process
2874
high_test = low_test + tests_per_process
2875
process_tests = tests[low_test:high_test]
2876
result.append(process_tests)
2880
def fork_for_tests(suite):
2881
"""Take suite and start up one runner per CPU by forking()
2883
:return: An iterable of TestCase-like objects which can each have
2884
run(result) called on them to feed tests to result, and
2885
cleanup() called on them to stop them/kill children/end threads.
2887
concurrency = local_concurrency()
2889
from subunit import TestProtocolClient, ProtocolTestCase
2890
class TestInOtherProcess(ProtocolTestCase):
2891
# Should be in subunit, I think. RBC.
2892
def __init__(self, stream, pid):
2893
ProtocolTestCase.__init__(self, stream)
2896
def run(self, result):
2898
ProtocolTestCase.run(self, result)
2900
os.waitpid(self.pid, os.WNOHANG)
2901
# print "pid %d finished" % finished_process
2903
test_blocks = partition_tests(suite, concurrency)
2904
for process_tests in test_blocks:
2905
process_suite = TestSuite()
2906
process_suite.addTests(process_tests)
2907
c2pread, c2pwrite = os.pipe()
2912
# Leave stderr and stdout open so we can see test noise
2913
# Close stdin so that the child goes away if it decides to
2914
# read from stdin (otherwise its a roulette to see what
2915
# child actually gets keystrokes for pdb etc.
2918
stream = os.fdopen(c2pwrite, 'wb', 0)
2919
subunit_result = TestProtocolClient(stream)
2920
process_suite.run(subunit_result)
2925
stream = os.fdopen(c2pread, 'rb', 1)
2926
test = TestInOtherProcess(stream, pid)
2931
def reinvoke_for_tests(suite):
2932
"""Take suite and start up one runner per CPU using subprocess().
2934
:return: An iterable of TestCase-like objects which can each have
2935
run(result) called on them to feed tests to result, and
2936
cleanup() called on them to stop them/kill children/end threads.
2938
concurrency = local_concurrency()
2940
from subunit import TestProtocolClient, ProtocolTestCase
2941
class TestInSubprocess(ProtocolTestCase):
2942
def __init__(self, process, name):
2943
ProtocolTestCase.__init__(self, process.stdout)
2944
self.process = process
2945
self.process.stdin.close()
2948
def run(self, result):
2950
ProtocolTestCase.run(self, result)
2953
os.unlink(self.name)
2954
# print "pid %d finished" % finished_process
2955
test_blocks = partition_tests(suite, concurrency)
2956
for process_tests in test_blocks:
2957
# ugly; currently reimplement rather than reuses TestCase methods.
2958
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
2959
if not os.path.isfile(bzr_path):
2960
# We are probably installed. Assume sys.argv is the right file
2961
bzr_path = sys.argv[0]
2962
fd, test_list_file_name = tempfile.mkstemp()
2963
test_list_file = os.fdopen(fd, 'wb', 1)
2964
for test in process_tests:
2965
test_list_file.write(test.id() + '\n')
2966
test_list_file.close()
2968
argv = [bzr_path, 'selftest', '--load-list', test_list_file_name,
2970
if '--no-plugins' in sys.argv:
2971
argv.append('--no-plugins')
2972
# stderr=STDOUT would be ideal, but until we prevent noise on
2973
# stderr it can interrupt the subunit protocol.
2974
process = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE,
2976
test = TestInSubprocess(process, test_list_file_name)
2979
os.unlink(test_list_file_name)
2984
def cpucount(content):
2985
lines = content.splitlines()
2986
prefix = 'processor'
2988
if line.startswith(prefix):
2989
concurrency = int(line[line.find(':')+1:]) + 1
2993
def local_concurrency():
2995
content = file('/proc/cpuinfo', 'rb').read()
2996
concurrency = cpucount(content)
2997
except Exception, e:
3002
class BZRTransformingResult(unittest.TestResult):
3004
def __init__(self, target):
3005
unittest.TestResult.__init__(self)
3006
self.result = target
3008
def startTest(self, test):
3009
self.result.startTest(test)
3011
def stopTest(self, test):
3012
self.result.stopTest(test)
3014
def addError(self, test, err):
3015
feature = self._error_looks_like('UnavailableFeature: ', err)
3016
if feature is not None:
3017
self.result.addNotSupported(test, feature)
3019
self.result.addError(test, err)
3021
def addFailure(self, test, err):
3022
known = self._error_looks_like('KnownFailure: ', err)
3023
if known is not None:
3024
self.result._addKnownFailure(test, [KnownFailure,
3025
KnownFailure(known), None])
3027
self.result.addFailure(test, err)
3029
def addSkip(self, test, reason):
3030
self.result.addSkip(test, reason)
3032
def addSuccess(self, test):
3033
self.result.addSuccess(test)
3035
def _error_looks_like(self, prefix, err):
3036
"""Deserialize exception and returns the stringify value."""
3040
if isinstance(exc, subunit.RemoteException):
3041
# stringify the exception gives access to the remote traceback
3042
# We search the last line for 'prefix'
3043
lines = str(exc).split('\n')
3045
last = lines[-2] # -1 is empty, final \n
3048
if last.startswith(prefix):
3049
value = last[len(prefix):]
2686
3053
# Controlled by "bzr selftest -E=..." option
added
removed
bzrlib/tests/__init__.py
