wip: don't mutate lines excluded from coverage

HISTORY.rst

@@ -1,6 +1,13 @@
 Changelog
 ---------
 
+Unreleased
+~~~~~~~~~~
+
+* Fix `mutate_only_covered_lines` mutating code that coverage.py excludes from measurement (`# pragma: no cover`, `exclude_lines`, `exclude_also`). Such lines are reported as covered when they run, so they used to produce mutants that could only ever survive
+
+* Fix mutations that delete ignored code. Dropping a `case` from a `match`, or an argument from a call, is a mutation of the enclosing node, so it used to be made even when the removed lines were themselves ignored
+
 3.7.0
 ~~~~~
 

README.rst

@@ -169,6 +169,11 @@ If you only want to mutate lines that are called (according to coverage.py), you
 
     mutate_only_covered_lines=true
 
+This also honours the lines coverage.py is told to leave out of its measurement, so code
+marked with `# pragma: no cover`, or matched by your `exclude_lines`/`exclude_also`
+settings, is not mutated either. Such code is not held to your test suite, so mutants
+there could only ever show up as survivors.
+
 
 Filter generated mutants with type checker
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

e2e_projects/mutate_only_covered_lines/pyproject.toml

@@ -21,6 +21,12 @@ omit = [
     "src/mutate_only_covered_lines/omit_me.py",
 ]
 
+[tool.coverage.report]
+exclude_also = [
+    "if debug_only:",
+    "case _NEVER:",
+]
+
 [tool.mutmut]
 debug = true
 mutate_only_covered_lines = true

e2e_projects/mutate_only_covered_lines/src/mutate_only_covered_lines/exclude_me.py

@@ -0,0 +1,42 @@
+"""Code that coverage.py is told not to measure.
+
+Every excluded region below is *executed* by the tests, so it ends up in the covered
+lines. Only coverage.py's exclusion rules can keep it from being mutated.
+"""
+
+debug_only = True
+
+
+def excluded_function(a: int, b: int) -> int:  # pragma: no cover
+    result = (
+        a
+        + b
+    )
+    return result
+
+
+def excluded_branch(flag: bool) -> int:
+    if flag:  # pragma: no cover
+        return 1 + 1
+    return 2 + 2
+
+
+def excluded_by_coverage_config(a: int, b: int) -> int:
+    total = a * b
+    if debug_only:
+        total = (
+            total
+            + 1
+        )
+    return total
+
+
+def excluded_case(kind: str) -> int:
+    # dropping a case is a mutation of the `match`, not of the excluded lines themselves
+    match kind:
+        case "a":
+            return 1
+        case "b":
+            return 2
+        case _NEVER:
+            raise Exception("Can't happen")

e2e_projects/mutate_only_covered_lines/tests/main/test_mutate_only_covered_lines.py

@@ -1,5 +1,6 @@
 from mutate_only_covered_lines.ignore_me import this_function_shall_NOT_be_mutated
 from mutate_only_covered_lines import hello_mutate_only_covered_lines, mutate_only_covered_lines_multiline, function_with_pragma, do_not_mutate_external_ommited_function
+from mutate_only_covered_lines.exclude_me import excluded_function, excluded_branch, excluded_by_coverage_config, excluded_case
 
 """This tests the mutate_only_covered_lines feature."""
 
@@ -17,3 +18,18 @@ def call_ignored_function():
 
 def test_do_not_mutate_external_ommited_function():
     assert do_not_mutate_external_ommited_function() == 7
+
+def test_excluded_function():
+    assert excluded_function(1, 2) == 3
+
+def test_excluded_branch():
+    # both branches run, so the excluded one is covered and only its exclusion can save it
+    assert excluded_branch(True) == 2
+    assert excluded_branch(False) == 4
+
+def test_excluded_by_coverage_config():
+    assert excluded_by_coverage_config(2, 3) == 7
+
+def test_excluded_case():
+    assert excluded_case("a") == 1
+    assert excluded_case("b") == 2

