Swap order of assertion generation and minimization

Author: LuKrO2011

src/pynguin/ga/postprocess.py

@@ -19,17 +19,53 @@
 import pynguin.ga.testsuitechromosome as tsc
 import pynguin.testcase.testcase as tc
 import pynguin.testcase.testcasevisitor as tcv
-from pynguin.assertion.assertion import Assertion, ExceptionAssertion
+from pynguin.assertion.assertion import (
+    Assertion,
+    ExceptionAssertion,
+    ReferenceAssertion,
+)
 from pynguin.testcase.statement import StatementVisitor
 from pynguin.utils.orderedset import OrderedSet
 
 if TYPE_CHECKING:
     import pynguin.ga.computations as ff
+    import pynguin.testcase.variablereference as vr
     from pynguin.testcase.execution import SubprocessTestCaseExecutor
 
 _LOGGER = logging.getLogger(__name__)
 
 
+def get_assertion_protected_variables(test_case: tc.TestCase) -> set[vr.VariableReference]:
+    """Get all variables that should be protected due to assertions.
+
+    Variables are protected if they are:
+    - Directly referenced by a ReferenceAssertion's source, OR
+    - In the backward dependency chain of an asserted variable
+
+    ExceptionAssertions are skipped (no source variable).
+
+    Args:
+        test_case: Test case to analyze
+
+    Returns:
+        Set of variable references that should not be removed during minimization
+    """
+    protected: set[vr.VariableReference] = set()
+    for stmt in test_case.statements:
+        for assertion in stmt.assertions:
+            # Skip ExceptionAssertion - has no source
+            if isinstance(assertion, ExceptionAssertion):
+                continue
+            # Handle ReferenceAssertion
+            if isinstance(assertion, ReferenceAssertion):
+                var_ref = assertion.source.get_variable_reference()
+                if var_ref is not None:
+                    protected.add(var_ref)
+                    # Add all backward dependencies
+                    protected.update(test_case.get_dependencies(var_ref))
+    return protected
+
+
 class ExceptionTruncation(cv.ChromosomeVisitor):
     """Truncates test cases after an exception-raising statement."""
 
