593
512
bench_history=None,
595
result_decorators=None,
597
"""Create a TextTestRunner.
599
:param result_decorators: An optional list of decorators to apply
600
to the result object being used by the runner. Decorators are
601
applied left to right - the first element in the list is the
604
# stream may know claim to know to write unicode strings, but in older
605
# pythons this goes sufficiently wrong that it is a bad idea. (
606
# specifically a built in file with encoding 'UTF-8' will still try
607
# to encode using ascii.
608
new_encoding = osutils.get_terminal_encoding()
609
codec = codecs.lookup(new_encoding)
610
if type(codec) is tuple:
614
encode = codec.encode
615
stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream)
616
stream.encoding = new_encoding
617
515
self.stream = unittest._WritelnDecorator(stream)
618
516
self.descriptions = descriptions
619
517
self.verbosity = verbosity
620
518
self._bench_history = bench_history
621
self._strict = strict
622
self._result_decorators = result_decorators or []
519
self.list_only = list_only
624
521
def run(self, test):
625
522
"Run the given test case or test suite."
523
startTime = time.time()
626
524
if self.verbosity == 1:
627
525
result_class = TextTestResult
628
526
elif self.verbosity >= 2:
629
527
result_class = VerboseTestResult
630
original_result = result_class(self.stream,
528
result = result_class(self.stream,
631
529
self.descriptions,
633
531
bench_history=self._bench_history,
532
num_tests=test.countTestCases(),
636
# Signal to result objects that look at stop early policy to stop,
637
original_result.stop_early = self.stop_on_failure
638
result = original_result
639
for decorator in self._result_decorators:
640
result = decorator(result)
641
result.stop_early = self.stop_on_failure
642
result.startTestRun()
534
result.stop_early = self.stop_on_failure
535
result.report_starting()
537
if self.verbosity >= 2:
538
self.stream.writeln("Listing tests only ...\n")
540
for t in iter_suite_tests(test):
541
self.stream.writeln("%s" % (t.id()))
543
actionTaken = "Listed"
647
# higher level code uses our extended protocol to determine
648
# what exit code to give.
649
return original_result
546
run = result.testsRun
548
stopTime = time.time()
549
timeTaken = stopTime - startTime
551
self.stream.writeln(result.separator2)
552
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
553
run, run != 1 and "s" or "", timeTaken))
554
self.stream.writeln()
555
if not result.wasSuccessful():
556
self.stream.write("FAILED (")
557
failed, errored = map(len, (result.failures, result.errors))
559
self.stream.write("failures=%d" % failed)
561
if failed: self.stream.write(", ")
562
self.stream.write("errors=%d" % errored)
563
if result.known_failure_count:
564
if failed or errored: self.stream.write(", ")
565
self.stream.write("known_failure_count=%d" %
566
result.known_failure_count)
567
self.stream.writeln(")")
569
if result.known_failure_count:
570
self.stream.writeln("OK (known_failures=%d)" %
571
result.known_failure_count)
573
self.stream.writeln("OK")
574
if result.skip_count > 0:
575
skipped = result.skip_count
576
self.stream.writeln('%d test%s skipped' %
577
(skipped, skipped != 1 and "s" or ""))
578
if result.unsupported:
579
for feature, count in sorted(result.unsupported.items()):
580
self.stream.writeln("Missing feature '%s' skipped %d tests." %
652
586
def iter_suite_tests(suite):
653
587
"""Return all tests in a suite, recursing through nested suites"""
654
if isinstance(suite, unittest.TestCase):
656
elif isinstance(suite, unittest.TestSuite):
588
for item in suite._tests:
589
if isinstance(item, unittest.TestCase):
591
elif isinstance(item, unittest.TestSuite):
658
592
for r in iter_suite_tests(item):
661
raise Exception('unknown type %r for object %r'
662
% (type(suite), suite))
665
TestSkipped = testtools.testcase.TestSkipped
595
raise Exception('unknown object %r inside test suite %r'
599
class TestSkipped(Exception):
600
"""Indicates that a test was intentionally skipped, rather than failing."""
668
603
class TestNotApplicable(TestSkipped):
793
752
_leaking_threads_tests = 0
794
753
_first_thread_leaker_id = None
795
754
_log_file_name = None
756
_keep_log_file = False
796
757
# record lsprof data when performing benchmark calls.
797
758
_gather_lsprof_in_benchmarks = False
759
attrs_to_keep = ('id', '_testMethodName', '_testMethodDoc',
760
'_log_contents', '_log_file_name', '_benchtime',
761
'_TestCase__testMethodName')
799
763
def __init__(self, methodName='testMethod'):
800
764
super(TestCase, self).__init__(methodName)
801
765
self._cleanups = []
802
self._directory_isolation = True
803
self.exception_handlers.insert(0,
804
(UnavailableFeature, self._do_unsupported_or_skip))
805
self.exception_handlers.insert(0,
806
(TestNotApplicable, self._do_not_applicable))
809
super(TestCase, self).setUp()
810
for feature in getattr(self, '_test_needs_features', []):
811
self.requireFeature(feature)
812
self._log_contents = None
813
self.addDetail("log", content.Content(content.ContentType("text",
814
"plain", {"charset": "utf8"}),
815
lambda:[self._get_log(keep_log_file=True)]))
768
unittest.TestCase.setUp(self)
816
769
self._cleanEnvironment()
817
770
self._silenceUI()
818
771
self._startLogFile()
819
772
self._benchcalls = []
820
773
self._benchtime = None
821
774
self._clear_hooks()
822
self._track_transports()
824
775
self._clear_debug_flags()
825
776
TestCase._active_threads = threading.activeCount()
826
777
self.addCleanup(self._check_leaked_threads)
831
pdb.Pdb().set_trace(sys._getframe().f_back)
780
absent_attr = object()
781
exc_info = getattr(self, '_exc_info', absent_attr)
782
if exc_info is absent_attr:
783
exc_info = getattr(self, '_TestCase__exc_info')
833
786
def _check_leaked_threads(self):
834
787
active = threading.activeCount()
835
788
leaked_threads = active - TestCase._active_threads
836
789
TestCase._active_threads = active
837
# If some tests make the number of threads *decrease*, we'll consider
838
# that they are just observing old threads dieing, not agressively kill
839
# random threads. So we don't report these tests as leaking. The risk
840
# is that we have false positives that way (the test see 2 threads
841
# going away but leak one) but it seems less likely than the actual
842
# false positives (the test see threads going away and does not leak).
843
if leaked_threads > 0:
844
791
TestCase._leaking_threads_tests += 1
845
792
if TestCase._first_thread_leaker_id is None:
846
793
TestCase._first_thread_leaker_id = self.id()
794
# we're not specifically told when all tests are finished.
795
# This will do. We use a function to avoid keeping a reference
796
# to a TestCase object.
797
atexit.register(_report_leaked_threads)
848
799
def _clear_debug_flags(self):
849
800
"""Prevent externally set debug flags affecting tests.
851
802
Tests that want to use debug flags can just set them in the
852
803
debug_flags set during setup/teardown.
854
# Start with a copy of the current debug flags we can safely modify.
855
self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
805
self._preserved_debug_flags = set(debug.debug_flags)
856
806
if 'allow_debug' not in selftest_debug_flags:
857
807
debug.debug_flags.clear()
858
if 'disable_lock_checks' not in selftest_debug_flags:
859
debug.debug_flags.add('strict_locks')
808
self.addCleanup(self._restore_debug_flags)
861
810
def _clear_hooks(self):
862
811
# prevent hooks affecting tests
863
self._preserved_hooks = {}
864
for key, factory in hooks.known_hooks.items():
865
parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
866
current_hooks = hooks.known_hooks_key_to_object(key)
867
self._preserved_hooks[parent] = (name, current_hooks)
813
import bzrlib.smart.client
814
import bzrlib.smart.server
815
self._preserved_hooks = {
816
bzrlib.branch.Branch: bzrlib.branch.Branch.hooks,
817
bzrlib.mutabletree.MutableTree: bzrlib.mutabletree.MutableTree.hooks,
818
bzrlib.smart.client._SmartClient: bzrlib.smart.client._SmartClient.hooks,
819
bzrlib.smart.server.SmartTCPServer: bzrlib.smart.server.SmartTCPServer.hooks,
820
bzrlib.commands.Command: bzrlib.commands.Command.hooks,
868
822
self.addCleanup(self._restoreHooks)
869
for key, factory in hooks.known_hooks.items():
870
parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
871
setattr(parent, name, factory())
872
# this hook should always be installed
873
request._install_hook()
875
def disable_directory_isolation(self):
876
"""Turn off directory isolation checks."""
877
self._directory_isolation = False
879
def enable_directory_isolation(self):
880
"""Enable directory isolation checks."""
881
self._directory_isolation = True
823
# reset all hooks to an empty instance of the appropriate type
824
bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()
825
bzrlib.smart.client._SmartClient.hooks = bzrlib.smart.client.SmartClientHooks()
826
bzrlib.smart.server.SmartTCPServer.hooks = bzrlib.smart.server.SmartServerHooks()
827
bzrlib.commands.Command.hooks = bzrlib.commands.CommandHooks()
883
829
def _silenceUI(self):
884
830
"""Turn off UI for duration of test"""
885
831
# by default the UI is off; tests can turn it on if they want it.
886
self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
888
def _check_locks(self):
889
"""Check that all lock take/release actions have been paired."""
890
# We always check for mismatched locks. If a mismatch is found, we
891
# fail unless -Edisable_lock_checks is supplied to selftest, in which
892
# case we just print a warning.
894
acquired_locks = [lock for action, lock in self._lock_actions
895
if action == 'acquired']
896
released_locks = [lock for action, lock in self._lock_actions
897
if action == 'released']
898
broken_locks = [lock for action, lock in self._lock_actions
899
if action == 'broken']
900
# trivially, given the tests for lock acquistion and release, if we
901
# have as many in each list, it should be ok. Some lock tests also
902
# break some locks on purpose and should be taken into account by
903
# considering that breaking a lock is just a dirty way of releasing it.
904
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
905
message = ('Different number of acquired and '
906
'released or broken locks. (%s, %s + %s)' %
907
(acquired_locks, released_locks, broken_locks))
908
if not self._lock_check_thorough:
909
# Rather than fail, just warn
910
print "Broken test %s: %s" % (self, message)
914
def _track_locks(self):
915
"""Track lock activity during tests."""
916
self._lock_actions = []
917
if 'disable_lock_checks' in selftest_debug_flags:
918
self._lock_check_thorough = False
920
self._lock_check_thorough = True
922
self.addCleanup(self._check_locks)
923
_mod_lock.Lock.hooks.install_named_hook('lock_acquired',
924
self._lock_acquired, None)
925
_mod_lock.Lock.hooks.install_named_hook('lock_released',
926
self._lock_released, None)
927
_mod_lock.Lock.hooks.install_named_hook('lock_broken',
928
self._lock_broken, None)
930
def _lock_acquired(self, result):
931
self._lock_actions.append(('acquired', result))
933
def _lock_released(self, result):
934
self._lock_actions.append(('released', result))
936
def _lock_broken(self, result):
937
self._lock_actions.append(('broken', result))
939
def permit_dir(self, name):
940
"""Permit a directory to be used by this test. See permit_url."""
941
name_transport = get_transport(name)
942
self.permit_url(name)
943
self.permit_url(name_transport.base)
945
def permit_url(self, url):
946
"""Declare that url is an ok url to use in this test.
948
Do this for memory transports, temporary test directory etc.
950
Do not do this for the current working directory, /tmp, or any other
951
preexisting non isolated url.
953
if not url.endswith('/'):
955
self._bzr_selftest_roots.append(url)
957
def permit_source_tree_branch_repo(self):
958
"""Permit the source tree bzr is running from to be opened.
960
Some code such as bzrlib.version attempts to read from the bzr branch
961
that bzr is executing from (if any). This method permits that directory
962
to be used in the test suite.
964
path = self.get_source_path()
965
self.record_directory_isolation()
968
workingtree.WorkingTree.open(path)
969
except (errors.NotBranchError, errors.NoWorkingTree):
972
self.enable_directory_isolation()
974
def _preopen_isolate_transport(self, transport):
975
"""Check that all transport openings are done in the test work area."""
976
while isinstance(transport, pathfilter.PathFilteringTransport):
977
# Unwrap pathfiltered transports
978
transport = transport.server.backing_transport.clone(
979
transport._filter('.'))
981
# ReadonlySmartTCPServer_for_testing decorates the backing transport
982
# urls it is given by prepending readonly+. This is appropriate as the
983
# client shouldn't know that the server is readonly (or not readonly).
984
# We could register all servers twice, with readonly+ prepending, but
985
# that makes for a long list; this is about the same but easier to
987
if url.startswith('readonly+'):
988
url = url[len('readonly+'):]
989
self._preopen_isolate_url(url)
991
def _preopen_isolate_url(self, url):
992
if not self._directory_isolation:
994
if self._directory_isolation == 'record':
995
self._bzr_selftest_roots.append(url)
997
# This prevents all transports, including e.g. sftp ones backed on disk
998
# from working unless they are explicitly granted permission. We then
999
# depend on the code that sets up test transports to check that they are
1000
# appropriately isolated and enable their use by calling
1001
# self.permit_transport()
1002
if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1003
raise errors.BzrError("Attempt to escape test isolation: %r %r"
1004
% (url, self._bzr_selftest_roots))
1006
def record_directory_isolation(self):
1007
"""Gather accessed directories to permit later access.
1009
This is used for tests that access the branch bzr is running from.
1011
self._directory_isolation = "record"
1013
def start_server(self, transport_server, backing_server=None):
1014
"""Start transport_server for this test.
1016
This starts the server, registers a cleanup for it and permits the
1017
server's urls to be used.
1019
if backing_server is None:
1020
transport_server.start_server()
1022
transport_server.start_server(backing_server)
1023
self.addCleanup(transport_server.stop_server)
1024
# Obtain a real transport because if the server supplies a password, it
1025
# will be hidden from the base on the client side.
1026
t = get_transport(transport_server.get_url())
1027
# Some transport servers effectively chroot the backing transport;
1028
# others like SFTPServer don't - users of the transport can walk up the
1029
# transport to read the entire backing transport. This wouldn't matter
1030
# except that the workdir tests are given - and that they expect the
1031
# server's url to point at - is one directory under the safety net. So
1032
# Branch operations into the transport will attempt to walk up one
1033
# directory. Chrooting all servers would avoid this but also mean that
1034
# we wouldn't be testing directly against non-root urls. Alternatively
1035
# getting the test framework to start the server with a backing server
1036
# at the actual safety net directory would work too, but this then
1037
# means that the self.get_url/self.get_transport methods would need
1038
# to transform all their results. On balance its cleaner to handle it
1039
# here, and permit a higher url when we have one of these transports.
1040
if t.base.endswith('/work/'):
1041
# we have safety net/test root/work
1042
t = t.clone('../..')
1043
elif isinstance(transport_server, server.SmartTCPServer_for_testing):
1044
# The smart server adds a path similar to work, which is traversed
1045
# up from by the client. But the server is chrooted - the actual
1046
# backing transport is not escaped from, and VFS requests to the
1047
# root will error (because they try to escape the chroot).
1049
while t2.base != t.base:
1052
self.permit_url(t.base)
1054
def _track_transports(self):
1055
"""Install checks for transport usage."""
1056
# TestCase has no safe place it can write to.
1057
self._bzr_selftest_roots = []
1058
# Currently the easiest way to be sure that nothing is going on is to
1059
# hook into bzr dir opening. This leaves a small window of error for
1060
# transport tests, but they are well known, and we can improve on this
1062
bzrdir.BzrDir.hooks.install_named_hook("pre_open",
1063
self._preopen_isolate_transport, "Check bzr directories are safe.")
832
saved = ui.ui_factory
834
ui.ui_factory = saved
835
ui.ui_factory = ui.SilentUIFactory()
836
self.addCleanup(_restore)
1065
838
def _ndiff_strings(self, a, b):
1066
839
"""Return ndiff between two strings containing lines.
1566
1293
def _do_skip(self, result, reason):
1567
1294
addSkip = getattr(result, 'addSkip', None)
1568
1295
if not callable(addSkip):
1569
result.addSuccess(result)
1296
result.addError(self, self.exc_info())
1571
1298
addSkip(self, reason)
1574
def _do_known_failure(self, result, e):
1575
err = sys.exc_info()
1576
addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1577
if addExpectedFailure is not None:
1578
addExpectedFailure(self, err)
1580
result.addSuccess(self)
1583
def _do_not_applicable(self, result, e):
1585
reason = 'No reason given'
1588
addNotApplicable = getattr(result, 'addNotApplicable', None)
1589
if addNotApplicable is not None:
1590
result.addNotApplicable(self, reason)
1592
self._do_skip(result, reason)
1595
def _do_unsupported_or_skip(self, result, e):
1597
addNotSupported = getattr(result, 'addNotSupported', None)
1598
if addNotSupported is not None:
1599
result.addNotSupported(self, reason)
1601
self._do_skip(result, reason)
1300
def run(self, result=None):
1301
if result is None: result = self.defaultTestResult()
1302
for feature in getattr(self, '_test_needs_features', []):
1303
if not feature.available():
1304
result.startTest(self)
1305
if getattr(result, 'addNotSupported', None):
1306
result.addNotSupported(self, feature)
1308
result.addSuccess(self)
1309
result.stopTest(self)
1313
result.startTest(self)
1314
absent_attr = object()
1316
method_name = getattr(self, '_testMethodName', absent_attr)
1317
if method_name is absent_attr:
1319
method_name = getattr(self, '_TestCase__testMethodName')
1320
testMethod = getattr(self, method_name)
1324
except KeyboardInterrupt:
1326
except TestSkipped, e:
1327
self._do_skip(result, e.args[0])
1331
result.addError(self, self.exc_info())
1338
except self.failureException:
1339
result.addFailure(self, self.exc_info())
1340
except TestSkipped, e:
1342
reason = "No reason given."
1345
self._do_skip(result, reason)
1346
except KeyboardInterrupt:
1349
result.addError(self, self.exc_info())
1353
except KeyboardInterrupt:
1356
result.addError(self, self.exc_info())
1358
if ok: result.addSuccess(self)
1360
result.stopTest(self)
1362
except TestNotApplicable:
1363
# Not moved from the result [yet].
1365
except KeyboardInterrupt:
1369
absent_attr = object()
1370
for attr_name in self.attrs_to_keep:
1371
attr = getattr(self, attr_name, absent_attr)
1372
if attr is not absent_attr:
1373
saved_attrs[attr_name] = attr
1374
self.__dict__ = saved_attrs
1378
self._log_contents = ''
1379
unittest.TestCase.tearDown(self)
1603
1381
def time(self, callable, *args, **kwargs):
1604
1382
"""Run callable and accrue the time it takes to the benchmark time.
2903
2614
if runner_class is None:
2904
2615
runner_class = TextTestRunner
2907
runner = runner_class(stream=stream,
2616
runner = runner_class(stream=sys.stdout,
2908
2617
descriptions=0,
2909
2618
verbosity=verbosity,
2910
2619
bench_history=bench_history,
2912
result_decorators=result_decorators,
2620
list_only=list_only,
2914
2622
runner.stop_on_failure=stop_on_failure
2915
# built in decorator factories:
2917
random_order(random_seed, runner),
2918
exclude_tests(exclude_pattern),
2920
if matching_tests_first:
2921
decorators.append(tests_first(pattern))
2623
# Initialise the random number generator and display the seed used.
2624
# We convert the seed to a long to make it reuseable across invocations.
2625
random_order = False
2626
if random_seed is not None:
2628
if random_seed == "now":
2629
random_seed = long(time.time())
2631
# Convert the seed to a long if we can
2633
random_seed = long(random_seed)
2636
runner.stream.writeln("Randomizing test order using seed %s\n" %
2638
random.seed(random_seed)
2639
# Customise the list of tests if requested
2640
if exclude_pattern is not None:
2641
suite = exclude_tests_by_re(suite, exclude_pattern)
2643
order_changer = randomize_suite
2923
decorators.append(filter_tests(pattern))
2924
if suite_decorators:
2925
decorators.extend(suite_decorators)
2926
# tell the result object how many tests will be running: (except if
2927
# --parallel=fork is being used. Robert said he will provide a better
2928
# progress design later -- vila 20090817)
2929
if fork_decorator not in decorators:
2930
decorators.append(CountingDecorator)
2931
for decorator in decorators:
2932
suite = decorator(suite)
2934
# Done after test suite decoration to allow randomisation etc
2935
# to take effect, though that is of marginal benefit.
2937
stream.write("Listing tests only ...\n")
2938
for t in iter_suite_tests(suite):
2939
stream.write("%s\n" % (t.id()))
2645
order_changer = preserve_input
2646
if pattern != '.*' or random_order:
2647
if matching_tests_first:
2648
suites = map(order_changer, split_suite_by_re(suite, pattern))
2649
suite = TestUtil.TestSuite(suites)
2651
suite = order_changer(filter_suite_by_re(suite, pattern))
2941
2653
result = runner.run(suite)
2943
2656
return result.wasStrictlySuccessful()
2945
return result.wasSuccessful()
2948
# A registry where get() returns a suite decorator.
2949
parallel_registry = registry.Registry()
2952
def fork_decorator(suite):
2953
concurrency = osutils.local_concurrency()
2954
if concurrency == 1:
2956
from testtools import ConcurrentTestSuite
2957
return ConcurrentTestSuite(suite, fork_for_tests)
2958
parallel_registry.register('fork', fork_decorator)
2961
def subprocess_decorator(suite):
2962
concurrency = osutils.local_concurrency()
2963
if concurrency == 1:
2965
from testtools import ConcurrentTestSuite
2966
return ConcurrentTestSuite(suite, reinvoke_for_tests)
2967
parallel_registry.register('subprocess', subprocess_decorator)
2970
def exclude_tests(exclude_pattern):
2971
"""Return a test suite decorator that excludes tests."""
2972
if exclude_pattern is None:
2973
return identity_decorator
2974
def decorator(suite):
2975
return ExcludeDecorator(suite, exclude_pattern)
2979
def filter_tests(pattern):
2981
return identity_decorator
2982
def decorator(suite):
2983
return FilterTestsDecorator(suite, pattern)
2987
def random_order(random_seed, runner):
2988
"""Return a test suite decorator factory for randomising tests order.
2990
:param random_seed: now, a string which casts to a long, or a long.
2991
:param runner: A test runner with a stream attribute to report on.
2993
if random_seed is None:
2994
return identity_decorator
2995
def decorator(suite):
2996
return RandomDecorator(suite, random_seed, runner.stream)
3000
def tests_first(pattern):
3002
return identity_decorator
3003
def decorator(suite):
3004
return TestFirstDecorator(suite, pattern)
3008
def identity_decorator(suite):
3013
class TestDecorator(TestSuite):
3014
"""A decorator for TestCase/TestSuite objects.
3016
Usually, subclasses should override __iter__(used when flattening test
3017
suites), which we do to filter, reorder, parallelise and so on, run() and
3021
def __init__(self, suite):
3022
TestSuite.__init__(self)
3025
def countTestCases(self):
3028
cases += test.countTestCases()
3035
def run(self, result):
3036
# Use iteration on self, not self._tests, to allow subclasses to hook
3039
if result.shouldStop:
3045
class CountingDecorator(TestDecorator):
3046
"""A decorator which calls result.progress(self.countTestCases)."""
3048
def run(self, result):
3049
progress_method = getattr(result, 'progress', None)
3050
if callable(progress_method):
3051
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3052
return super(CountingDecorator, self).run(result)
3055
class ExcludeDecorator(TestDecorator):
3056
"""A decorator which excludes test matching an exclude pattern."""
3058
def __init__(self, suite, exclude_pattern):
3059
TestDecorator.__init__(self, suite)
3060
self.exclude_pattern = exclude_pattern
3061
self.excluded = False
3065
return iter(self._tests)
3066
self.excluded = True
3067
suite = exclude_tests_by_re(self, self.exclude_pattern)
3069
self.addTests(suite)
3070
return iter(self._tests)
3073
class FilterTestsDecorator(TestDecorator):
3074
"""A decorator which filters tests to those matching a pattern."""
3076
def __init__(self, suite, pattern):
3077
TestDecorator.__init__(self, suite)
3078
self.pattern = pattern
3079
self.filtered = False
3083
return iter(self._tests)
3084
self.filtered = True
3085
suite = filter_suite_by_re(self, self.pattern)
3087
self.addTests(suite)
3088
return iter(self._tests)
3091
class RandomDecorator(TestDecorator):
3092
"""A decorator which randomises the order of its tests."""
3094
def __init__(self, suite, random_seed, stream):
3095
TestDecorator.__init__(self, suite)
3096
self.random_seed = random_seed
3097
self.randomised = False
3098
self.stream = stream
3102
return iter(self._tests)
3103
self.randomised = True
3104
self.stream.write("Randomizing test order using seed %s\n\n" %
3105
(self.actual_seed()))
3106
# Initialise the random number generator.
3107
random.seed(self.actual_seed())
3108
suite = randomize_suite(self)
3110
self.addTests(suite)
3111
return iter(self._tests)
3113
def actual_seed(self):
3114
if self.random_seed == "now":
3115
# We convert the seed to a long to make it reuseable across
3116
# invocations (because the user can reenter it).
3117
self.random_seed = long(time.time())
3119
# Convert the seed to a long if we can
3121
self.random_seed = long(self.random_seed)
3124
return self.random_seed
3127
class TestFirstDecorator(TestDecorator):
3128
"""A decorator which moves named tests to the front."""
3130
def __init__(self, suite, pattern):
3131
TestDecorator.__init__(self, suite)
3132
self.pattern = pattern
3133
self.filtered = False
3137
return iter(self._tests)
3138
self.filtered = True
3139
suites = split_suite_by_re(self, self.pattern)
3141
self.addTests(suites)
3142
return iter(self._tests)
3145
def partition_tests(suite, count):
3146
"""Partition suite into count lists of tests."""
3148
tests = list(iter_suite_tests(suite))
3149
tests_per_process = int(math.ceil(float(len(tests)) / count))
3150
for block in range(count):
3151
low_test = block * tests_per_process
3152
high_test = low_test + tests_per_process
3153
process_tests = tests[low_test:high_test]
3154
result.append(process_tests)
3158
def fork_for_tests(suite):
3159
"""Take suite and start up one runner per CPU by forking()
3161
:return: An iterable of TestCase-like objects which can each have
3162
run(result) called on them to feed tests to result.
3164
concurrency = osutils.local_concurrency()
3166
from subunit import TestProtocolClient, ProtocolTestCase
3167
from subunit.test_results import AutoTimingTestResultDecorator
3168
class TestInOtherProcess(ProtocolTestCase):
3169
# Should be in subunit, I think. RBC.
3170
def __init__(self, stream, pid):
3171
ProtocolTestCase.__init__(self, stream)
3174
def run(self, result):
3176
ProtocolTestCase.run(self, result)
3178
os.waitpid(self.pid, os.WNOHANG)
3180
test_blocks = partition_tests(suite, concurrency)
3181
for process_tests in test_blocks:
3182
process_suite = TestSuite()
3183
process_suite.addTests(process_tests)
3184
c2pread, c2pwrite = os.pipe()
3189
# Leave stderr and stdout open so we can see test noise
3190
# Close stdin so that the child goes away if it decides to
3191
# read from stdin (otherwise its a roulette to see what
3192
# child actually gets keystrokes for pdb etc).
3195
stream = os.fdopen(c2pwrite, 'wb', 1)
3196
subunit_result = AutoTimingTestResultDecorator(
3197
TestProtocolClient(stream))
3198
process_suite.run(subunit_result)
3203
stream = os.fdopen(c2pread, 'rb', 1)
3204
test = TestInOtherProcess(stream, pid)
3209
def reinvoke_for_tests(suite):
3210
"""Take suite and start up one runner per CPU using subprocess().
3212
:return: An iterable of TestCase-like objects which can each have
3213
run(result) called on them to feed tests to result.
3215
concurrency = osutils.local_concurrency()
3217
from subunit import ProtocolTestCase
3218
class TestInSubprocess(ProtocolTestCase):
3219
def __init__(self, process, name):
3220
ProtocolTestCase.__init__(self, process.stdout)
3221
self.process = process
3222
self.process.stdin.close()
3225
def run(self, result):
3227
ProtocolTestCase.run(self, result)
3230
os.unlink(self.name)
3231
# print "pid %d finished" % finished_process
3232
test_blocks = partition_tests(suite, concurrency)
3233
for process_tests in test_blocks:
3234
# ugly; currently reimplement rather than reuses TestCase methods.
3235
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3236
if not os.path.isfile(bzr_path):
3237
# We are probably installed. Assume sys.argv is the right file
3238
bzr_path = sys.argv[0]
3239
bzr_path = [bzr_path]
3240
if sys.platform == "win32":
3241
# if we're on windows, we can't execute the bzr script directly
3242
bzr_path = [sys.executable] + bzr_path
3243
fd, test_list_file_name = tempfile.mkstemp()
3244
test_list_file = os.fdopen(fd, 'wb', 1)
3245
for test in process_tests:
3246
test_list_file.write(test.id() + '\n')
3247
test_list_file.close()
3249
argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
3251
if '--no-plugins' in sys.argv:
3252
argv.append('--no-plugins')
3253
# stderr=STDOUT would be ideal, but until we prevent noise on
3254
# stderr it can interrupt the subunit protocol.
3255
process = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE,
3257
test = TestInSubprocess(process, test_list_file_name)
3260
os.unlink(test_list_file_name)
3265
class ForwardingResult(unittest.TestResult):
3267
def __init__(self, target):
3268
unittest.TestResult.__init__(self)
3269
self.result = target
3271
def startTest(self, test):
3272
self.result.startTest(test)
3274
def stopTest(self, test):
3275
self.result.stopTest(test)
3277
def startTestRun(self):
3278
self.result.startTestRun()
3280
def stopTestRun(self):
3281
self.result.stopTestRun()
3283
def addSkip(self, test, reason):
3284
self.result.addSkip(test, reason)
3286
def addSuccess(self, test):
3287
self.result.addSuccess(test)
3289
def addError(self, test, err):
3290
self.result.addError(test, err)
3292
def addFailure(self, test, err):
3293
self.result.addFailure(test, err)
3294
ForwardingResult = testtools.ExtendedToOriginalDecorator
3297
class ProfileResult(ForwardingResult):
3298
"""Generate profiling data for all activity between start and success.
3300
The profile data is appended to the test's _benchcalls attribute and can
3301
be accessed by the forwarded-to TestResult.
3303
While it might be cleaner do accumulate this in stopTest, addSuccess is
3304
where our existing output support for lsprof is, and this class aims to
3305
fit in with that: while it could be moved it's not necessary to accomplish
3306
test profiling, nor would it be dramatically cleaner.
3309
def startTest(self, test):
3310
self.profiler = bzrlib.lsprof.BzrProfiler()
3311
self.profiler.start()
3312
ForwardingResult.startTest(self, test)
3314
def addSuccess(self, test):
3315
stats = self.profiler.stop()
3317
calls = test._benchcalls
3318
except AttributeError:
3319
test._benchcalls = []
3320
calls = test._benchcalls
3321
calls.append(((test.id(), "", ""), stats))
3322
ForwardingResult.addSuccess(self, test)
3324
def stopTest(self, test):
3325
ForwardingResult.stopTest(self, test)
3326
self.profiler = None
2658
return result.wasSuccessful()
3329
2661
# Controlled by "bzr selftest -E=..." option
3330
# Currently supported:
3331
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3332
# preserves any flags supplied at the command line.
3333
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3334
# rather than failing tests. And no longer raise
3335
# LockContention when fctnl locks are not being used
3336
# with proper exclusion rules.
3337
2662
selftest_debug_flags = set()
3566
2874
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3569
def _test_suite_testmod_names():
3570
"""Return the standard list of test module names to test."""
3573
'bzrlib.tests.blackbox',
3574
'bzrlib.tests.commands',
3575
'bzrlib.tests.per_branch',
3576
'bzrlib.tests.per_bzrdir',
3577
'bzrlib.tests.per_foreign_vcs',
3578
'bzrlib.tests.per_interrepository',
3579
'bzrlib.tests.per_intertree',
3580
'bzrlib.tests.per_inventory',
3581
'bzrlib.tests.per_interbranch',
3582
'bzrlib.tests.per_lock',
3583
'bzrlib.tests.per_merger',
3584
'bzrlib.tests.per_transport',
3585
'bzrlib.tests.per_tree',
3586
'bzrlib.tests.per_pack_repository',
3587
'bzrlib.tests.per_repository',
3588
'bzrlib.tests.per_repository_chk',
3589
'bzrlib.tests.per_repository_reference',
3590
'bzrlib.tests.per_uifactory',
3591
'bzrlib.tests.per_versionedfile',
3592
'bzrlib.tests.per_workingtree',
3593
'bzrlib.tests.test__annotator',
3594
'bzrlib.tests.test__bencode',
3595
'bzrlib.tests.test__chk_map',
3596
'bzrlib.tests.test__dirstate_helpers',
3597
'bzrlib.tests.test__groupcompress',
3598
'bzrlib.tests.test__known_graph',
3599
'bzrlib.tests.test__rio',
3600
'bzrlib.tests.test__simple_set',
3601
'bzrlib.tests.test__static_tuple',
3602
'bzrlib.tests.test__walkdirs_win32',
3603
'bzrlib.tests.test_ancestry',
3604
'bzrlib.tests.test_annotate',
3605
'bzrlib.tests.test_api',
3606
'bzrlib.tests.test_atomicfile',
3607
'bzrlib.tests.test_bad_files',
3608
'bzrlib.tests.test_bisect_multi',
3609
'bzrlib.tests.test_branch',
3610
'bzrlib.tests.test_branchbuilder',
3611
'bzrlib.tests.test_btree_index',
3612
'bzrlib.tests.test_bugtracker',
3613
'bzrlib.tests.test_bundle',
3614
'bzrlib.tests.test_bzrdir',
3615
'bzrlib.tests.test__chunks_to_lines',
3616
'bzrlib.tests.test_cache_utf8',
3617
'bzrlib.tests.test_chk_map',
3618
'bzrlib.tests.test_chk_serializer',
3619
'bzrlib.tests.test_chunk_writer',
3620
'bzrlib.tests.test_clean_tree',
3621
'bzrlib.tests.test_cleanup',
3622
'bzrlib.tests.test_commands',
3623
'bzrlib.tests.test_commit',
3624
'bzrlib.tests.test_commit_merge',
3625
'bzrlib.tests.test_config',
3626
'bzrlib.tests.test_conflicts',
3627
'bzrlib.tests.test_counted_lock',
3628
'bzrlib.tests.test_crash',
3629
'bzrlib.tests.test_decorators',
3630
'bzrlib.tests.test_delta',
3631
'bzrlib.tests.test_debug',
3632
'bzrlib.tests.test_deprecated_graph',
3633
'bzrlib.tests.test_diff',
3634
'bzrlib.tests.test_directory_service',
3635
'bzrlib.tests.test_dirstate',
3636
'bzrlib.tests.test_email_message',
3637
'bzrlib.tests.test_eol_filters',
3638
'bzrlib.tests.test_errors',
3639
'bzrlib.tests.test_export',
3640
'bzrlib.tests.test_extract',
3641
'bzrlib.tests.test_fetch',
3642
'bzrlib.tests.test_fifo_cache',
3643
'bzrlib.tests.test_filters',
3644
'bzrlib.tests.test_ftp_transport',
3645
'bzrlib.tests.test_foreign',
3646
'bzrlib.tests.test_generate_docs',
3647
'bzrlib.tests.test_generate_ids',
3648
'bzrlib.tests.test_globbing',
3649
'bzrlib.tests.test_gpg',
3650
'bzrlib.tests.test_graph',
3651
'bzrlib.tests.test_groupcompress',
3652
'bzrlib.tests.test_hashcache',
3653
'bzrlib.tests.test_help',
3654
'bzrlib.tests.test_hooks',
3655
'bzrlib.tests.test_http',
3656
'bzrlib.tests.test_http_response',
3657
'bzrlib.tests.test_https_ca_bundle',
3658
'bzrlib.tests.test_identitymap',
3659
'bzrlib.tests.test_ignores',
3660
'bzrlib.tests.test_index',
3661
'bzrlib.tests.test_info',
3662
'bzrlib.tests.test_inv',
3663
'bzrlib.tests.test_inventory_delta',
3664
'bzrlib.tests.test_knit',
3665
'bzrlib.tests.test_lazy_import',
3666
'bzrlib.tests.test_lazy_regex',
3667
'bzrlib.tests.test_lock',
3668
'bzrlib.tests.test_lockable_files',
3669
'bzrlib.tests.test_lockdir',
3670
'bzrlib.tests.test_log',
3671
'bzrlib.tests.test_lru_cache',
3672
'bzrlib.tests.test_lsprof',
3673
'bzrlib.tests.test_mail_client',
3674
'bzrlib.tests.test_memorytree',
3675
'bzrlib.tests.test_merge',
3676
'bzrlib.tests.test_merge3',
3677
'bzrlib.tests.test_merge_core',
3678
'bzrlib.tests.test_merge_directive',
3679
'bzrlib.tests.test_missing',
3680
'bzrlib.tests.test_msgeditor',
3681
'bzrlib.tests.test_multiparent',
3682
'bzrlib.tests.test_mutabletree',
3683
'bzrlib.tests.test_nonascii',
3684
'bzrlib.tests.test_options',
3685
'bzrlib.tests.test_osutils',
3686
'bzrlib.tests.test_osutils_encodings',
3687
'bzrlib.tests.test_pack',
3688
'bzrlib.tests.test_patch',
3689
'bzrlib.tests.test_patches',
3690
'bzrlib.tests.test_permissions',
3691
'bzrlib.tests.test_plugins',
3692
'bzrlib.tests.test_progress',
3693
'bzrlib.tests.test_read_bundle',
3694
'bzrlib.tests.test_reconcile',
3695
'bzrlib.tests.test_reconfigure',
3696
'bzrlib.tests.test_registry',
3697
'bzrlib.tests.test_remote',
3698
'bzrlib.tests.test_rename_map',
3699
'bzrlib.tests.test_repository',
3700
'bzrlib.tests.test_revert',
3701
'bzrlib.tests.test_revision',
3702
'bzrlib.tests.test_revisionspec',
3703
'bzrlib.tests.test_revisiontree',
3704
'bzrlib.tests.test_rio',
3705
'bzrlib.tests.test_rules',
3706
'bzrlib.tests.test_sampler',
3707
'bzrlib.tests.test_script',
3708
'bzrlib.tests.test_selftest',
3709
'bzrlib.tests.test_serializer',
3710
'bzrlib.tests.test_setup',
3711
'bzrlib.tests.test_sftp_transport',
3712
'bzrlib.tests.test_shelf',
3713
'bzrlib.tests.test_shelf_ui',
3714
'bzrlib.tests.test_smart',
3715
'bzrlib.tests.test_smart_add',
3716
'bzrlib.tests.test_smart_request',
3717
'bzrlib.tests.test_smart_transport',
3718
'bzrlib.tests.test_smtp_connection',
3719
'bzrlib.tests.test_source',
3720
'bzrlib.tests.test_ssh_transport',
3721
'bzrlib.tests.test_status',
3722
'bzrlib.tests.test_store',
3723
'bzrlib.tests.test_strace',
3724
'bzrlib.tests.test_subsume',
3725
'bzrlib.tests.test_switch',
3726
'bzrlib.tests.test_symbol_versioning',
3727
'bzrlib.tests.test_tag',
3728
'bzrlib.tests.test_testament',
3729
'bzrlib.tests.test_textfile',
3730
'bzrlib.tests.test_textmerge',
3731
'bzrlib.tests.test_timestamp',
3732
'bzrlib.tests.test_trace',
3733
'bzrlib.tests.test_transactions',
3734
'bzrlib.tests.test_transform',
3735
'bzrlib.tests.test_transport',
3736
'bzrlib.tests.test_transport_log',
3737
'bzrlib.tests.test_tree',
3738
'bzrlib.tests.test_treebuilder',
3739
'bzrlib.tests.test_tsort',
3740
'bzrlib.tests.test_tuned_gzip',
3741
'bzrlib.tests.test_ui',
3742
'bzrlib.tests.test_uncommit',
3743
'bzrlib.tests.test_upgrade',
3744
'bzrlib.tests.test_upgrade_stacked',
3745
'bzrlib.tests.test_urlutils',
3746
'bzrlib.tests.test_version',
3747
'bzrlib.tests.test_version_info',
3748
'bzrlib.tests.test_weave',
3749
'bzrlib.tests.test_whitebox',
3750
'bzrlib.tests.test_win32utils',
3751
'bzrlib.tests.test_workingtree',
3752
'bzrlib.tests.test_workingtree_4',
3753
'bzrlib.tests.test_wsgi',
3754
'bzrlib.tests.test_xml',
3758
def _test_suite_modules_to_doctest():
3759
"""Return the list of modules to doctest."""
3762
'bzrlib.branchbuilder',
3763
'bzrlib.decorators',
3766
'bzrlib.iterablefile',
3770
'bzrlib.symbol_versioning',
3773
'bzrlib.version_info_formats.format_custom',
3777
2877
def test_suite(keep_only=None, starting_with=None):
3778
2878
"""Build and return TestSuite for the whole of bzrlib.
3785
2885
This function can be replaced if you need to change the default test
3786
2886
suite on a global basis, but it is not encouraged.
2890
'bzrlib.tests.blackbox',
2891
'bzrlib.tests.branch_implementations',
2892
'bzrlib.tests.bzrdir_implementations',
2893
'bzrlib.tests.commands',
2894
'bzrlib.tests.interrepository_implementations',
2895
'bzrlib.tests.intertree_implementations',
2896
'bzrlib.tests.inventory_implementations',
2897
'bzrlib.tests.per_interbranch',
2898
'bzrlib.tests.per_lock',
2899
'bzrlib.tests.per_repository',
2900
'bzrlib.tests.per_repository_reference',
2901
'bzrlib.tests.test__dirstate_helpers',
2902
'bzrlib.tests.test__walkdirs_win32',
2903
'bzrlib.tests.test_ancestry',
2904
'bzrlib.tests.test_annotate',
2905
'bzrlib.tests.test_api',
2906
'bzrlib.tests.test_atomicfile',
2907
'bzrlib.tests.test_bad_files',
2908
'bzrlib.tests.test_bisect_multi',
2909
'bzrlib.tests.test_branch',
2910
'bzrlib.tests.test_branchbuilder',
2911
'bzrlib.tests.test_btree_index',
2912
'bzrlib.tests.test_bugtracker',
2913
'bzrlib.tests.test_bundle',
2914
'bzrlib.tests.test_bzrdir',
2915
'bzrlib.tests.test_cache_utf8',
2916
'bzrlib.tests.test_chunk_writer',
2917
'bzrlib.tests.test__chunks_to_lines',
2918
'bzrlib.tests.test_commands',
2919
'bzrlib.tests.test_commit',
2920
'bzrlib.tests.test_commit_merge',
2921
'bzrlib.tests.test_config',
2922
'bzrlib.tests.test_conflicts',
2923
'bzrlib.tests.test_counted_lock',
2924
'bzrlib.tests.test_decorators',
2925
'bzrlib.tests.test_delta',
2926
'bzrlib.tests.test_debug',
2927
'bzrlib.tests.test_deprecated_graph',
2928
'bzrlib.tests.test_diff',
2929
'bzrlib.tests.test_directory_service',
2930
'bzrlib.tests.test_dirstate',
2931
'bzrlib.tests.test_email_message',
2932
'bzrlib.tests.test_errors',
2933
'bzrlib.tests.test_export',
2934
'bzrlib.tests.test_extract',
2935
'bzrlib.tests.test_fetch',
2936
'bzrlib.tests.test_fifo_cache',
2937
'bzrlib.tests.test_ftp_transport',
2938
'bzrlib.tests.test_foreign',
2939
'bzrlib.tests.test_generate_docs',
2940
'bzrlib.tests.test_generate_ids',
2941
'bzrlib.tests.test_globbing',
2942
'bzrlib.tests.test_gpg',
2943
'bzrlib.tests.test_graph',
2944
'bzrlib.tests.test_hashcache',
2945
'bzrlib.tests.test_help',
2946
'bzrlib.tests.test_hooks',
2947
'bzrlib.tests.test_http',
2948
'bzrlib.tests.test_http_implementations',
2949
'bzrlib.tests.test_http_response',
2950
'bzrlib.tests.test_https_ca_bundle',
2951
'bzrlib.tests.test_identitymap',
2952
'bzrlib.tests.test_ignores',
2953
'bzrlib.tests.test_index',
2954
'bzrlib.tests.test_info',
2955
'bzrlib.tests.test_inv',
2956
'bzrlib.tests.test_knit',
2957
'bzrlib.tests.test_lazy_import',
2958
'bzrlib.tests.test_lazy_regex',
2959
'bzrlib.tests.test_lockable_files',
2960
'bzrlib.tests.test_lockdir',
2961
'bzrlib.tests.test_log',
2962
'bzrlib.tests.test_lru_cache',
2963
'bzrlib.tests.test_lsprof',
2964
'bzrlib.tests.test_mail_client',
2965
'bzrlib.tests.test_memorytree',
2966
'bzrlib.tests.test_merge',
2967
'bzrlib.tests.test_merge3',
2968
'bzrlib.tests.test_merge_core',
2969
'bzrlib.tests.test_merge_directive',
2970
'bzrlib.tests.test_missing',
2971
'bzrlib.tests.test_msgeditor',
2972
'bzrlib.tests.test_multiparent',
2973
'bzrlib.tests.test_mutabletree',
2974
'bzrlib.tests.test_nonascii',
2975
'bzrlib.tests.test_options',
2976
'bzrlib.tests.test_osutils',
2977
'bzrlib.tests.test_osutils_encodings',
2978
'bzrlib.tests.test_pack',
2979
'bzrlib.tests.test_pack_repository',
2980
'bzrlib.tests.test_patch',
2981
'bzrlib.tests.test_patches',
2982
'bzrlib.tests.test_permissions',
2983
'bzrlib.tests.test_plugins',
2984
'bzrlib.tests.test_progress',
2985
'bzrlib.tests.test_read_bundle',
2986
'bzrlib.tests.test_reconcile',
2987
'bzrlib.tests.test_reconfigure',
2988
'bzrlib.tests.test_registry',
2989
'bzrlib.tests.test_remote',
2990
'bzrlib.tests.test_repository',
2991
'bzrlib.tests.test_revert',
2992
'bzrlib.tests.test_revision',
2993
'bzrlib.tests.test_revisionspec',
2994
'bzrlib.tests.test_revisiontree',
2995
'bzrlib.tests.test_rio',
2996
'bzrlib.tests.test_rules',
2997
'bzrlib.tests.test_sampler',
2998
'bzrlib.tests.test_selftest',
2999
'bzrlib.tests.test_setup',
3000
'bzrlib.tests.test_sftp_transport',
3001
'bzrlib.tests.test_shelf',
3002
'bzrlib.tests.test_shelf_ui',
3003
'bzrlib.tests.test_smart',
3004
'bzrlib.tests.test_smart_add',
3005
'bzrlib.tests.test_smart_request',
3006
'bzrlib.tests.test_smart_transport',
3007
'bzrlib.tests.test_smtp_connection',
3008
'bzrlib.tests.test_source',
3009
'bzrlib.tests.test_ssh_transport',
3010
'bzrlib.tests.test_status',
3011
'bzrlib.tests.test_store',
3012
'bzrlib.tests.test_strace',
3013
'bzrlib.tests.test_subsume',
3014
'bzrlib.tests.test_switch',
3015
'bzrlib.tests.test_symbol_versioning',
3016
'bzrlib.tests.test_tag',
3017
'bzrlib.tests.test_testament',
3018
'bzrlib.tests.test_textfile',
3019
'bzrlib.tests.test_textmerge',
3020
'bzrlib.tests.test_timestamp',
3021
'bzrlib.tests.test_trace',
3022
'bzrlib.tests.test_transactions',
3023
'bzrlib.tests.test_transform',
3024
'bzrlib.tests.test_transport',
3025
'bzrlib.tests.test_transport_implementations',
3026
'bzrlib.tests.test_transport_log',
3027
'bzrlib.tests.test_tree',
3028
'bzrlib.tests.test_treebuilder',
3029
'bzrlib.tests.test_tsort',
3030
'bzrlib.tests.test_tuned_gzip',
3031
'bzrlib.tests.test_ui',
3032
'bzrlib.tests.test_uncommit',
3033
'bzrlib.tests.test_upgrade',
3034
'bzrlib.tests.test_upgrade_stacked',
3035
'bzrlib.tests.test_urlutils',
3036
'bzrlib.tests.test_version',
3037
'bzrlib.tests.test_version_info',
3038
'bzrlib.tests.test_versionedfile',
3039
'bzrlib.tests.test_weave',
3040
'bzrlib.tests.test_whitebox',
3041
'bzrlib.tests.test_win32utils',
3042
'bzrlib.tests.test_workingtree',
3043
'bzrlib.tests.test_workingtree_4',
3044
'bzrlib.tests.test_wsgi',
3045
'bzrlib.tests.test_xml',
3046
'bzrlib.tests.tree_implementations',
3047
'bzrlib.tests.workingtree_implementations',
3048
'bzrlib.util.tests.test_bencode',
3789
3051
loader = TestUtil.TestLoader()
3791
if keep_only is not None:
3792
id_filter = TestIdList(keep_only)
3793
3053
if starting_with:
3054
starting_with = [test_prefix_alias_registry.resolve_alias(start)
3055
for start in starting_with]
3794
3056
# We take precedence over keep_only because *at loading time* using
3795
3057
# both options means we will load less tests for the same final result.
3796
3058
def interesting_module(name):
3892
3221
for right_name, right_dict in scenarios_right]
3895
def multiply_tests(tests, scenarios, result):
3896
"""Multiply tests_list by scenarios into result.
3898
This is the core workhorse for test parameterisation.
3900
Typically the load_tests() method for a per-implementation test suite will
3901
call multiply_tests and return the result.
3903
:param tests: The tests to parameterise.
3904
:param scenarios: The scenarios to apply: pairs of (scenario_name,
3905
scenario_param_dict).
3906
:param result: A TestSuite to add created tests to.
3908
This returns the passed in result TestSuite with the cross product of all
3909
the tests repeated once for each scenario. Each test is adapted by adding
3910
the scenario name at the end of its id(), and updating the test object's
3911
__dict__ with the scenario_param_dict.
3913
>>> import bzrlib.tests.test_sampler
3914
>>> r = multiply_tests(
3915
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
3916
... [('one', dict(param=1)),
3917
... ('two', dict(param=2))],
3919
>>> tests = list(iter_suite_tests(r))
3923
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
3929
for test in iter_suite_tests(tests):
3930
apply_scenarios(test, scenarios, result)
3934
def apply_scenarios(test, scenarios, result):
3935
"""Apply the scenarios in scenarios to test and add to result.
3937
:param test: The test to apply scenarios to.
3938
:param scenarios: An iterable of scenarios to apply to test.
3940
:seealso: apply_scenario
3942
for scenario in scenarios:
3943
result.addTest(apply_scenario(test, scenario))
3947
def apply_scenario(test, scenario):
3948
"""Copy test and apply scenario to it.
3950
:param test: A test to adapt.
3951
:param scenario: A tuple describing the scenarion.
3952
The first element of the tuple is the new test id.
3953
The second element is a dict containing attributes to set on the
3955
:return: The adapted test.
3957
new_id = "%s(%s)" % (test.id(), scenario[0])
3958
new_test = clone_test(test, new_id)
3959
for name, value in scenario[1].items():
3960
setattr(new_test, name, value)
3964
def clone_test(test, new_id):
3965
"""Clone a test giving it a new id.
3967
:param test: The test to clone.
3968
:param new_id: The id to assign to it.
3969
:return: The new test.
3971
new_test = copy(test)
3972
new_test.id = lambda: new_id
3976
def permute_tests_for_extension(standard_tests, loader, py_module_name,
3978
"""Helper for permutating tests against an extension module.
3980
This is meant to be used inside a modules 'load_tests()' function. It will
3981
create 2 scenarios, and cause all tests in the 'standard_tests' to be run
3982
against both implementations. Setting 'test.module' to the appropriate
3983
module. See bzrlib.tests.test__chk_map.load_tests as an example.
3985
:param standard_tests: A test suite to permute
3986
:param loader: A TestLoader
3987
:param py_module_name: The python path to a python module that can always
3988
be loaded, and will be considered the 'python' implementation. (eg
3989
'bzrlib._chk_map_py')
3990
:param ext_module_name: The python path to an extension module. If the
3991
module cannot be loaded, a single test will be added, which notes that
3992
the module is not available. If it can be loaded, all standard_tests
3993
will be run against that module.
3994
:return: (suite, feature) suite is a test-suite that has all the permuted
3995
tests. feature is the Feature object that can be used to determine if
3996
the module is available.
3999
py_module = __import__(py_module_name, {}, {}, ['NO_SUCH_ATTRIB'])
4001
('python', {'module': py_module}),
4003
suite = loader.suiteClass()
4004
feature = ModuleAvailableFeature(ext_module_name)
4005
if feature.available():
4006
scenarios.append(('C', {'module': feature.module}))
4008
# the compiled module isn't available, so we add a failing test
4009
class FailWithoutFeature(TestCase):
4010
def test_fail(self):
4011
self.requireFeature(feature)
4012
suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4013
result = multiply_tests(standard_tests, scenarios, suite)
4014
return result, feature
4017
def _rmtree_temp_dir(dirname, test_id=None):
3225
def adapt_modules(mods_list, adapter, loader, suite):
3226
"""Adapt the modules in mods_list using adapter and add to suite."""
3227
tests = loader.loadTestsFromModuleNames(mods_list)
3228
adapt_tests(tests, adapter, suite)
3231
def adapt_tests(tests_list, adapter, suite):
3232
"""Adapt the tests in tests_list using adapter and add to suite."""
3233
for test in iter_suite_tests(tests_list):
3234
suite.addTests(adapter.adapt(test))
3237
def _rmtree_temp_dir(dirname):
4018
3238
# If LANG=C we probably have created some bogus paths
4019
3239
# which rmtree(unicode) will fail to delete
4020
3240
# so make sure we are using rmtree(str) to delete everything
4124
3342
UnicodeFilenameFeature = _UnicodeFilenameFeature()
4127
class _CompatabilityThunkFeature(Feature):
4128
"""This feature is just a thunk to another feature.
4130
It issues a deprecation warning if it is accessed, to let you know that you
4131
should really use a different feature.
4134
def __init__(self, dep_version, module, name,
4135
replacement_name, replacement_module=None):
4136
super(_CompatabilityThunkFeature, self).__init__()
4137
self._module = module
4138
if replacement_module is None:
4139
replacement_module = module
4140
self._replacement_module = replacement_module
4142
self._replacement_name = replacement_name
4143
self._dep_version = dep_version
4144
self._feature = None
4147
if self._feature is None:
4148
depr_msg = self._dep_version % ('%s.%s'
4149
% (self._module, self._name))
4150
use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4151
self._replacement_name)
4152
symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4153
# Import the new feature and use it as a replacement for the
4155
mod = __import__(self._replacement_module, {}, {},
4156
[self._replacement_name])
4157
self._feature = getattr(mod, self._replacement_name)
4161
return self._feature._probe()
4164
class ModuleAvailableFeature(Feature):
4165
"""This is a feature than describes a module we want to be available.
4167
Declare the name of the module in __init__(), and then after probing, the
4168
module will be available as 'self.module'.
4170
:ivar module: The module if it is available, else None.
4173
def __init__(self, module_name):
4174
super(ModuleAvailableFeature, self).__init__()
4175
self.module_name = module_name
4179
self._module = __import__(self.module_name, {}, {}, [''])
4186
if self.available(): # Make sure the probe has been done
4190
def feature_name(self):
4191
return self.module_name
4194
# This is kept here for compatibility, it is recommended to use
4195
# 'bzrlib.tests.feature.paramiko' instead
4196
ParamikoFeature = _CompatabilityThunkFeature(
4197
deprecated_in((2,1,0)),
4198
'bzrlib.tests.features', 'ParamikoFeature', 'paramiko')
3345
class TestScenarioApplier(object):
3346
"""A tool to apply scenarios to tests."""
3348
def adapt(self, test):
3349
"""Return a TestSuite containing a copy of test for each scenario."""
3350
result = unittest.TestSuite()
3351
for scenario in self.scenarios:
3352
result.addTest(self.adapt_test_to_scenario(test, scenario))
3355
def adapt_test_to_scenario(self, test, scenario):
3356
"""Copy test and apply scenario to it.
3358
:param test: A test to adapt.
3359
:param scenario: A tuple describing the scenarion.
3360
The first element of the tuple is the new test id.
3361
The second element is a dict containing attributes to set on the
3363
:return: The adapted test.
3365
from copy import deepcopy
3366
new_test = deepcopy(test)
3367
for name, value in scenario[1].items():
3368
setattr(new_test, name, value)
3369
new_id = "%s(%s)" % (new_test.id(), scenario[0])
3370
new_test.id = lambda: new_id
4201
3374
def probe_unicode_in_user_encoding():
added
removed
bzrlib/tests/__init__.py