src/mutmut/__init__.py

@@ -16,6 +16,7 @@
 
 _stats: set[str] = set()
 _covered_lines: dict[str, set[int]] | None = None
+_excluded_lines: dict[str, set[int]] | None = None
 
 
 def __getattr__(name: str) -> object:
@@ -33,12 +34,13 @@ def __getattr__(name: str) -> object:
 
 def _reset_globals() -> None:
     global duration_by_test, stats_time, _stats, tests_by_mangled_function_name
-    global _covered_lines
+    global _covered_lines, _excluded_lines
 
     duration_by_test.clear()
     stats_time = None
     Config.reset()
     _stats = set()
     tests_by_mangled_function_name = defaultdict(set)
     _covered_lines = None
+    _excluded_lines = None
     reset_state()

src/mutmut/__main__.py

@@ -60,6 +60,7 @@
 import mutmut
 from mutmut.code_coverage import gather_coverage
 from mutmut.code_coverage import get_covered_lines_for_file
+from mutmut.code_coverage import get_excluded_lines_for_file
 from mutmut.configuration import Config
 from mutmut.mutation.data import MutantLineSpans
 from mutmut.mutation.data import SourceFileMutationData
@@ -278,7 +279,9 @@ def setup_source_paths() -> None:
 
 def store_lines_covered_by_tests() -> None:
     if Config.get().mutate_only_covered_lines:
-        mutmut._covered_lines = gather_coverage(PytestRunner(), list(walk_source_files()))
+        coverage_info = gather_coverage(PytestRunner(), list(walk_source_files()))
+        mutmut._covered_lines = coverage_info.covered_lines
+        mutmut._excluded_lines = coverage_info.excluded_lines
 
 
 def copy_also_copy_files() -> None:
@@ -371,7 +374,10 @@ def create_mutants_for_file(filename: Path, output_path: Path) -> FileMutationRe
 
 def write_all_mutants_to_file(*, out: TextIOBase, source: str, filename: Path) -> MutatedFile:
     mutated_file = mutate_file_contents(
-        str(filename), source, get_covered_lines_for_file(str(filename), mutmut._covered_lines)
+        str(filename),
+        source,
+        get_covered_lines_for_file(str(filename), mutmut._covered_lines),
+        get_excluded_lines_for_file(str(filename), mutmut._excluded_lines),
     )
     out.write(mutated_file.code)
 

src/mutmut/code_coverage.py

@@ -3,17 +3,36 @@
 import importlib
 import sys
 from collections.abc import Iterable
+from dataclasses import dataclass
+from dataclasses import field
 from pathlib import Path
 from types import ModuleType
 from typing import TYPE_CHECKING
 
 import coverage
 from coverage import CoverageData
+from coverage.exceptions import CoverageException
 
 if TYPE_CHECKING:
     from mutmut.__main__ import TestRunner
 
 
+@dataclass
+class CoverageInfo:
+    """What coverage.py knows about the source files, keyed by absolute path in `mutants/`.
+
+    `covered_lines` are the lines that actually executed. `excluded_lines` are the lines
+    coverage.py was told to ignore (`# pragma: no cover`, `exclude_lines`, `exclude_also`).
+    The two are kept apart rather than subtracted, because an excluded line that runs is
+    also a covered line, and the two are used differently: coverage.py reports only the
+    *first* line of each excluded statement, so those have to be expanded to full
+    statements before they mean anything to the mutation visitor.
+    """
+
+    covered_lines: dict[str, set[int]] = field(default_factory=dict)
+    excluded_lines: dict[str, set[int]] = field(default_factory=dict)
+
+
 # Returns a set of lines that are covered in this file gvein the covered_lines dict
 #  returned by gather_coverage
 # None means it's not enabled, set() means no lines are covered
