Externalize test runner

README.rst

@@ -169,6 +169,7 @@ List of all arguments with which you can run MutPy:
 -  ``-u UNIT_TEST [UNIT_TEST ...]``,
    ``--unit-test UNIT_TEST [UNIT_TEST ...]`` - test class, test method,
    module or package with unit tests,
+-  ``--runner RUNNER`` - currently supported are: unittest (default), pytest (experimental)
 -  ``-m``, ``--show-mutants`` - show mutants source code,
 -  ``-r REPORT_FILE``, ``--report REPORT_FILE`` - generate YAML report,
 -  ``--report-html DIR_NAME`` - generate HTML report,
@@ -240,6 +241,7 @@ Supported Test Runners
 Currently the following test runners are supported by MutPy:
 
 - `unittest <https://docs.python.org/3/library/unittest.html>`_
+- `pytest <https://docs.pytest.org/en/latest/>`_
 
 License
 -------

mutpy/__init__.py

@@ -1 +1 @@
-__version__ = '0.5.1'
+__version__ = '0.6.0'

mutpy/commandline.py

@@ -1,7 +1,8 @@
 import argparse
 import sys
-from mutpy import controller, views, operators, utils
+
 from mutpy import __version__ as version
+from mutpy import controller, views, operators, utils
 
 
 def main(argv):
@@ -17,6 +18,8 @@ def build_parser():
     parser.add_argument('--target', '-t', type=str, nargs='+', help='target module or package to mutate')
     parser.add_argument('--unit-test', '-u', type=str, nargs='+',
                         help='test class, test method, module or package with unit tests')
+    parser.add_argument('--runner', type=str, choices=['unittest', 'pytest'], default='unittest',
+                        metavar='RUNNER', help='test runner')
     parser.add_argument('--report', '-r', type=str, help='generate YAML report', metavar='REPORT_FILE')
     parser.add_argument('--report-html', type=str, help='generate HTML report', metavar='DIR_NAME')
     parser.add_argument('--timeout-factor', '-f', type=float, default=DEF_TIMEOUT_FACTOR,
@@ -27,7 +30,7 @@ def build_parser():
     parser.add_argument('--colored-output', '-c', action='store_true', help='try print colored output')
     parser.add_argument('--disable-stdout', '-d', action='store_true',
                         help='try disable stdout during mutation '
-                        '(this option can damage your tests if you interact with sys.stdout)')
+                             '(this option can damage your tests if you interact with sys.stdout)')
     parser.add_argument('--experimental-operators', '-e', action='store_true', help='use experimental operators')
     parser.add_argument('--operator', '-o', type=str, nargs='+',
                         help='use only selected operators', metavar='OPERATOR')
@@ -62,11 +65,13 @@ def run_mutpy(parser):
 
 
 def build_controller(cfg):
+    runner_cls = get_runner_cls(cfg.runner)
     built_views = build_views(cfg)
     mutant_generator = build_mutator(cfg)
     target_loader = utils.ModulesLoader(cfg.target, cfg.path)
     test_loader = utils.ModulesLoader(cfg.unit_test, cfg.path)
     return controller.MutationController(
+        runner_cls=runner_cls,
         target_loader=target_loader,
         test_loader=test_loader,
         views=built_views,
@@ -78,6 +83,16 @@ def build_controller(cfg):
     )
 
 
+def get_runner_cls(runner):
+    if runner == 'unittest':
+        from mutpy.test_runners import UnittestTestRunner
+        return UnittestTestRunner
+    elif runner == 'pytest':
+        from mutpy.test_runners import PytestTestRunner
+        return PytestTestRunner
+    raise ValueError('Unknown runner: {0}'.format(runner))
+
+
 def build_mutator(cfg):
     operators_set = set()
 

mutpy/controller.py

@@ -1,7 +1,8 @@
 import random
 import sys
-import unittest
-from mutpy import views, utils, coverage
+import time
+
+from mutpy import views, utils
 
 
 class TestsFailAtOriginal(Exception):
@@ -47,17 +48,16 @@ def all_mutants(self):
 
 class MutationController(views.ViewNotifier):
 
