implement assertion minimization

LuKrO2011 · LuKrO2011 · commit 1d85191a0ecc · 2026-07-06T16:47:44.000+02:00
diff --git a/src/pynguin/assertion/assertiongenerator.py b/src/pynguin/assertion/assertiongenerator.py
@@ -223,6 +223,59 @@ def get_score(self) -> float:
         return self.num_killed_mutants / divisor
 
 
+def _select_minimal_assertions(
+    kill_map: dict[tuple[int, int], set[int]],
+) -> set[tuple[int, int]]:
+    """Greedy set-cover selection of a minimal assertion subset.
+
+    Given a mapping from assertion key ``(stmt_idx, assertion_idx)`` to the set of
+    mutant indices that assertion kills, return the subset of keys that together
+    cover every killed mutant (the union of all kill sets). Assertions with an empty
+    kill set are never selected. Ties are broken by ascending key (insertion order)
+    for deterministic output.
+
+    Args:
+        kill_map: Mapping from assertion key to the set of mutants it kills.
+
+    Returns:
+        The set of assertion keys to keep.
+    """
+    universe: set[int] = set()
+    for kills in kill_map.values():
+        universe |= kills
+
+    # Only assertions that kill at least one mutant can ever be selected.
+    candidates = {key: kills for key, kills in kill_map.items() if kills}
+
+    uncovered = set(universe)
+    keep: set[tuple[int, int]] = set()
+    while uncovered:
+        best_key: tuple[int, int] | None = None
+        best_cover = 0
+        for key in sorted(candidates):
+            cover = len(candidates[key] & uncovered)
+            if cover > best_cover:
+                best_cover = cover
+                best_key = key
+        if best_key is None:
+            break
+        keep.add(best_key)
+        uncovered -= candidates[best_key]
+        del candidates[best_key]
+
+    # Greedy may leave a selection whose kills are fully covered by the others it
+    # later picked. Prune such redundant assertions, dropping higher keys first so
+    # lower (earlier) assertions are preferred, keeping coverage unchanged.
+    for key in sorted(keep, reverse=True):
+        others: set[int] = set()
+        for other in keep:
+            if other != key:
+                others |= kill_map[other]
+        if kill_map[key] <= others:
+            keep.discard(key)
+    return keep
+
+
 class MutationAnalysisAssertionGenerator(AssertionGenerator):
     """Uses mutation analysis to filter out less relevant assertions."""
 
@@ -381,6 +434,11 @@ def __remove_non_relevant_assertions(
         tests_mutants_results: list[list[ex.ExecutionResult | None]],
         mutation_summary: _MutationSummary,
     ) -> None:
+        if config.configuration.test_case_output.assertion_minimization:
+            MutationAnalysisAssertionGenerator.__minimize_assertions(
+                test_cases, tests_mutants_results, mutation_summary
+            )
+            return
         for test, results in zip(test_cases, tests_mutants_results, strict=True):
             merged = at.AssertionVerificationTrace()
             for result, mut in zip(results, mutation_summary.mutant_information, strict=True):