@@ -29,11 +48,22 @@ def get_covered_lines_for_file(filename: str, covered_lines: dict[str, set[int]]
     return lines
 
 
+# Returns the lines coverage.py excludes from measurement in this file, given the
+# excluded_lines dict returned by gather_coverage. An empty set means nothing is excluded,
+# which is also what we get when the feature is disabled.
+def get_excluded_lines_for_file(filename: str, excluded_lines: dict[str, set[int]] | None) -> set[int]:
+    if excluded_lines is None or filename is None:
+        return set()
+
+    abs_filename = str((Path("mutants") / filename).absolute())
+    return set(excluded_lines.get(abs_filename, ()))
+
+
 # Gathers coverage for the given source files and
-# Returns a dict of filenames to sets of lines that are covered
+# Returns the covered and excluded lines of each of them
 # Since this is run on the source files before we create mutations,
 # we need to unload any modules that get loaded during the test run
-def gather_coverage(runner: TestRunner, source_files: Iterable[Path]) -> dict[str, set[int]]:
+def gather_coverage(runner: TestRunner, source_files: Iterable[Path]) -> CoverageInfo:
     # We want to unload any python modules that get loaded
     # because we plan to mutate them and want them to be reloaded
     modules = dict(sys.modules)
@@ -47,17 +77,30 @@ def gather_coverage(runner: TestRunner, source_files: Iterable[Path]) -> dict[st
     # Build mapping of filenames to covered lines
     # The CoverageData object is a wrapper around sqlite, and this
     # will make it more efficient to access the data
-    covered_lines: dict[str, set[int]] = {}
+    info = CoverageInfo()
     coverage_data: CoverageData = cov.get_data()
 
     for filename in source_files:
         abs_filename = str((mutants_path / filename).absolute())
-        lines = set(coverage_data.lines(abs_filename) or [])
-        covered_lines[abs_filename] = lines
+        info.covered_lines[abs_filename] = set(coverage_data.lines(abs_filename) or [])
+        info.excluded_lines[abs_filename] = _excluded_lines(cov, abs_filename)
 
     _unload_modules_not_in(modules)
 
-    return covered_lines
+    return info
+
+
+# Asks coverage.py which lines of this file are excluded from measurement.
+# This is analysis of the source, not of the collected data, so it is also
+# correct for files the test run never imported.
+def _excluded_lines(cov: coverage.Coverage, abs_filename: str) -> set[int]:
+    try:
+        _, _, excluded, _, _ = cov.analysis2(abs_filename)
+    except CoverageException:
+        # Unparseable or missing source: nothing we can say about it
+        return set()
+
+    return set(excluded)
 
 
 # Unloads modules that are not in the 'modules' list

src/mutmut/mutation/file_mutation.py

@@ -5,6 +5,7 @@
 from collections import defaultdict
 from collections.abc import Callable
 from collections.abc import Iterable
+from collections.abc import Iterator
 from collections.abc import Mapping
 from collections.abc import Sequence
 from dataclasses import dataclass
@@ -99,28 +100,40 @@ class MutatedFile:
     hash_by_function_name: Mapping[str, str]
 
 
-def mutate_file_contents(filename: str, code: str, covered_lines: set[int] | None = None) -> MutatedFile:
+def mutate_file_contents(
+    filename: str,
+    code: str,
+    covered_lines: set[int] | None = None,
+    coverage_excluded_lines: set[int] | None = None,
+) -> MutatedFile:
     """Create mutations for `code` and merge them to a single mutated file with trampolines."""
-    module, mutations, ignored_classes, ignored_functions = create_mutations(filename, code, covered_lines)
+    module, mutations, ignored_classes, ignored_functions = create_mutations(
+        filename, code, covered_lines, coverage_excluded_lines
+    )
 
     mutated_file = combine_mutations_to_source(module, mutations, ignored_classes, ignored_functions)
 
     return replace(mutated_file, hash_by_function_name=_compute_mutated_function_hashes(code, module, mutations))
 
 
 def create_mutations(
-    filename: str, code: str, covered_lines: set[int] | None = None
+    filename: str,
+    code: str,
+    covered_lines: set[int] | None = None,
+    coverage_excluded_lines: set[int] | None = None,
 ) -> tuple[cst.Module, list[Mutation], set[str], set[str]]:
     """Parse the code and create mutations.
 
     :param filename: File path forwarded to :class:`PragmaVisitor` for error messages.
     :param code: Python source code to parse and mutate.
     :param covered_lines: If provided, only lines in this set are considered for mutation.
+    :param coverage_excluded_lines: Lines coverage.py excludes from measurement. The
+        statements starting on them are ignored, like a `# pragma: no mutate block`.
     :return: A tuple of (module, mutations, ignored_classes, ignored_functions)."""
     module = cst.parse_module(code)
     metadata_wrapper = MetadataWrapper(module)
 
-    ignored_code = get_ignored_lines(filename, code, metadata_wrapper)
+    ignored_code = get_ignored_lines(filename, code, metadata_wrapper, coverage_excluded_lines)
 
     visitor = MutationVisitor(
         mutation_operators,
@@ -199,6 +212,9 @@ def __init__(
         self._covered_lines = covered_lines
         self._ignored_node_lines = ignored_code.ignore_node_lines
         self._ignored_pattern_lines = ignored_code.ignore_pattern_lines
+        self._all_ignored_lines = (
+            ignored_code.no_mutate_lines | ignored_code.ignore_node_lines | ignored_code.ignore_pattern_lines
+        )
         self.ignored_classes: set[str] = set()
         self.ignored_functions: set[str] = set()
 
@@ -216,13 +232,41 @@ def _create_mutations(self, node: cst.CSTNode) -> None:
         for t, operator in self._operators:
             if isinstance(node, t):
                 for mutated_node in operator(node):
+                    if self._removes_ignored_code(node, mutated_node):
+                        continue
                     mutation = Mutation(
                         original_node=node,
                         mutated_node=mutated_node,
                         contained_by_top_level_function=self.get_metadata(OuterFunctionProvider, node, None),  # type: ignore
                     )
                     self.mutations.append(mutation)
 
+    def _removes_ignored_code(self, node: cst.CSTNode, mutated_node: cst.CSTNode) -> bool:
+        """Does this mutation delete code that we were told not to mutate?
+
+        Some operators mutate by removing a part of a node: a `case` of a `match`, an
+        argument of a call. Such a mutation is anchored on the enclosing node, so the
+        line checks in `_should_mutate_node` never get to see the removed lines."""
+        if not self._all_ignored_lines or not self._spans_ignored_line(node):
+            return False
+
+        kept = {id(child) for child in _walk(mutated_node)}
+        for child in _walk(node):
+            if id(child) in kept:
+                continue
+            position = self.get_metadata(PositionProvider, child, None)
+            if position and position.start.line in self._all_ignored_lines:
+                return True
+
+        return False
+
+    def _spans_ignored_line(self, node: cst.CSTNode) -> bool:
+        position = self.get_metadata(PositionProvider, node, None)
+        if position is None:
+            return True
+
+        return any(line in self._all_ignored_lines for line in range(position.start.line, position.end.line + 1))
+
     def _should_mutate_node(self, node: cst.CSTNode) -> bool:
         # currently, the position metadata does not always exist
         # (see https://github.com/Instagram/LibCST/issues/1322)
@@ -295,6 +339,15 @@ def _skip_node_and_children(self, node: cst.CSTNode) -> bool:
         return False
 
 
+def _walk(node: cst.CSTNode) -> Iterator[cst.CSTNode]:
+    """Yield `node` and every node below it."""
+    stack = [node]
+    while stack:
+        current = stack.pop()
+        yield current
+        stack.extend(current.children)
+
+
 MODULE_STATEMENT = Union[cst.SimpleStatementLine, cst.BaseCompoundStatement]
 
 # convert str trampoline implementations to CST nodes with some whitespace