diff --git a/rampart/pytest_plugin/_collection.py b/rampart/pytest_plugin/_collection.py index 3701ecf1..cc54bf31 100644 --- a/rampart/pytest_plugin/_collection.py +++ b/rampart/pytest_plugin/_collection.py @@ -58,6 +58,16 @@ def deactivate_collector(token: Token[ResultCollector | None]) -> None: _active_collector.reset(token) +def get_active_collector() -> ResultCollector | None: + """Return the collector active for the current test, if any. + + Returns: + ResultCollector | None: The active collector, or None when no + test collector is installed on the current context. + """ + return _active_collector.get() + + class ResultCollector: """Accumulates Result objects produced during a single test. diff --git a/rampart/pytest_plugin/plugin.py b/rampart/pytest_plugin/plugin.py index 0531ed66..337b10d5 100644 --- a/rampart/pytest_plugin/plugin.py +++ b/rampart/pytest_plugin/plugin.py @@ -41,6 +41,7 @@ ResultCollector, activate_collector, deactivate_collector, + get_active_collector, ) from rampart.pytest_plugin._session import RampartSession from rampart.pytest_plugin._xdist import ( @@ -69,15 +70,20 @@ "pytest_addoption", "pytest_collection_modifyitems", "pytest_configure", + "pytest_runtest_makereport", "pytest_sessionfinish", "pytest_terminal_summary", "pytest_testnodedown", "pytest_unconfigure", ] +# Config-scoped stash keys: one entry per pytest session. _rampart_key = pytest.StashKey[RampartSession]() _session_start_key = pytest.StashKey[float]() +# Item-scoped stash key: a per-test snapshot consumed by xdist streaming. +_call_results_key = pytest.StashKey[list[Result]]() + # Module-level constants are an acceptable exception in a hook-based # plugin module where there is no natural owning class. @@ -454,6 +460,42 @@ def _rampart_collect( # pytest discovers this via autouse=True ) +@pytest.hookimpl(wrapper=True) +def pytest_runtest_makereport( + item: pytest.Item, + call: pytest.CallInfo[None], +) -> Generator[None, pytest.TestReport, pytest.TestReport]: + """Snapshot the active collector's results at the call phase. + + On xdist workers, copies the per-test results onto the item stash + while the collector is still active — before the autouse fixture's + teardown drains and deactivates it. Downstream xdist streaming reads + this snapshot to deliver results per test rather than in a single + end-of-worker batch. + + Restricted to worker processes: single-process and controller runs + never consume the snapshot, so skipping it there keeps those paths + allocation-free and byte-identical to their pre-snapshot behavior. + + Runs as a wrapper so it is never skipped by the firstresult builtin + and always returns the report unchanged. + + Args: + item (pytest.Item): The test item being reported. + call (pytest.CallInfo[None]): The call phase information. + + Returns: + Generator[None, pytest.TestReport, pytest.TestReport]: The + unchanged report produced by downstream hookimpls. + """ + report = yield + if call.when == "call" and is_xdist_worker(config=item.config): + collector = get_active_collector() + if collector is not None: + item.stash[_call_results_key] = collector.results + return report + + def _has_sink_hook_impl(*, config: pytest.Config) -> bool: """Return True if any plugin implements ``pytest_rampart_sinks``. diff --git a/tests/unit/pytest_plugin/test_collection.py b/tests/unit/pytest_plugin/test_collection.py index a2ce68cf..4aeda58c 100644 --- a/tests/unit/pytest_plugin/test_collection.py +++ b/tests/unit/pytest_plugin/test_collection.py @@ -5,6 +5,7 @@ from __future__ import annotations +import asyncio from unittest.mock import MagicMock from rampart.core.execution import ExecutionEvent, ExecutionEventData @@ -15,6 +16,7 @@ _active_collector, activate_collector, deactivate_collector, + get_active_collector, record_result, ) @@ -190,3 +192,49 @@ def test_nested_activation(self) -> None: assert len(outer.results) == 1 deactivate_collector(outer_token) + + +class TestGetActiveCollector: + """get_active_collector exposes the ContextVar-scoped collector.""" + + def test_returns_none_when_inactive(self) -> None: + token = _active_collector.set(None) + try: + assert get_active_collector() is None + finally: + _active_collector.reset(token) + + def test_returns_active_collector(self) -> None: + collector = ResultCollector() + token = activate_collector(collector) + try: + assert get_active_collector() is collector + finally: + deactivate_collector(token) + + def test_returns_none_after_deactivate(self) -> None: + baseline = _active_collector.set(None) + try: + collector = ResultCollector() + token = activate_collector(collector) + deactivate_collector(token) + assert get_active_collector() is None + finally: + _active_collector.reset(baseline) + + async def test_sees_results_from_child_task_async(self) -> None: + collector = ResultCollector() + token = activate_collector(collector) + + async def _body() -> None: + await asyncio.sleep(0) + record_result(result=_make_result(summary="child")) + + try: + await asyncio.create_task(_body()) + active = get_active_collector() + assert active is collector + assert len(active.results) == 1 + assert active.results[0].summary == "child" + finally: + deactivate_collector(token) diff --git a/tests/unit/pytest_plugin/test_plugin.py b/tests/unit/pytest_plugin/test_plugin.py index d5de5211..2af144fb 100644 --- a/tests/unit/pytest_plugin/test_plugin.py +++ b/tests/unit/pytest_plugin/test_plugin.py @@ -5,6 +5,7 @@ from __future__ import annotations +from types import SimpleNamespace from typing import TYPE_CHECKING, Any, cast from unittest.mock import AsyncMock, MagicMock @@ -12,9 +13,16 @@ from rampart.core.result import Result, SafetyStatus from rampart.core.types import ObservabilityLevel -from rampart.pytest_plugin._collection import ResultCollectionHandler, ResultCollector +from rampart.pytest_plugin._collection import ( + ResultCollectionHandler, + ResultCollector, + _active_collector, + activate_collector, + deactivate_collector, +) from rampart.pytest_plugin._session import RampartSession from rampart.pytest_plugin.plugin import ( + _call_results_key, _emit_sinks, _enforce_incomplete_exit_status, _evaluate_gates, @@ -26,6 +34,7 @@ _write_trial_group_lines, pytest_collection_modifyitems, pytest_configure, + pytest_runtest_makereport, pytest_sessionfinish, pytest_terminal_summary, pytest_unconfigure, @@ -969,3 +978,111 @@ def test_incomplete_run_does_not_mask_existing_failure(self) -> None: rampart_session=rampart_session, ) assert session.exitstatus == pytest.ExitCode.INTERRUPTED + + +def _make_result(*, summary: str = "result") -> Result: + """Build a minimal Result for makereport tests.""" + return Result(status=SafetyStatus.SAFE, summary=summary) + + +def _make_reporting_item(*, worker: bool = True) -> Any: + """Build a mock pytest.Item backed by a real Stash. + + Defaults to a worker-like config so the call-phase snapshot fires; + pass worker=False for a single-process or controller config. + """ + item = MagicMock() + item.stash = pytest.Stash() + if worker: + item.config = SimpleNamespace(workerinput={"workerid": "gw0"}) + else: + item.config = SimpleNamespace() + return item + + +def _drive_makereport(*, item: Any, when: str, report: Any = None) -> Any: + """Drive the makereport wrapper generator and return its result.""" + call = MagicMock() + call.when = when + sent = report if report is not None else MagicMock() + gen = pytest_runtest_makereport( + item=cast("pytest.Item", item), + call=cast("pytest.CallInfo[None]", call), + ) + next(gen) + try: + gen.send(sent) + except StopIteration as stop: + return stop.value + raise AssertionError("makereport wrapper did not return") + + +class TestPytestRuntestMakereport: + """The makereport hook snapshots collector results at the call phase.""" + + def test_snapshots_results_at_call_phase(self) -> None: + item = _make_reporting_item() + collector = ResultCollector() + collector.record(result=_make_result(summary="captured")) + token = activate_collector(collector) + try: + _drive_makereport(item=item, when="call") + finally: + deactivate_collector(token) + + snapshot = item.stash[_call_results_key] + assert len(snapshot) == 1 + assert snapshot[0].summary == "captured" + + def test_snapshots_empty_list_when_no_results(self) -> None: + item = _make_reporting_item() + collector = ResultCollector() + token = activate_collector(collector) + try: + _drive_makereport(item=item, when="call") + finally: + deactivate_collector(token) + + assert item.stash[_call_results_key] == [] + + def test_no_snapshot_at_setup_phase(self) -> None: + item = _make_reporting_item() + collector = ResultCollector() + collector.record(result=_make_result()) + token = activate_collector(collector) + try: + _drive_makereport(item=item, when="setup") + finally: + deactivate_collector(token) + + assert _call_results_key not in item.stash + + def test_no_snapshot_when_no_collector_active(self) -> None: + item = _make_reporting_item() + token = _active_collector.set(None) + try: + _drive_makereport(item=item, when="call") + finally: + _active_collector.reset(token) + + assert _call_results_key not in item.stash + + def test_no_snapshot_when_not_xdist_worker(self) -> None: + item = _make_reporting_item(worker=False) + collector = ResultCollector() + collector.record(result=_make_result()) + token = activate_collector(collector) + try: + _drive_makereport(item=item, when="call") + finally: + deactivate_collector(token) + + assert _call_results_key not in item.stash + + def test_returns_report_unchanged(self) -> None: + item = _make_reporting_item() + report = object() + + returned = _drive_makereport(item=item, when="call", report=report) + + assert returned is report