diff --git a/rampart/pytest_plugin/plugin.py b/rampart/pytest_plugin/plugin.py index 9c30b4c..4268b96 100644 --- a/rampart/pytest_plugin/plugin.py +++ b/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. diff --git a/tests/unit/pytest_plugin/test_xdist_aggregation.py b/tests/unit/pytest_plugin/test_xdist_aggregation.py index 6c60159..d39624e 100644 --- a/tests/unit/pytest_plugin/test_xdist_aggregation.py +++ b/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]