FEAT Handle trial assertions through aggregate threshold evaluation

rampart/pytest_plugin/plugin.py

@@ -68,6 +68,7 @@
     "pytest_addoption",
     "pytest_collection_modifyitems",
     "pytest_configure",
+    "pytest_runtest_makereport",
     "pytest_sessionfinish",
     "pytest_terminal_summary",
     "pytest_testnodedown",
@@ -76,6 +77,7 @@
 
 _rampart_key = pytest.StashKey[RampartSession]()
 _session_start_key = pytest.StashKey[float]()
+_collector_key = pytest.StashKey[ResultCollector]()
 
 # Module-level constants are an acceptable exception in a hook-based
 # plugin module where there is no natural owning class.
@@ -413,6 +415,81 @@ def _absorb_results(
         )
 
 
+def _record_missing_trial_result(
+    *,
+    collector: ResultCollector,
+    call: pytest.CallInfo[None],
+) -> None:
+    """Record a synthetic result when a trial did not produce one.
+
+    Args:
+        collector (ResultCollector): The active trial result collector.
+        call (pytest.CallInfo[None]): The completed pytest call phase.
+    """
+    if collector.results:
+        return
+
+    if call.excinfo is None:
+        result = Result(
+            safe=True,
+            status=SafetyStatus.SAFE,
+            summary="Trial completed without recording a Result.",
+        )
+    else:
+        summary = str(call.excinfo.value) or "Trial assertion did not pass."
+        result = Result(
+            safe=False,
+            status=SafetyStatus.UNDETERMINED,
+            summary=summary,
+        )
+    collector.record(result=result)
+
+
+@pytest.hookimpl(wrapper=True, tryfirst=True)
+def pytest_runtest_makereport(
+    item: pytest.Item,
+    call: pytest.CallInfo[None],
+) -> Generator[None, pytest.TestReport, pytest.TestReport]:
+    """Treat call-phase assertions in trial clones as trial outcomes.
+
+    A trial assertion is an observation used by the aggregate threshold,
+    not an individual pytest failure. Assertions without an explicitly
+    recorded result synthesize SAFE or UNDETERMINED results. Failures from
+    other exceptions and from setup or teardown retain normal pytest
+    semantics.
+
+    Args:
+        item (pytest.Item): The pytest item being reported.
+        call (pytest.CallInfo[None]): Information about the completed phase.
+
+    Returns:
+        pytest.TestReport: The original or adjusted test report.
+    """
+    report = yield
+    if report.when != "call" or item.get_closest_marker("trial") is None:
+        return report
+
+    collector = item.stash.get(_collector_key, None)
+    if collector is None:
+        return report
+
+    if report.passed:
+        _record_missing_trial_result(collector=collector, call=call)
+        return report
+
+    if (
+        not report.failed
+        or call.excinfo is None
+        or not call.excinfo.errisinstance(AssertionError)
+    ):
+        return report
+
+    _record_missing_trial_result(collector=collector, call=call)
+    report.outcome = "passed"
+    report.longrepr = None
+    return report
+
+
 @pytest.fixture(autouse=True)
 def _rampart_collect(  # pytest discovers this via autouse=True
     request: pytest.FixtureRequest,
@@ -435,10 +512,18 @@ def _rampart_collect(  # pytest discovers this via autouse=True
     node = cast("pytest.Item", request.node)
     rampart_session = request.config.stash.get(_rampart_key, None)
     token = activate_collector(collector)
-    yield
-    deactivate_collector(token)
-    if rampart_session is not None:
-        _absorb_results(rampart_session=rampart_session, node=node, collector=collector)
+    node.stash[_collector_key] = collector
+    try:
+        yield
+    finally:
+        deactivate_collector(token)
+        if rampart_session is not None:
+            _absorb_results(
+                rampart_session=rampart_session,
+                node=node,
+                collector=collector,
+            )
+        del node.stash[_collector_key]
 
     # Note: collector.results returns a copy of the internal list,
     # so reading it after deactivation and absorption is safe.

tests/unit/pytest_plugin/test_xdist_aggregation.py

@@ -119,6 +119,142 @@ def test_baseline_emits_one_report(self, pytester: Pytester) -> None:
         assert reports[0]["total_runs"] == 4
 
 
+class TestTrialAssertionInterception:
+    def test_assertion_misses_pass_when_threshold_is_met(
+        self,
+        pytester: Pytester,
+    ) -> None:
+        pytester.makeconftest(_CONFTEST)
+        pytester.makepyfile(  # pyright: ignore[reportUnknownMemberType]
+            test_trial_assert="""
+            import pytest
+            from rampart.core.result import Result, SafetyStatus
+
+            @pytest.mark.trial(n=5, threshold=0.6)
+            def test_trial_assert(request):
+                safe = not request.node.name.endswith(
+                    ("[trial-3]", "[trial-4]"),
+                )
+                result = Result(
+                    safe=safe,
+                    status=SafetyStatus.SAFE if safe else SafetyStatus.UNSAFE,
+                    summary="safe" if safe else "trial miss",
+                )
+                assert result, result.summary
+            """,
+        )
+
+        result = pytester.runpytest("-p", "no:cacheprovider")
+
+        result.assert_outcomes(passed=5)
+        assert result.ret == pytest.ExitCode.OK
+        report = _load_reports(pytester)[0]
+        assert report["total_runs"] == 5
+        assert report["passed"] == 3
+        assert report["undetermined"] == 2
+        assert "PASS  test_trial_assert [3/5 safe" in "\n".join(result.outlines)
+
+    def test_assertion_misses_fail_only_when_threshold_is_missed(
+        self,
+        pytester: Pytester,
+    ) -> None:
+        pytester.makeconftest(_CONFTEST)
+        pytester.makepyfile(  # pyright: ignore[reportUnknownMemberType]
+            test_trial_assert="""
+            import pytest
+            from rampart.core.result import Result, SafetyStatus
+
+            @pytest.mark.trial(n=5, threshold=0.8)
+            def test_trial_assert(request):
+                safe = not request.node.name.endswith(
+                    ("[trial-3]", "[trial-4]"),
+                )
+                result = Result(
+                    safe=safe,
+                    status=SafetyStatus.SAFE if safe else SafetyStatus.UNSAFE,
+                    summary="safe" if safe else "trial miss",
+                )
+                assert result, result.summary
+            """,
+        )
+
+        result = pytester.runpytest("-p", "no:cacheprovider")
+
+        result.assert_outcomes(passed=5)
+        assert result.ret == pytest.ExitCode.TESTS_FAILED
+        assert "FAIL  test_trial_assert [3/5 safe" in "\n".join(result.outlines)
+
+    def test_assertion_does_not_duplicate_an_existing_result(
+        self,
+        pytester: Pytester,
+    ) -> None:
+        pytester.makeconftest(_CONFTEST)
+        pytester.makepyfile(  # pyright: ignore[reportUnknownMemberType]
+            test_trial_assert="""
+            import pytest
+            from rampart import record_result
+            from rampart.core.result import Result, SafetyStatus
+
+            @pytest.mark.trial(n=4, threshold=0.5)
+            def test_trial_assert(request):
+                safe = not request.node.name.endswith("[trial-3]")
+                result = Result(
+                    safe=safe,
+                    status=SafetyStatus.SAFE if safe else SafetyStatus.UNSAFE,
+                    summary="safe" if safe else "trial miss",
+                )
+                record_result(result)
+                assert result, result.summary
+            """,
+        )
+
+        result = pytester.runpytest("-p", "no:cacheprovider")
+
+        result.assert_outcomes(passed=4)
+        assert result.ret == pytest.ExitCode.TESTS_FAILED
+        report = _load_reports(pytester)[0]
+        assert report["total_runs"] == 4
+        assert report["passed"] == 3
+        assert report["failed"] == 1
+
+    def test_non_assert_exception_remains_a_pytest_failure(
+        self,
+        pytester: Pytester,
+    ) -> None:
+        pytester.makeconftest(_CONFTEST)
+        pytester.makepyfile(  # pyright: ignore[reportUnknownMemberType]
+            test_trial_error="""
+            import pytest
+
+            @pytest.mark.trial(n=1, threshold=0.0)
+            def test_trial_error():
+                raise RuntimeError("broken trial harness")
+            """,
+        )
+
+        result = pytester.runpytest("-p", "no:cacheprovider")
+
+        result.assert_outcomes(failed=1)
+        assert result.ret == pytest.ExitCode.TESTS_FAILED
+
+    def test_non_trial_assertion_remains_a_pytest_failure(
+        self,
+        pytester: Pytester,
+    ) -> None:
+        pytester.makeconftest(_CONFTEST)
+        pytester.makepyfile(  # pyright: ignore[reportUnknownMemberType]
+            test_plain_assert="""
+            def test_plain_assert():
+                assert False, "ordinary pytest failure"
+            """,
+        )
+
+        result = pytester.runpytest("-p", "no:cacheprovider")
+
+        result.assert_outcomes(failed=1)
+        assert result.ret == pytest.ExitCode.TESTS_FAILED
+
+
 class TestXdistConsolidation:
     def test_xdist_emits_single_consolidated_report(
         self,
@@ -228,14 +364,13 @@ def test_trial_group_fails_when_any_unsafe_under_loadgroup(
         self,
         pytester: Pytester,
     ) -> None:
-        """An UNSAFE trial fails the whole group regardless of pass rate.
+        """An UNSAFE trial fails the group regardless of pass rate.
 
         Trial body switches on the clone name (``[trial-0]``..``[trial-3]``)
         so the same outcome distribution is produced regardless of which
         worker executes the clone. Three trials are SAFE and one is UNSAFE;
         with threshold=0.5 the group would otherwise pass on rate alone,
-        so the only way the group can FAIL is if controller-side
-        aggregation correctly merged the worker results.
+        so failure proves controller-side aggregation merged worker results.
         """
         pytester.makeconftest(_CONFTEST)
         pytester.makepyfile(  # pyright: ignore[reportUnknownMemberType]
@@ -249,7 +384,7 @@ def test_trial_group_fails_when_any_unsafe_under_loadgroup(
             @pytest.mark.trial(n=4, threshold=0.5)
             def test_trial_mixed(request):
                 # Trial-3 is UNSAFE; the rest are SAFE. With threshold=0.5
-                # the group MUST FAIL on the unconditional unsafe rule.
+                # the group still fails because UNSAFE is unconditional.
                 unsafe = request.node.name.endswith("[trial-3]")
                 record_result(Result(
                     safe=not unsafe,
@@ -294,8 +429,8 @@ def test_trial_group_fails_when_any_unsafe_under_load(
 
         The PR docs claim aggregation remains correct under --dist=load
         because the controller merges all worker results. This test
-        protects that contract: an UNSAFE clone produced on any worker
-        must propagate into the controller's trial-group verdict.
+        protects that contract while applying the threshold to an
+        UNSAFE clone produced on a worker.
         """
         pytester.makeconftest(_CONFTEST)
         pytester.makepyfile(  # pyright: ignore[reportUnknownMemberType]