@@ -392,6 +450,77 @@ def __remove_non_relevant_assertions(
                     if not merged.was_violated(stmt_idx, assertion_idx):
                         statement.assertions.remove(assertion)
 
+    @staticmethod
+    def __minimize_assertions(
+        test_cases: list[tc.TestCase],
+        tests_mutants_results: list[list[ex.ExecutionResult | None]],
+        mutation_summary: _MutationSummary,
+    ) -> None:
+        """Keep a minimal subset of assertions preserving the killed mutants.
+
+        For each test case, a greedy set-cover selection keeps only enough
+        assertions to preserve the full set of assertion-attributable mutant kills.
+        Assertions that kill nothing are dropped (subsuming the plain
+        non-relevant-assertion removal); redundant assertions that only re-kill
+        mutants already covered by kept assertions are removed as well.
+
+        Statements carrying only an exception assertion are left untouched to
+        avoid the fragility of mixing exception and value assertions.
+
+        Args:
+            test_cases: The test cases whose assertions are minimized.
+            tests_mutants_results: Per-test, per-mutant execution results.
+            mutation_summary: Summary with per-mutant timeout information.
+        """
+        for test, results in zip(test_cases, tests_mutants_results, strict=True):
+            kill_map = MutationAnalysisAssertionGenerator.__build_kill_map(
+                test, results, mutation_summary
+            )
+            keep = _select_minimal_assertions(kill_map)
+
+            for stmt_idx, statement in enumerate(test.statements):
+                if statement.has_only_exception_assertion():
+                    continue
+                for assertion_idx, assertion in reversed(list(enumerate(statement.assertions))):
+                    if (stmt_idx, assertion_idx) not in keep:
+                        statement.assertions.remove(assertion)
+
+    @staticmethod
+    def __build_kill_map(
+        test: tc.TestCase,
+        results: list[ex.ExecutionResult | None],
+        mutation_summary: _MutationSummary,
+    ) -> dict[tuple[int, int], set[int]]:
+        """Map each assertion to the set of mutants it kills via violation.
+
+        Statements carrying only an exception assertion are skipped. Only
+        assertion-attributable kills on non-timed-out mutants are counted.
+
+        Args:
+            test: The test case whose assertions are inspected.
+            results: Per-mutant execution results for this test.
+            mutation_summary: Summary with per-mutant timeout information.
+
+        Returns:
+            Mapping from ``(stmt_idx, assertion_idx)`` to the killed mutant indices.
+        """
+        kill_map: dict[tuple[int, int], set[int]] = {}
+        for stmt_idx, statement in enumerate(test.statements):
+            if statement.has_only_exception_assertion():
+                continue
+            for assertion_idx in range(len(statement.assertions)):
+                kills = {
+                    mutant_idx
+                    for mutant_idx, (result, mut) in enumerate(
+                        zip(results, mutation_summary.mutant_information, strict=True)
+                    )
+                    if result is not None
+                    and len(mut.timed_out_by) == 0
+                    and result.assertion_verification_trace.was_violated(stmt_idx, assertion_idx)
+                }
+                kill_map[stmt_idx, assertion_idx] = kills
+        return kill_map
+
     @staticmethod
     def __compute_mutation_summary(
         number_of_mutants: int,
diff --git a/src/pynguin/configuration.py b/src/pynguin/configuration.py
@@ -367,6 +367,10 @@ class TestCaseOutputConfiguration:
     marking the test with @pytest.mark.xfail(strict=True). Use this for better mutation
     scores."""
 
+    assertion_minimization: bool = True
+    """If enabled, remove redundant assertions after mutation analysis, keeping a
+    minimal subset that preserves the set of killed mutants (greedy set cover)."""
+
 
 @dataclasses.dataclass
 class SeedingConfiguration:
diff --git a/tests/assertion/test_assertion_generation_integration.py b/tests/assertion/test_assertion_generation_integration.py
@@ -490,6 +490,114 @@ def test_mutation_analysis_truncated_by_time_budget(subject_properties: SubjectP
         assert len(threading.enumerate()) == 1
 
 
+@pytest.mark.parametrize(
+    "module,test_case_str,test_case_str_with_assertions,killed,timeout",
+    [
+        # Value assertion that kills mutants is kept.
+        (
+            "tests.fixtures.mutation.mutation",
+            "def test_case_0():\n    int_0 = 0\n    int_1 = 0\n    int_2 = 0\n"
+            "    int_3 = 1\n    float_0 = module_0.foo(int_3)",
+            "int_0 = 0\nint_1 = 0\nint_2 = 0\nint_3 = 1\nfloat_0 = "
+            "module_0.foo(int_3)\n"
+            "assert float_0 == pytest.approx(2.0, abs=0.01, rel=0.01)",
+            {0, 1, 3, 4},
+            set(),
+        ),
+        # Assertion that kills nothing is dropped (subsumes kills-nothing removal).
+        (
+            "tests.fixtures.mutation.mutation",
+            "def test_case_0():\n    int_0 = 0\n    int_1 = 0\n    int_2 = 0\n"
+            "    int_3 = 0\n    float_0 = module_0.foo(int_3)",
+            "int_0 = 0\nint_1 = 0\nint_2 = 0\nint_3 = 0\nmodule_0.foo(int_3)",
+            set(),
+            set(),
+        ),
+        # Statement with only an exception assertion is left untouched.
+        (
+            "tests.fixtures.mutation.exception",
+            "def test_case_0():\n    float_0 = module_0.foo()",
+            "module_0.foo()",
+            {0, 3, 4},
+            set(),
+        ),
+        # Expected-exception (pytest.raises) case is preserved.
+        (
+            "tests.fixtures.mutation.expected",
+            "def test_case_0():\n    int_0 = 2\n    var_0 = module_0.bar(int_0)",
+            "int_0 = 2\nwith pytest.raises(ValueError):\n    module_0.bar(int_0)",
+            {0, 1, 2},
+            set(),
+        ),
+    ],
+)
+def test_mutation_analysis_minimization_preserves_output(  # noqa: PLR0914, PLR0917
+    module,
+    test_case_str,
+    test_case_str_with_assertions,
+    killed,
+    timeout,
+    subject_properties: SubjectProperties,
+):
+    """With minimization on, already-minimal outputs are reproduced exactly.
+
+    These fixtures produce assertions that are either uniquely required or kill
+    nothing, so greedy set cover must keep exactly the same assertions as the
+    default path while preserving the killed mutants.
+    """
+    original_minimization = config.configuration.test_case_output.assertion_minimization
+    config.configuration.test_case_output.assertion_minimization = True
+    try:
+        config.configuration.module_name = module
+        module_name = config.configuration.module_name
+        with install_import_hook(module_name, subject_properties):
+            with subject_properties.instrumentation_tracer:
+                module_type = importlib.import_module(module_name)
+                importlib.reload(module_type)
+            cluster = generate_test_cluster(module_name)
+            transformer = AstToTestCaseTransformer(
+                cluster,
+                False,  # noqa: FBT003
+                EmptyConstantProvider(),
+            )
+            transformer.visit(ast.parse(test_case_str))
+            test_case = transformer.testcases[0]
+
+            chromosome = tcc.TestCaseChromosome(test_case)
+            suite = tsc.TestSuiteChromosome()
+            suite.add_test_case_chromosome(chromosome)
+
+            mutant_generator = mu.FirstOrderMutator([
+                *test_operators,
+                mo.OneIterationLoop,
+                mo.ReverseIterationLoop,
+                mo.ZeroIterationLoop,
+            ])
+            module_source_code = inspect.getsource(module_type)
+            module_ast = ParentNodeTransformer.create_ast(module_source_code)
+            mutation_controller = MutationController(mutant_generator, module_ast, module_type)
+            gen = ag.MutationAnalysisAssertionGenerator(
+                TestCaseExecutor(subject_properties), mutation_controller, testing=True
+            )
+            suite.accept(gen)
+
+            summary = gen._testing_mutation_summary
+            assert {k.mut_num for k in summary.get_killed()} == killed
+            assert {k.mut_num for k in summary.get_timeout()} == timeout
+
+            visitor = tc_to_ast.TestCaseToAstVisitor(ns.NamingScope(prefix="module"), set())
+            test_case.accept(visitor)
+            source = ast.unparse(
+                ast.fix_missing_locations(ast.Module(body=visitor.test_case_ast, type_ignores=[]))
+            )
+            assert source == test_case_str_with_assertions
+            for thread in threading.enumerate():
+                if "_execute_test_case" in thread.name:
+                    thread.join()
+    finally:
+        config.configuration.test_case_output.assertion_minimization = original_minimization
+
+
 def _add_assertions(module_name: str, test_case_code: str) -> str:
     """Add assertions to a test case and return the test case with assertions.
 
diff --git a/tests/assertion/test_assertiongenerator.py b/tests/assertion/test_assertiongenerator.py
@@ -5,6 +5,8 @@
 #  SPDX-License-Identifier: MIT
 #
 import pytest
+from hypothesis import given
+from hypothesis import strategies as st
 
 import pynguin.assertion.assertiongenerator as ag
 
@@ -77,3 +79,78 @@ def test_score_excludes_unchecked_mutants():
 )
 def test_compute_metrics(inp, result):
     assert ag._MutationSummary(inp).get_metrics() == result
+
+
+def test_select_minimal_assertions_empty():
+    assert ag._select_minimal_assertions({}) == set()
+
+
+def test_select_minimal_assertions_drops_assertion_killing_nothing():
+    kill_map = {(0, 0): {1, 2}, (0, 1): set()}
+    assert ag._select_minimal_assertions(kill_map) == {(0, 0)}
+
+
+def test_select_minimal_assertions_identical_kills_keeps_one():
+    # Two assertions killing exactly the same mutants -> keep exactly one.
+    kill_map = {(0, 0): {1, 2}, (0, 1): {1, 2}}
+    assert ag._select_minimal_assertions(kill_map) == {(0, 0)}
+
+
+def test_select_minimal_assertions_subset_dropped():
+    # (0, 1) kills a subset of what (0, 0) kills -> only the superset is kept.
+    kill_map = {(0, 0): {1, 2, 3}, (0, 1): {2}}
+    assert ag._select_minimal_assertions(kill_map) == {(0, 0)}
+
+
+def test_select_minimal_assertions_disjoint_keeps_both():
+    kill_map = {(0, 0): {1}, (0, 1): {2}}
+    assert ag._select_minimal_assertions(kill_map) == {(0, 0), (0, 1)}
+
+
+def test_select_minimal_assertions_tie_break_lowest_index():
+    # Both cover the whole universe -> the lowest key is chosen deterministically.
+    kill_map = {(1, 0): {1, 2}, (0, 0): {1, 2}}
+    assert ag._select_minimal_assertions(kill_map) == {(0, 0)}
+
+
+def test_select_minimal_assertions_covers_universe():
+    kill_map = {(0, 0): {1, 2}, (0, 1): {2, 3}, (0, 2): {4}}
+    keep = ag._select_minimal_assertions(kill_map)
+    covered: set[int] = set()
+    for key in keep:
+        covered |= kill_map[key]
+    assert covered == {1, 2, 3, 4}
+
+
+@given(
+    st.dictionaries(
+        keys=st.tuples(
+            st.integers(min_value=0, max_value=5), st.integers(min_value=0, max_value=5)
+        ),
+        values=st.sets(st.integers(min_value=0, max_value=20), max_size=8),
+        max_size=12,
+    )
+)
+def test_select_minimal_assertions_invariants(kill_map):
+    keep = ag._select_minimal_assertions(kill_map)
+
+    universe: set[int] = set()
+    for kills in kill_map.values():
+        universe |= kills
+
+    # Mutation score preserved: the kept assertions cover the whole universe.
+    covered: set[int] = set()
+    for key in keep:
+        covered |= kill_map[key]
+    assert covered == universe
+
+    # Only kill-bearing assertions are ever kept.
+    assert all(kill_map[key] for key in keep)
+
+    # Minimality: dropping any kept assertion loses coverage.
+    for key in keep:
+        others: set[int] = set()
+        for other in keep:
+            if other != key:
+                others |= kill_map[other]
+        assert not kill_map[key] <= others
diff --git a/tests/utils/test_configuration_writer.py b/tests/utils/test_configuration_writer.py
@@ -51,6 +51,7 @@ def expected_toml(tmp_path):
 float_precision = 0.01
 format_with_black = true
 no_xfail = false
+assertion_minimization = true
 
 [statistics_output]
 report_dir = "{REPORT_DIR}"
@@ -254,8 +255,8 @@ def expected_txt(tmp_path):
  'maximum_mutants=-1, post_process=True, '
  'minimization=Minimization(test_case_minimization_strategy=<MinimizationStrategy.CASE: '
  "'CASE'>, test_case_minimization_direction=<MinimizationDirection.BACKWARD: "
- "'BACKWARD'>), float_precision=0.01, format_with_black=True, no_xfail=False), "
- "algorithm=<Algorithm.RANDOM: 'RANDOM'>, "
+ "'BACKWARD'>), float_precision=0.01, format_with_black=True, no_xfail=False, "
+ "assertion_minimization=True), algorithm=<Algorithm.RANDOM: 'RANDOM'>, "
  "statistics_output=StatisticsOutputConfiguration(report_dir='{REPORT_DIR}', "
  "statistics_backend=<StatisticsBackend.CSV: 'CSV'>, "
  'timeline_interval=1000000000, timeline_interpolation=True, '
@@ -574,6 +575,8 @@ def expected_parameters() -> str:
 False
 --test_case_output.assertion_generation
 MUTATION_ANALYSIS
+--test_case_output.assertion_minimization
+True
 --test_case_output.export_strategy
 PY_TEST
 --test_case_output.float_precision