-    def __init__(self, target_loader, test_loader, views, mutant_generator,
+    def __init__(self, runner_cls, target_loader, test_loader, views, mutant_generator,
                  timeout_factor=5, disable_stdout=False, mutate_covered=False, mutation_number=None):
         super().__init__(views)
         self.target_loader = target_loader
         self.test_loader = test_loader
         self.mutant_generator = mutant_generator
         self.timeout_factor = timeout_factor
         self.stdout_manager = utils.StdoutManager(disable_stdout)
-        self.mutate_covered = mutate_covered
         self.mutation_number = mutation_number
-        self.store_init_modules()
+        self.runner = runner_cls(self.test_loader, self.timeout_factor, self.stdout_manager, mutate_covered)
 
     def run(self):
         self.notify_initialize(self.target_loader.names, self.test_loader.names)
@@ -92,34 +92,22 @@ def load_and_check_tests(self):
         total_duration = 0
         for test_module, target_test in self.test_loader.load():
             result, duration = self.run_test(test_module, target_test)
-            if result.wasSuccessful():
+            if result.was_successful():
                 test_modules.append((test_module, target_test, duration))
             else:
                 raise TestsFailAtOriginal(result)
-            number_of_tests += result.testsRun
+            number_of_tests += result.tests_run()
             total_duration += duration
 
         return test_modules, total_duration, number_of_tests
 
     def run_test(self, test_module, target_test):
-        suite = self.get_test_suite(test_module, target_test)
-        result = unittest.TestResult()
-        timer = utils.Timer()
-        with self.stdout_manager:
-            suite.run(result)
-        return result, timer.stop()
-
-    def get_test_suite(self, test_module, target_test):
-        if target_test:
-            return unittest.TestLoader().loadTestsFromName(target_test, test_module)
-        else:
-            return unittest.TestLoader().loadTestsFromModule(test_module)
+        return self.runner.run_test(test_module, target_test)
 
     @utils.TimeRegister
     def mutate_module(self, target_module, to_mutate, total_duration):
         target_ast = self.create_target_ast(target_module)
         coverage_injector, coverage_result = self.inject_coverage(target_ast, target_module)
-
         if coverage_injector:
             self.score.update_coverage(*coverage_injector.get_result())
         for mutations, mutant_ast in self.mutant_generator.mutate(target_ast, to_mutate, coverage_injector,
@@ -136,15 +124,7 @@ def mutate_module(self, target_module, to_mutate, total_duration):
                 self.score.inc_incompetent()
 
     def inject_coverage(self, target_ast, target_module):
-        if not self.mutate_covered:
-            return None, None
-        coverage_injector = coverage.CoverageInjector()
-        coverage_module = coverage_injector.inject(target_ast, target_module.__name__)
-        suite = self.create_test_suite(coverage_module)
-        coverage_result = coverage.CoverageTestResult(coverage_injector=coverage_injector)
-        with self.stdout_manager:
-            suite.run(coverage_result)
-        return coverage_injector, coverage_result
+        return self.runner.inject_coverage(target_ast, target_module)
 
     @utils.TimeRegister
     def create_target_ast(self, target_module):
@@ -163,51 +143,9 @@ def create_mutant_module(self, target_module, mutant_ast):
             self.notify_incompetent(0, exception, tests_run=0)
             return None
 
-    def create_test_suite(self, mutant_module):
-        suite = unittest.TestSuite()
-        utils.InjectImporter(mutant_module).install()
-        self.remove_loaded_modules()
-        for test_module, target_test in self.test_loader.load():
-            suite.addTests(self.get_test_suite(test_module, target_test))
-        utils.InjectImporter.uninstall()
-        return suite
-
-    def mark_not_covered_tests_as_skip(self, mutations, coverage_result, suite):
-        mutated_nodes = {mutation.node.marker for mutation in mutations}
-
-        def iter_tests(tests):
-            try:
-                for test in tests:
-                    iter_tests(test)
-            except TypeError:
-                add_skip(tests)
-
-        def add_skip(test):
-            if mutated_nodes.isdisjoint(coverage_result.test_covered_nodes[repr(test)]):
-                test_method = getattr(test, test._testMethodName)
-                setattr(test, test._testMethodName, unittest.skip('not covered')(test_method))
-
-        iter_tests(suite)
-
-    @utils.TimeRegister
     def run_tests_with_mutant(self, total_duration, mutant_module, mutations, coverage_result):
-        suite = self.create_test_suite(mutant_module)
-        if coverage_result:
-            self.mark_not_covered_tests_as_skip(mutations, coverage_result, suite)
-        timer = utils.Timer()
-        result = self.run_mutation_test_runner(suite, total_duration)
-        timer.stop()
-        self.update_score_and_notify_views(result, timer.duration)
-
-    def run_mutation_test_runner(self, suite, total_duration):
-        live_time = self.timeout_factor * (total_duration if total_duration > 1 else 1)
-        test_runner_class = utils.get_mutation_test_runner_class()
-        test_runner = test_runner_class(suite=suite)
-        with self.stdout_manager:
-            test_runner.start()
-            result = test_runner.get_result(live_time)
-            test_runner.terminate()
-        return result
+        result, duration = self.runner.run_tests_with_mutant(total_duration, mutant_module, mutations, coverage_result)
+        self.update_score_and_notify_views(result, duration)
 
     def update_score_and_notify_views(self, result, mutant_duration):
         if not result:
@@ -235,17 +173,6 @@ def update_killed_mutant(self, result, duration):
         self.notify_killed(duration, result.killer, result.exception_traceback, result.tests_run)
         self.score.inc_killed()
 
-    def store_init_modules(self):
-        test_runner_class = utils.get_mutation_test_runner_class()
-        test_runner = test_runner_class(suite=unittest.TestSuite())
-        test_runner.start()
-        self.init_modules = list(sys.modules.keys())
-
-    def remove_loaded_modules(self):
-        for module in list(sys.modules.keys()):
-            if module not in self.init_modules:
-                del sys.modules[module]
-
 
 class HOMStrategy:
 
@@ -256,9 +183,9 @@ def remove_bad_mutations(self, mutations_to_apply, available_mutations, allow_sa
         for mutation_to_apply in mutations_to_apply:
             for available_mutation in available_mutations[:]:
                 if mutation_to_apply.node == available_mutation.node or \
-                   mutation_to_apply.node in available_mutation.node.children or \
-                   available_mutation.node in mutation_to_apply.node.children or \
-                   (not allow_same_operators and mutation_to_apply.operator == available_mutation.operator):
+                        mutation_to_apply.node in available_mutation.node.children or \
+                        available_mutation.node in mutation_to_apply.node.children or \
+                        (not allow_same_operators and mutation_to_apply.operator == available_mutation.operator):
                     available_mutations.remove(available_mutation)
 
 

mutpy/coverage.py

@@ -1,6 +1,6 @@
 import ast
 import copy
-import unittest
+
 from mutpy import utils
 
 COVERAGE_SET_NAME = '__covered_nodes__'
@@ -79,7 +79,6 @@ def get_markers_from_body_node(self, node):
 
 
 class CoverageNodeTransformerPython32(AbstractCoverageNodeTransformer):
-
     __python_version__ = (3, 2)
 
     @classmethod
@@ -111,7 +110,6 @@ def get_coverable_nodes(cls):
 
 
 class CoverageNodeTransformerPython33(AbstractCoverageNodeTransformer):
-
     __python_version__ = (3, 3)
 
     @classmethod
@@ -170,22 +168,3 @@ def is_covered(self, child_node):
 
     def get_result(self):
         return len(self.covered_nodes), self.marker_transformer.last_marker
-
-
-class CoverageTestResult(unittest.TestResult):
-
-    def __init__(self, *args, coverage_injector=None, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.coverage_injector = coverage_injector
-        self.always_covered_nodes = coverage_injector.covered_nodes.copy()
-        self.test_covered_nodes = {}
-
-    def startTest(self, test):
-        super().startTest(test)
-        self.covered_nodes = self.coverage_injector.covered_nodes.copy()
-        self.coverage_injector.covered_nodes.clear()
-
-    def stopTest(self, test):
-        super().stopTest(test)
-        self.test_covered_nodes[repr(test)] = self.coverage_injector.covered_nodes.copy() | self.always_covered_nodes
-        self.coverage_injector.covered_nodes.update(self.covered_nodes)

mutpy/test/test_controller.py

@@ -4,6 +4,8 @@
 import unittest
 
 from mutpy import controller, operators, utils, codegen
+from mutpy.test.utils import MockModulesLoader
+from mutpy.test_runners import UnittestTestRunner
 
 
 class MutationScoreTest(unittest.TestCase):
@@ -48,23 +50,6 @@ def test_update_coverage(self):
         self.assertEqual(self.score.all_nodes, 1)
 
 
-class MockModulesLoader:
-    def __init__(self, name, source):
-        self.names = [name]
-        self.source = source
-        self.module = types.ModuleType(name)
-        self.module.__file__ = '<string>'
-        self.load()
-
-    def load(self, *args, **kwargs):
-        exec(self.source, self.module.__dict__)
-        sys.modules[self.names[0]] = self.module
-        return [(self.module, None)]
-
-    def get_source(self):
-        return self.source
-
-
 class MockMutationController(controller.MutationController):
     def create_target_ast(self, target_module):
         return utils.create_ast(self.target_loader.get_source())
@@ -93,6 +78,7 @@ def setUp(self):
         self.score_view = MutationScoreStoreView()
         mutator = controller.FirstOrderMutator([operators.ArithmeticOperatorReplacement], percentage=100)
         self.mutation_controller = MockMutationController(
+            runner_cls=UnittestTestRunner,
             target_loader=target_loader,
             test_loader=test_loader,
             views=[self.score_view],

mutpy/test/test_coverage.py

@@ -1,6 +1,7 @@
 import unittest
 
 from mutpy import coverage, utils
+from mutpy.test_runners.unittest_runner import UnittestCoverageResult
 
 
 class MarkerNodeTransformerTest(unittest.TestCase):
@@ -189,7 +190,7 @@ def test_for_coverage(self):
         self.assert_not_covered([for_body_el])
 
 
-class CoverageTestResultTest(unittest.TestCase):
+class UnittestCoverageResultTest(unittest.TestCase):
     def test_run(self):
         coverage_injector = coverage.CoverageInjector()
 
@@ -204,7 +205,7 @@ def test_x(self):
             def test_y(self):
                 pass
 
-        result = coverage.CoverageTestResult(coverage_injector=coverage_injector)
+        result = UnittestCoverageResult(coverage_injector=coverage_injector)
         suite = unittest.TestSuite()
         test_x = ATest(methodName='test_x')
         suite.addTest(test_x)