@@ -201,7 +237,9 @@ class ForwardIterativeMinimizationVisitor(IterativeMinimizationVisitor):
     4. If fitness remains the same or improves, remove the statement from the original test case
     """
 
-    def visit_default_test_case(self, test_case: tc.TestCase) -> None:  # noqa: D102
+    def visit_default_test_case(  # noqa: D102, PLR0914
+        self, test_case: tc.TestCase
+    ) -> None:
         original_test_case = tcc.TestCaseChromosome(test_case=test_case)
         original_test_suite = tsc.TestSuiteChromosome()
         original_test_suite.add_test_case_chromosome(original_test_case)
@@ -235,13 +273,18 @@ def visit_default_test_case(self, test_case: tc.TestCase) -> None:  # noqa: D102
                     for fitness_function in self._fitness_functions
                 ]
                 if all(map(math.isclose, original_coverages, minimized_coverages)):
-                    removed = test_case.remove_statement_with_forward_dependencies(stmt)
-                    self._removed_statements += len(removed)
+                    protected_vars = get_assertion_protected_variables(test_case)
+                    is_statement_protected = stmt.ret_val in protected_vars
+                    if not is_statement_protected:
+                        removed = test_case.remove_statement_with_forward_dependencies(stmt)
+                        self._removed_statements += len(removed)
 
-                    # Update the statements list to reflect the changes in the test case
-                    statements = list(test_case.statements)
-                    # Don't increment i since we've removed elements and the list has shifted
-                    statements_changed = True
+                        # Update the statements list to reflect the changes in the test case
+                        statements = list(test_case.statements)
+                        # Don't increment i since we've removed elements and the list has shifted
+                        statements_changed = True
+                    else:
+                        i += 1
                 else:
                     i += 1
 
@@ -295,11 +338,16 @@ def visit_default_test_case(self, test_case: tc.TestCase) -> None:  # noqa: D102
                     for fitness_function in self._fitness_functions
                 ]
                 if all(map(math.isclose, original_coverages, minimized_coverages)):
-                    removed = test_case.remove_statement_with_forward_dependencies(stmt)
-                    self._removed_statements += len(removed)
-                    statements_changed = True
-                    break
-                i -= 1
+                    protected_vars = get_assertion_protected_variables(test_case)
+                    is_statement_protected = stmt.ret_val in protected_vars
+                    if not is_statement_protected:
+                        removed = test_case.remove_statement_with_forward_dependencies(stmt)
+                        self._removed_statements += len(removed)
+                        statements_changed = True
+                        break
+                    i -= 1
+                else:
+                    i -= 1
 
         _LOGGER.debug(
             "Removed %s statement(s) from test case using backward iterative minimization",

src/pynguin/generator.py

@@ -543,13 +543,6 @@ def _track_final_metrics(
             runtime_variable, generation_result.get_coverage_for(coverage_ff)
         )
 
-    ass_gen = config.configuration.test_case_output.assertion_generation
-    if (
-        ass_gen == config.AssertionGenerator.CHECKED_MINIMIZING
-        and RuntimeVariable.AssertionCheckedCoverage in output_variables
-    ):
-        _minimize_assertions(generation_result)
-
     # Collect other final stats on result
     stat.track_output_variable(RuntimeVariable.FinalLength, generation_result.length())
     stat.track_output_variable(RuntimeVariable.FinalSize, generation_result.size())
@@ -626,12 +619,27 @@ def _run() -> ReturnCode:  # noqa: C901
     executor.clear_remote_observers()
 
     _track_search_metrics(algorithm, generation_result, coverage_metrics)
+
+    # Generate assertions FIRST
+    _generate_assertions(executor, generation_result, test_cluster)
+
+    # Minimize assertions if configured (requires re-instrumentation for checked_instructions)
+    ass_gen = config.configuration.test_case_output.assertion_generation
+    if (
+        ass_gen == config.AssertionGenerator.CHECKED_MINIMIZING
+        and RuntimeVariable.AssertionCheckedCoverage
+        in config.configuration.statistics_output.output_variables
+    ):
+        _prepare_for_assertion_minimization(executor, generation_result, constant_provider)
+        _LOGGER.info("Minimizing assertions")
+        _minimize_assertions(generation_result)
+
+    # Statement minimization LAST (now assertion-aware)
     try:
         _LOGGER.info("Minimizing test cases")
         _minimize(generation_result, algorithm)
     except Exception as ex:
         _LOGGER.exception("Minimization failed: %s", ex)
-    _generate_assertions(executor, generation_result, test_cluster)
 
     if (
         tracked_metrics := _track_final_metrics(
@@ -841,6 +849,55 @@ def _minimize_assertions(generation_result: tsc.TestSuiteChromosome):
     )
 
 
+def _prepare_for_assertion_minimization(
+    executor: TestCaseExecutor,
+    generation_result: tsc.TestSuiteChromosome,
+    constant_provider: ConstantProvider,
+) -> None:
+    """Prepare for assertion minimization by populating checked_instructions.
+
+    Re-instruments the module with RemoteAssertionExecutionObserver,
+    executes the test suite to populate checked_instructions on assertions,
+    then returns executor to non-instrumented state.
+
+    Args:
+        executor: Test case executor
+        generation_result: Test suite with assertions to minimize
+        constant_provider: Constant provider for re-instrumentation
+    """
+    # Add assertion execution observer
+    executor.set_instrument(True)
+    executor.add_remote_observer(RemoteAssertionExecutionObserver())
+
+    # Re-instrument with CHECKED metric
+    metrics_for_reinstrumentation = {config.CoverageMetric.CHECKED}
+    dynamic_constant_provider = None
+    if isinstance(constant_provider, DynamicConstantProvider):
+        dynamic_constant_provider = constant_provider
+
+    if not _reload_instrumentation_loader(
+        metrics_for_reinstrumentation,
+        dynamic_constant_provider,
+        executor.subject_properties,
+    ):
+        _LOGGER.warning("Failed to reload instrumentation for assertion minimization")
+        return
+
+    # Force re-execution to populate checked_instructions
+    _reset_cache_for_result(generation_result)
+
+    # Execute to populate checked_instructions
+    # This happens automatically when coverage is computed
+    assertion_checked_ff = ff.TestSuiteAssertionCheckedCoverageFunction(executor)
+    generation_result.add_coverage_function(assertion_checked_ff)
+    _ = generation_result.get_coverage_for(assertion_checked_ff)
+
+    # Return executor to non-instrumented state
+    executor.set_instrument(False)
+    executor.clear_observers()
+    executor.clear_remote_observers()
+
+
 _strategies: dict[config.MutationStrategy, Callable[[int], ms.HOMStrategy]] = {
     config.MutationStrategy.FIRST_TO_LAST: ms.FirstToLastHOMStrategy,
     config.MutationStrategy.BETWEEN_OPERATORS: ms.BetweenOperatorsHOMStrategy,

tests/ga/test_postprocess.py

@@ -17,8 +17,9 @@
 import pynguin.testcase.defaulttestcase as dtc
 import pynguin.testcase.statement as stmt
 from pynguin.analyses.module import ModuleTestCluster, generate_test_cluster
-from pynguin.assertion.assertion import ExceptionAssertion
+from pynguin.assertion.assertion import ExceptionAssertion, ObjectAssertion
 from pynguin.ga.computations import TestSuiteBranchCoverageFunction, TestSuiteLineCoverageFunction
+from pynguin.ga.postprocess import get_assertion_protected_variables
 from pynguin.ga.testsuitechromosome import TestSuiteChromosome
 from pynguin.instrumentation.machinery import install_import_hook
 from pynguin.instrumentation.tracer import SubjectProperties
@@ -122,6 +123,64 @@ def test_test_case_assertion_minimization_does_not_remove_empty_assertion(
     assert default_test_case.get_assertions() == [assertion_1]
 
 
+def test_get_assertion_protected_variables_no_assertions(default_test_case):
+    """Test that a test case with no assertions returns empty protected set."""
+    statement = stmt.IntPrimitiveStatement(default_test_case)
+    default_test_case.add_statement(statement)
+
+    result = get_assertion_protected_variables(default_test_case)
+
+    assert result == set()
+
+
+def test_get_assertion_protected_variables_single_reference_assertion(default_test_case):
+    """Test that ReferenceAssertion source variable is protected."""
+    statement = stmt.IntPrimitiveStatement(default_test_case, value=42)
+    default_test_case.add_statement(statement)
+
+    # Create ObjectAssertion on the statement's return value
+    var_ref = statement.ret_val
+    assertion = ObjectAssertion(var_ref, 42)  # Asserts var_ref == 42
+    statement.add_assertion(assertion)
+
+    result = get_assertion_protected_variables(default_test_case)
+
+    assert var_ref in result
+
+
+def test_get_assertion_protected_variables_skips_exception_assertion(default_test_case):
+    """Test that ExceptionAssertion (no source) doesn't add to protected set."""
+    statement = stmt.IntPrimitiveStatement(default_test_case)
+    default_test_case.add_statement(statement)
+
+    # ExceptionAssertion has no source variable
+    exc_assertion = MagicMock(spec=ExceptionAssertion)
+    statement.add_assertion(exc_assertion)
+
+    result = get_assertion_protected_variables(default_test_case)
+
+    assert result == set()
+
+
+def test_get_assertion_protected_variables_with_dependencies(default_test_case):
+    """Test that assertion source and its dependencies are protected."""
+    # Create a statement
+    statement = stmt.IntPrimitiveStatement(default_test_case, value=42)
+    default_test_case.add_statement(statement)
+    var_ref = statement.ret_val
+
+    # Add assertion
+    assertion = ObjectAssertion(var_ref, 42)
+    statement.add_assertion(assertion)
+
+    result = get_assertion_protected_variables(default_test_case)
+
+    # var_ref should be protected
+    assert var_ref in result
+    # Since IntPrimitiveStatement has no dependencies, only var_ref is protected
+    # This is correct behavior - primitives have no backward deps
+
+
 def test_test_case_postprocessor_suite():
     dummy_visitor = MagicMock()
     tcpp = pp.TestCasePostProcessor([dummy_visitor])