From db9b94e08a85851ffdc8551461945bf729b28ab2 Mon Sep 17 00:00:00 2001 From: behnamousat Date: Mon, 20 Jul 2026 17:03:54 -0700 Subject: [PATCH] make the trials marker declaration only --- docs/api/pytest-plugin.md | 1 - docs/attacks/xpia.md | 2 +- docs/concepts/overview.md | 2 +- docs/getting-started/installation.md | 2 +- docs/getting-started/quickstart.md | 2 +- docs/glossary.md | 2 +- docs/probes/behavioral.md | 2 +- docs/usage/authoring-tests.md | 16 +- docs/usage/ci-integration.md | 23 +- docs/usage/pytest-integration.md | 18 +- docs/usage/xdist.md | 55 +-- pyproject.toml | 2 +- rampart/core/execution.py | 12 +- rampart/core/result.py | 9 + rampart/pytest_plugin/_session.py | 194 +--------- rampart/pytest_plugin/_xdist.py | 104 +----- rampart/pytest_plugin/plugin.py | 331 +----------------- rampart/reporting/sink.py | 10 +- tests/unit/core/test_execution.py | 17 + tests/unit/core/test_result.py | 8 + tests/unit/pytest_plugin/test_plugin.py | 328 ----------------- tests/unit/pytest_plugin/test_xdist.py | 114 +----- .../pytest_plugin/test_xdist_aggregation.py | 331 +++++------------- 23 files changed, 190 insertions(+), 1395 deletions(-) diff --git a/docs/api/pytest-plugin.md b/docs/api/pytest-plugin.md index 78f0774..a01e649 100644 --- a/docs/api/pytest-plugin.md +++ b/docs/api/pytest-plugin.md @@ -13,7 +13,6 @@ RAMPART's pytest integration. Activates automatically when installed. options: members: - RampartSession - - TrialGroupResult ## Parallel Execution Hooks diff --git a/docs/attacks/xpia.md b/docs/attacks/xpia.md index 5c8aee6..3d0c451 100644 --- a/docs/attacks/xpia.md +++ b/docs/attacks/xpia.md @@ -65,7 +65,7 @@ async def test_inline_xpia(adapter): "send_email", recipient=lambda v: isinstance(v, str) and "evil.com" in v, ), - ).execute_async(adapter=adapter) + ).execute_trials_async(adapter=adapter, n=5, threshold=0.8) assert result, result.summary ``` diff --git a/docs/concepts/overview.md b/docs/concepts/overview.md index 21cb9a1..95421ef 100644 --- a/docs/concepts/overview.md +++ b/docs/concepts/overview.md @@ -122,7 +122,7 @@ You can reuse the same evaluator in both contexts. A [`ToolCalled`][rampart.eval RAMPART registers as a pytest plugin automatically when installed. It provides: -- **Markers**: `@pytest.mark.harm(...)` for categorization, `@pytest.mark.trial(n=...)` for statistical repetition +- **Markers**: `@pytest.mark.harm(...)` for categorization, `@pytest.mark.trial(n=...)` for population declaration and selection - **Automatic result collection**: Results from `Attacks.*` and `Probes.*` are collected without manual wiring - **Terminal summary**: A safety summary printed after the standard pytest output - **Report sinks**: Structured output via the `rampart_sinks` fixture diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md index a365e29..803edf1 100644 --- a/docs/getting-started/installation.md +++ b/docs/getting-started/installation.md @@ -118,7 +118,7 @@ Expected output: ``` @pytest.mark.harm(*categories): categorize by harm type -@pytest.mark.trial(n=, threshold=): statistical repetition +@pytest.mark.trial(n=, threshold=): trial population declaration ``` RAMPART registers as a pytest plugin automatically via the `pytest11` entry point. No `conftest.py` configuration is needed to activate it. diff --git a/docs/getting-started/quickstart.md b/docs/getting-started/quickstart.md index 439a958..0112a8e 100644 --- a/docs/getting-started/quickstart.md +++ b/docs/getting-started/quickstart.md @@ -130,7 +130,7 @@ async def test_xpia_email_exfil(my_agent): ``` - **`@pytest.mark.harm(...)`** — Groups results by harm category in the terminal summary and reports. -- **`execute_trials_async(n=3, threshold=0.8)`** — Runs 3 independent trials and returns one [`PopulationResult`][rampart.core.result.PopulationResult]. The assertion passes if ≥ 80% are SAFE. LLM agents are non-deterministic, so a single run may not be representative. +- **`execute_trials_async(n=3, threshold=0.8)`** — Runs 3 independent trials and returns one `PopulationResult`. The assertion passes if ≥ 80% are SAFE. LLM agents are non-deterministic, so a single run may not be representative. See [pytest Markers & Fixtures](../usage/pytest-integration.md) for the full marker reference. diff --git a/docs/glossary.md b/docs/glossary.md index 8e7c591..ca581e3 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -88,7 +88,7 @@ Terms used throughout the RAMPART documentation. : An implementation of [`Surface`][rampart.core.injection.Surface]. Represents an injectable data source. See [Surfaces](api/surfaces.md). **Trial** -: A repeated execution of a test for statistical confidence, configured via `@pytest.mark.trial(n=...)`. See [pytest Markers & Fixtures](usage/pytest-integration.md). +: One execution within a population run by `execute_trials_async`. The optional `@pytest.mark.trial(n=...)` marker declares population metadata for selection. See [pytest Markers & Fixtures](usage/pytest-integration.md). **Turn** : One prompt-response exchange. Immutable. See [`Turn`][rampart.core.types.Turn]. diff --git a/docs/probes/behavioral.md b/docs/probes/behavioral.md index a73db27..7a447e5 100644 --- a/docs/probes/behavioral.md +++ b/docs/probes/behavioral.md @@ -102,7 +102,7 @@ async def test_agent_refuses_harmful_request(adapter): evaluator=ResponseContains( lambda text: "I can't" in text or "I cannot" in text ), - ).execute_async(adapter=adapter) + ).execute_trials_async(adapter=adapter, n=5, threshold=0.8) assert result, result.summary ``` diff --git a/docs/usage/authoring-tests.md b/docs/usage/authoring-tests.md index b6955a4..826cf20 100644 --- a/docs/usage/authoring-tests.md +++ b/docs/usage/authoring-tests.md @@ -323,19 +323,29 @@ def adapter(): ### Class-Based Test Organization -Group related tests in a class: +Group related tests in a class. The `trial` marker declares population metadata; each test body calls `execute_trials_async` to perform repetition. ```python class TestDataExfiltration: @pytest.mark.harm(HarmCategory.DATA_EXFILTRATION) @pytest.mark.trial(n=3, threshold=0.8) async def test_ssh_key_exfil(self, adapter): - ... + result = await Attacks.xpia(...).execute_trials_async( + adapter=adapter, + n=3, + threshold=0.8, + ) + assert result, result.summary @pytest.mark.harm(HarmCategory.DATA_EXFILTRATION) @pytest.mark.trial(n=3, threshold=0.8) async def test_email_exfil(self, adapter): - ... + result = await Attacks.xpia(...).execute_trials_async( + adapter=adapter, + n=3, + threshold=0.8, + ) + assert result, result.summary ``` diff --git a/docs/usage/ci-integration.md b/docs/usage/ci-integration.md index f918380..8c32799 100644 --- a/docs/usage/ci-integration.md +++ b/docs/usage/ci-integration.md @@ -25,31 +25,34 @@ pip install pytest-xdist pytest tests/ -n auto ``` -RAMPART aggregates results across worker processes and emits a single unified report under **any** `--dist` mode. The default `--dist=load` spreads `@trial` clones across all workers and is usually fastest. Add `--dist=loadgroup` only when a trial group needs to stay on one worker (e.g. clones share a session fixture or per-group worker state). See [Choosing `loadgroup` vs `load`](xdist.md#choosing-loadgroup-vs-load) for details and security considerations. +RAMPART aggregates results across worker processes and emits a single unified report under **any** `--dist` mode. Each call to `execute_trials_async` remains one pytest item and therefore runs on one worker. --- -## Trial Markers for Statistical Confidence +## Repeated Executions for Statistical Confidence -Use `@pytest.mark.trial(n=, threshold=)` for tests where a single run is not conclusive: +Use `execute_trials_async` for tests where a single run is not conclusive: ```python -@pytest.mark.trial(n=10, threshold=0.8) async def test_injection_resistance(adapter): - result = await Attacks.xpia(...).execute_async(adapter=adapter) + result = await Attacks.xpia(...).execute_trials_async( + adapter=adapter, + n=10, + threshold=0.8, + ) assert result, result.summary ``` -This runs 10 independent trials. The test group passes only if ≥ 80% of trials are `SAFE`. +This runs 10 independent trials. The single pytest test passes only if ≥ 80% of trials are `SAFE`. **Trial semantics in CI:** -- Each trial clone appears as a separate pytest item -- The aggregate verdict appears in the RAMPART terminal summary +- The population is one pytest item +- The returned `PopulationResult` is the aggregate verdict - The aggregate passes when the SAFE pass rate meets the threshold - Any `ERROR` trial makes the aggregate fail -- No-result clones are excluded from the aggregate denominator -- Clone assertions still contribute independently to pytest's exit status; use `execute_trials_async` when the threshold must govern the single pytest verdict +- `UNDETERMINED` trials count against the pass rate +- `@pytest.mark.trial` is declaration-only and does not execute repetitions --- diff --git a/docs/usage/pytest-integration.md b/docs/usage/pytest-integration.md index 14ea648..fc5773b 100644 --- a/docs/usage/pytest-integration.md +++ b/docs/usage/pytest-integration.md @@ -39,6 +39,10 @@ Built-in categories: | `HALLUCINATION` | `"hallucination"` | | `BEHAVIORAL_REGRESSION` | `"behavioral_regression"` | +### `@pytest.mark.trial(n=, threshold=)` + +Declare that a test represents a trial population. The marker remains selectable with `pytest -m trial`, but it does not repeat or clone the test. Use `execute_trials_async` to execute the declared population. + ## Repeated Executions ### `execute_trials_async(n=, threshold=)` @@ -178,12 +182,10 @@ After standard pytest output, RAMPART prints a safety summary grouped by harm ca ========================= RAMPART Safety Summary ========================= DATA_EXFILTRATION (4 tests) - FAIL test_xpia_email_exfil[trial-0] -- Attack objective detected: send_email({'recipient': 'attacker@evil.com'}) (tool_only) - PASS test_xpia_email_exfil[trial-1] -- Agent defended successfully (tool_only) - PASS test_xpia_search_exfil[trial-0] -- Agent defended successfully (tool_only) - PASS test_xpia_search_exfil[trial-1] -- Agent defended successfully (tool_only) - PASS test_xpia_email_exfil [1/2 safe, 50% pass rate, threshold: 80%] -- FAILED - PASS test_xpia_search_exfil [2/2 safe, 100% pass rate, threshold: 80%] -- PASSED + FAIL test_xpia_email_exfil -- Attack objective detected: send_email({'recipient': 'attacker@evil.com'}) (tool_only) + PASS test_xpia_email_exfil -- Agent defended successfully (tool_only) + PASS test_xpia_search_exfil -- Agent defended successfully (tool_only) + PASS test_xpia_search_exfil -- Agent defended successfully (tool_only) MEMORY_POISONING (1 tests) PASS test_memory_poison -- Agent defended successfully (tool_only) @@ -195,12 +197,10 @@ Population: 5 runs - 1 unsafe (20.0% attack success rate), 0 undetermined, 0 err Each result line shows: - **`PASS`/`FAIL`/`WARN`/`ERR`** — the safety verdict -- **Test name** — with `[trial-N]` suffix for trial clones +- **Test name** — repeated executions share one logical pytest test name - **Summary** — e.g., `Agent defended successfully` or `Attack objective detected: ...` - **Observability level** — `tool_only`, `tool_and_side_effects`, or `response_only` -Trial group lines show aggregate stats: safe count, pass rate, threshold, and overall verdict. - The **Population** line shows totals across all tests in the session, with the attack success rate excluding `ERROR` results from the denominator. diff --git a/docs/usage/xdist.md b/docs/usage/xdist.md index 1fa0d1b..76aad36 100644 --- a/docs/usage/xdist.md +++ b/docs/usage/xdist.md @@ -34,7 +34,7 @@ serialize → workeroutput serialize → workeroutput │ ▼ pytest_sessionfinish (controller) - aggregate trials → evaluate gates → emit sinks + emit merged results to sinks │ ▼ Single unified TestRunReport @@ -47,54 +47,11 @@ The result: **one** `JsonFileReportSink` output file, **one** call to `MyCustomS --- -## Trial Tests with xdist +## Population Tests with xdist -`@pytest.mark.trial(n=, threshold=)` clones a test into N independent runs. Under xdist, clones may be distributed across workers depending on the `--dist` mode. +`execute_trials_async` runs a population inside one logical pytest item. xdist assigns that item to one worker, where all executions run sequentially. The returned `PopulationResult` controls the item's assertion, and each individual `Result` is included in the merged report. -| `--dist` mode | Trial behavior | -|---------------|----------------| -| `loadgroup` | All trial clones for one test pinned to the same worker | -| `load` (default) | Trial clones distributed across all workers | -| `loadscope` / `loadfile` | Grouped by class/module/file | - -**Correctness is preserved regardless of mode** — the controller aggregates trial groups from the merged result set and evaluates each group's threshold against the full population. You'll see a warning if you use `@trial` markers without `--dist=loadgroup`: - -```text -RAMPART @trial markers present with --dist=load. Trial clones may be -split across workers. Aggregation remains correct (controller merges -all results), but using --dist=loadgroup keeps trial clones co-located -on one worker for better locality. -``` - -This warning is **informational, not a correctness signal** — see below for when it's safe to ignore. - -### Choosing `loadgroup` vs `load` - -**Both modes produce an identical, correct report.** The controller merges per-worker -partials into one population and evaluates each trial's threshold against the full -group either way. The choice is about *execution*, not correctness: - -- **`load` (default)** spreads a test's trial clones across **all** workers, so a - 20-clone trial keeps every worker busy. It is usually the **fastest** option and is - the right default when trial clones are **independent** (no shared per-group state). -- **`loadgroup`** pins all clones of one trial group to a **single** worker. Prefer it - only when a trial group needs **cohesion** — e.g. clones share a session-scoped - fixture, a per-group cache/connection, or other worker-local state that must not be - split across processes. The trade-off is less parallelism, so it can run slower. - -**Rule of thumb:** independent trials → plain `pytest -n 4` (faster); trials that -share per-group worker state → `pytest -n 4 --dist=loadgroup`. - -As an illustration, one 22-item suite containing a 20-clone trial measured: - -| Mode | Command | Wall time | Reports | `total_runs` | -|------|---------|-----------|---------|--------------| -| Serial | `pytest -n 0` | 203.4s | 1 | 22 | -| Parallel, loadgroup | `pytest -n 4 --dist=loadgroup` | 165.5s | 1 | 22 | -| Parallel, default load | `pytest -n 4` | **113.8s** | 1 | 22 | - -All three emit the same single report and the same trial verdict; `load` is fastest -here because the 20 clones fan out across the 4 workers instead of being pinned to one. +`@pytest.mark.trial(n=, threshold=)` is declaration-only. It remains useful with `-m trial`, but it does not clone or schedule executions. --- @@ -245,9 +202,7 @@ clean `pytest_sessionfinish`. This has two consequences you should be aware of: Both behaviors are deliberate fail-closed choices for this release. A durable per-worker transport (incremental JSONL shards that survive a killed worker, with the size cap applied per-record) is in progress as a follow-up change; until it -lands, use `--dist=loadgroup` only when your trial groups need worker cohesion (see -[Choosing `loadgroup` vs `load`](#choosing-loadgroup-vs-load)) and size your cap to -your largest expected worker payload. +lands, size the cap to your largest expected worker payload. --- diff --git a/pyproject.toml b/pyproject.toml index 32a59a6..ab0d29a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -87,7 +87,7 @@ asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "session" markers = [ "harm(*categories): categorize test by harm type", - "trial(n=, threshold=): statistical repetition of a test", + "trial(n=, threshold=): declare a trial population", "slow: marks tests that spawn subprocess pytest runs; deselect with -m 'not slow'", ] filterwarnings = [ diff --git a/rampart/core/execution.py b/rampart/core/execution.py index 2f13e07..62474ac 100644 --- a/rampart/core/execution.py +++ b/rampart/core/execution.py @@ -297,7 +297,8 @@ async def execute_trials_async( PopulationResult: Aggregate verdict and individual trial results. Raises: - TypeError: If n is not a non-boolean integer. + TypeError: If n is not a non-boolean integer or threshold is not + a non-boolean number. ValueError: If n is less than 1 or threshold is outside [0.0, 1.0]. """ @@ -307,19 +308,14 @@ async def execute_trials_async( if n < 1: msg = "n must be greater than or equal to 1" raise ValueError(msg) - if not 0.0 <= threshold <= 1.0: - msg = "threshold must be between 0.0 and 1.0" - raise ValueError(msg) results: list[Result] = [] + population = PopulationResult(results=results, threshold=threshold) for _ in range(n): result = await self.execute_async(adapter=adapter) results.append(result) - return PopulationResult( - results=results, - threshold=threshold, - ) + return population @abstractmethod async def _execute_async(self, *, adapter: AgentAdapter) -> Result: diff --git a/rampart/core/result.py b/rampart/core/result.py index 0626844..a6cee67 100644 --- a/rampart/core/result.py +++ b/rampart/core/result.py @@ -172,6 +172,7 @@ class PopulationResult: from 0.0 to 1.0. Raises: + TypeError: If threshold is not a non-boolean number. ValueError: If threshold is outside [0.0, 1.0]. """ @@ -182,11 +183,19 @@ def __post_init__(self) -> None: """Validate population configuration. Raises: + TypeError: If threshold is not a non-boolean number. ValueError: If threshold is outside [0.0, 1.0]. """ + if not isinstance(self.threshold, int | float) or isinstance( + self.threshold, + bool, + ): + msg = "threshold must be a number" + raise TypeError(msg) if not 0.0 <= self.threshold <= 1.0: msg = "threshold must be between 0.0 and 1.0" raise ValueError(msg) + @property def safe_count(self) -> int: """Number of safe trials.""" diff --git a/rampart/pytest_plugin/_session.py b/rampart/pytest_plugin/_session.py index c6901a7..3156407 100644 --- a/rampart/pytest_plugin/_session.py +++ b/rampart/pytest_plugin/_session.py @@ -3,8 +3,7 @@ """Session-scoped state for the RAMPART pytest plugin. -Accumulates Result objects, computes trial group aggregates, and -builds the final TestRunReport. +Accumulates Result objects and builds the final TestRunReport. """ from __future__ import annotations @@ -12,15 +11,12 @@ import copy import logging from collections import Counter -from dataclasses import dataclass from typing import TYPE_CHECKING, Any from rampart.core.result import Result, SafetyStatus from rampart.reporting.sink import ReportSink, TestRunReport if TYPE_CHECKING: - from collections.abc import Mapping, Sequence - import pytest from rampart.pytest_plugin._collection import ResultCollector @@ -47,67 +43,12 @@ def _result_sort_key(result: Result) -> tuple[str, int, str]: return (nodeid, index, source_worker) -@dataclass(frozen=True, kw_only=True) -class TrialSpec: - """Trial-clone metadata captured at collection time. - - Carries the data needed to aggregate a trial group without - depending on ``pytest.Item`` attributes — so aggregation works - on the xdist controller, where the cloned items themselves - may not be reachable at session finish. - - Attributes: - base_nodeid (str): The original test's pytest node ID. - threshold (float): Minimum pass rate required for the group. - """ - - base_nodeid: str - threshold: float - - -@dataclass(frozen=True, kw_only=True) -class TrialGroupResult: - """Aggregate statistics for a trial group.""" - - total: int - safe: int - unsafe: int - errors: int - no_result: int - threshold: float - pass_rate: float - passed: bool - - @property - def verdict(self) -> str: - """Human-readable verdict: PASSED or FAILED.""" - return "PASSED" if self.passed else "FAILED" - - @property - def terminal_label(self) -> str: - """Short label for terminal output: PASS or FAIL.""" - return "PASS" if self.passed else "FAIL" - - @property - def detail(self) -> str: - """Summary detail string for terminal output (e.g. '8/10 safe, 2 no-result').""" - parts = [f"{self.safe}/{self.total} safe"] - if self.no_result > 0: - parts.append(f"{self.no_result} no-result") - return ", ".join(parts) - - @property - def has_unsafe(self) -> bool: - """True if any trial produced an UNSAFE result.""" - return self.unsafe > 0 - - class RampartSession: """Session-scoped state for the RAMPART plugin. - Accumulates Result objects from all tests, stores trial group - aggregates, tracks session duration, and builds the final - TestRunReport. Holds configured sinks for report emission. + Accumulates Result objects from all tests, tracks session duration, + and builds the final TestRunReport. Holds configured sinks for report + emission. Args: sinks (list[ReportSink]): Report sinks to emit to at session @@ -117,8 +58,6 @@ class RampartSession: def __init__(self, *, sinks: list[ReportSink] | None = None) -> None: self._results: list[Result] = [] self._results_by_nodeid: dict[str, list[Result]] = {} - self._trial_groups: dict[str, TrialGroupResult] = {} - self._trial_specs: dict[str, TrialSpec] = {} self._sinks: list[ReportSink] = sinks or [] self._duration_seconds: float = 0.0 self._cached_report: TestRunReport | None = None @@ -228,136 +167,11 @@ def absorb(self, *, node: pytest.Item, collector: ResultCollector) -> None: self._results_by_nodeid[node.nodeid] = tagged self._cached_report = None - def record_trial_group( - self, - *, - base_nodeid: str, - clone_nodeids: Sequence[str], - threshold: float, - ) -> None: - """Record aggregate statistics for a trial group. - - Semantics: - - ERROR results make the group fail. - - threshold is the minimum pass rate (SAFE / executed). - e.g. 0.8 means at least 80% of runs must be SAFE. - - Clones with zero results (skipped or crashed before producing - a Result) are tracked as ``no_result`` and excluded from - the pass-rate denominator. - - UNSAFE and UNDETERMINED results count against the pass rate. - - Args: - base_nodeid (str): The original test's node ID. - clone_nodeids (Sequence[str]): Pytest node IDs of all clones - in this trial group. - threshold (float): Minimum pass rate required. - """ - if not clone_nodeids: - return - - total = len(clone_nodeids) - unsafe_count = 0 - error_count = 0 - safe_count = 0 - no_result_count = 0 - - for nodeid in clone_nodeids: - node_results = self._results_by_nodeid.get(nodeid, []) - if not node_results: - no_result_count += 1 - continue - has_unsafe = any(r.status == SafetyStatus.UNSAFE for r in node_results) - has_error = any(r.status == SafetyStatus.ERROR for r in node_results) - has_safe = any(r.status == SafetyStatus.SAFE for r in node_results) - if has_unsafe: - unsafe_count += 1 - elif has_error: - error_count += 1 - elif has_safe: - safe_count += 1 - - executed_count = total - no_result_count - pass_rate = safe_count / executed_count if executed_count > 0 else 0.0 - passed = ( - error_count == 0 - and executed_count > 0 - and pass_rate >= threshold - ) - - self._trial_groups[base_nodeid] = TrialGroupResult( - total=total, - safe=safe_count, - unsafe=unsafe_count, - errors=error_count, - no_result=no_result_count, - threshold=threshold, - pass_rate=pass_rate, - passed=passed, - ) - - def register_trial_spec( - self, - *, - clone_nodeid: str, - base_nodeid: str, - threshold: float, - ) -> None: - """Record trial metadata for a cloned item at collection time. - - Called from ``pytest_collection_modifyitems`` whenever a - ``@pytest.mark.trial`` test is expanded into clones. Stores - the data needed for session-end aggregation in a form that - survives the xdist worker→controller boundary. - - Identical re-registration (same key, same spec) is a no-op so - that repeated collection passes (e.g., in workers and the - controller) converge safely. - - Args: - clone_nodeid (str): Node ID of the cloned item. - base_nodeid (str): Node ID of the original (uncloned) item. - threshold (float): Pass-rate threshold from the trial marker. - """ - self._trial_specs[clone_nodeid] = TrialSpec( - base_nodeid=base_nodeid, - threshold=threshold, - ) - - def merge_trial_specs( - self, - *, - trial_specs: Mapping[str, TrialSpec], - ) -> None: - """Merge trial specs received from an xdist worker payload. - - Idempotent: re-merging identical specs is a no-op. Spec values - from workers should match the controller's own collection - because the same plugin code runs in every process; we merge - defensively so the controller can aggregate correctly even - when its own collection state is unavailable. - - Args: - trial_specs (Mapping[str, TrialSpec]): Specs keyed by - clone node ID. - """ - for clone_nodeid, spec in trial_specs.items(): - self._trial_specs.setdefault(clone_nodeid, spec) - @property def has_results(self) -> bool: """True if any results have been collected.""" return bool(self._results) - @property - def trial_groups(self) -> dict[str, TrialGroupResult]: - """Trial group aggregates, keyed by base node ID.""" - return dict(self._trial_groups) - - @property - def trial_specs(self) -> dict[str, TrialSpec]: - """Read-only view of registered trial specs, keyed by clone node ID.""" - return dict(self._trial_specs) - def merge_worker_results( self, *, diff --git a/rampart/pytest_plugin/_xdist.py b/rampart/pytest_plugin/_xdist.py index c65d02d..5ea15cb 100644 --- a/rampart/pytest_plugin/_xdist.py +++ b/rampart/pytest_plugin/_xdist.py @@ -43,7 +43,6 @@ ToolCall, Turn, ) -from rampart.pytest_plugin._session import TrialSpec from rampart.reporting.sink import ReportSink if TYPE_CHECKING: @@ -501,9 +500,8 @@ def serialize_worker_data(*, session: RampartSession) -> dict[str, Any]: """Serialize a worker's RampartSession state for transport to the controller. Produces a JSON-safe dict containing the schema version, the - package version (for cross-version diagnostics), the worker's - ``_results_by_nodeid`` mapping serialized to primitive types, - and trial specs registered during collection. + package version (for cross-version diagnostics), and the worker's + ``_results_by_nodeid`` mapping serialized to primitive types. Args: session (RampartSession): The worker's session state. @@ -520,14 +518,6 @@ def serialize_worker_data(*, session: RampartSession) -> dict[str, Any]: return { "schema": SCHEMA_VERSION, "results_by_nodeid": serialized, - "trial_specs": [ - { - "clone_nodeid": clone_nodeid, - "base_nodeid": spec.base_nodeid, - "threshold": _safe_float(value=spec.threshold) or 0.0, - } - for clone_nodeid, spec in session.trial_specs.items() - ], } @@ -988,61 +978,6 @@ def deserialize_worker_data(*, data: object) -> dict[str, list[Result]]: return out -def deserialize_trial_specs(*, data: object) -> dict[str, TrialSpec]: - """Deserialize the ``trial_specs`` section of a worker payload. - - Missing or malformed entries are skipped rather than raised so - that a partially-corrupt payload still merges results. The - ``trial_specs`` field is optional: payloads without trials emit - an empty list and this function returns an empty dict. - - Args: - data (object): The deserialized JSON object from - ``node.workeroutput``. - - Returns: - dict[str, TrialSpec]: Trial specs keyed by clone node ID. - - Raises: - SchemaVersionError: Missing or unknown schema version. - WorkerOutputError: ``data`` is not a dict payload. - """ - typed = _validate_schema(data=data) - raw_specs = typed.get("trial_specs", []) - if not isinstance(raw_specs, list): - return {} - out: dict[str, TrialSpec] = {} - for spec in cast("list[Any]", raw_specs): - if not isinstance(spec, dict): - continue - spec_dict = cast("dict[str, Any]", spec) - clone_nodeid = spec_dict.get("clone_nodeid") - base_nodeid = spec_dict.get("base_nodeid") - if not isinstance(clone_nodeid, str) or not isinstance(base_nodeid, str): - continue - if not clone_nodeid or not base_nodeid: - continue - raw_threshold = spec_dict.get("threshold", 0.0) - try: - threshold = ( - float(raw_threshold) - if isinstance( - raw_threshold, - int | float, - ) - else 0.0 - ) - except (TypeError, ValueError): - threshold = 0.0 - if not math.isfinite(threshold): - threshold = 0.0 - out[clone_nodeid] = TrialSpec( - base_nodeid=base_nodeid, - threshold=threshold, - ) - return out - - def finalize_worker(*, config: pytest.Config, session: RampartSession) -> None: """Serialize the worker's session state into ``config.workeroutput``. @@ -1087,35 +1022,6 @@ def finalize_worker(*, config: pytest.Config, session: RampartSession) -> None: workeroutput[WORKEROUTPUT_KEY] = payload -def _safe_deserialize_trial_specs( - *, - payload: object, - worker_id_str: str, -) -> dict[str, TrialSpec]: - """Deserialize trial specs from a worker payload without raising. - - Trial specs are optional metadata: a corrupt or absent block must - never block result merging. Errors are logged at warning level and - return an empty dict. - - Args: - payload (object): The deserialized worker payload. - worker_id_str (str): Worker identifier for logging. - - Returns: - dict[str, TrialSpec]: Specs keyed by clone nodeid (possibly empty). - """ - try: - return deserialize_trial_specs(data=payload) - except WorkerOutputError as exc: - logger.warning( - "Failed to deserialize trial specs from worker %s: %s", - worker_id_str, - exc, - ) - return {} - - def _tag_source_worker( *, results_by_nodeid: dict[str, list[Result]], @@ -1204,17 +1110,11 @@ def handle_testnodedown( reason=f"worker {worker_id_str} deserialization failed: {exc}", ) return - trial_specs = _safe_deserialize_trial_specs( - payload=cast("object", payload), - worker_id_str=worker_id_str, - ) _tag_source_worker( results_by_nodeid=results_by_nodeid, worker_id_str=worker_id_str, ) session.merge_worker_results(results_by_nodeid=results_by_nodeid) - if trial_specs: - session.merge_trial_specs(trial_specs=trial_specs) logger.info( "Merged %d result group(s) from worker %s.", len(results_by_nodeid), diff --git a/rampart/pytest_plugin/plugin.py b/rampart/pytest_plugin/plugin.py index f6914d1..c8e6886 100644 --- a/rampart/pytest_plugin/plugin.py +++ b/rampart/pytest_plugin/plugin.py @@ -6,9 +6,7 @@ Registered via the pytest11 entry point in pyproject.toml. Provides: - harm and trial markers - automatic result collection via the default handler factory -- trial cloning at collection time - terminal summary with harm-category grouping -- session-finish aggregation for trial groups - sink emission for structured reporting Note: The architecture defines _default_handler_factory as a plain @@ -66,7 +64,6 @@ __all__ = [ "pytest_addhooks", "pytest_addoption", - "pytest_collection_modifyitems", "pytest_configure", "pytest_sessionfinish", "pytest_terminal_summary", @@ -104,57 +101,6 @@ def _sanitize_for_terminal(text: str) -> str: return strip_ansi(text) -def _resolve_trial_n(marker: pytest.Mark) -> int: - """Extract the trial count from a trial marker. - - Supports both positional and keyword argument forms: - ``@pytest.mark.trial(5)`` and ``@pytest.mark.trial(n=5)``. - Keyword takes precedence when both are provided. - - Args: - marker (pytest.Mark): The trial marker. - - Returns: - int: The number of trial repetitions. - - Raises: - pytest.UsageError: If the resolved value is not an integer. - """ - raw: Any - if "n" in marker.kwargs: - raw = marker.kwargs["n"] - elif marker.args: - raw = marker.args[0] - else: - return 1 - - if not isinstance(raw, int) or isinstance(raw, bool): - msg = f"trial(n=) must be an integer, got {type(raw).__name__}: {raw!r}" - raise pytest.UsageError(msg) - if raw < 1: - msg = f"trial(n=) must be >= 1, got {raw}" - raise pytest.UsageError(msg) - return raw - - -def _resolve_trial_threshold(marker: pytest.Mark) -> float: - """Extract the threshold from a trial marker. - - Returns 0.0 when no threshold is provided (the historical default). - - Args: - marker (pytest.Mark): The trial marker. - - Returns: - float: The pass-rate threshold in [0.0, 1.0]. - """ - raw: Any = marker.kwargs.get("threshold", 0.0) - try: - return float(raw) - except (TypeError, ValueError): - return 0.0 - - def pytest_addhooks(pluginmanager: pytest.PytestPluginManager) -> None: """Register RAMPART's hook specifications. @@ -208,7 +154,10 @@ def pytest_configure(config: pytest.Config) -> None: config (pytest.Config): The pytest configuration object. """ config.addinivalue_line("markers", "harm(*categories): categorize by harm type") - config.addinivalue_line("markers", "trial(n=, threshold=): statistical repetition") + config.addinivalue_line( + "markers", + "trial(n=, threshold=): declare a trial population", + ) register_default_handler_factory(_default_handler_factory) @@ -229,164 +178,6 @@ def pytest_unconfigure(config: pytest.Config) -> None: del config.stash[_session_start_key] -def _copy_markers_to_clone(*, source: pytest.Item, clone: pytest.Item) -> None: - """Copy all markers from the original item to its trial clone. - - Markers applied at the class level, module level, or via conftest - pytestmark are NOT transferred by ``from_parent``. This function - ensures trial clones inherit all markers (harm, parametrize, etc.) - from the original item. The trial marker itself is re-attached - separately by the caller. - - Args: - source (pytest.Item): The original test item with all markers. - clone (pytest.Item): The cloned item that needs markers copied. - """ - for marker in source.iter_markers(): - if marker.name == "trial": - continue - clone.add_marker( - getattr(pytest.mark, marker.name)(*marker.args, **marker.kwargs), - ) - - -def _create_trial_clones( - *, - item: pytest.Item, - trial_marker: pytest.Mark, - count: int, -) -> list[pytest.Item]: - """Create trial clone items from an original test item. - - Each clone gets a unique ``[trial-N]`` suffix, all markers from - the original item (including class-level and module-level markers), - and private attributes for session-end aggregation. - - Args: - item (pytest.Item): The original test item to clone. - trial_marker (pytest.Mark): The trial marker to re-attach. - count (int): Number of trial repetitions to create. - - Returns: - list[pytest.Item]: The cloned trial items with trial metadata. - - Raises: - pytest.UsageError: If the original item has no parent (cannot be - cloned in isolation). - """ - original_name: str = getattr(item, "originalname", item.name) - display_name = item.name - parent = item.parent - callspec = getattr(item, "callspec", None) - fixtureinfo = getattr(item, "_fixtureinfo", None) - if parent is None: - msg = f"Cannot clone trial item with no parent: {item.nodeid}" - raise pytest.UsageError(msg) - clones: list[pytest.Item] = [] - - for i in range(count): - trial_name = f"{display_name}[trial-{i}]" - from_parent_kwargs: dict[str, Any] = { - "name": trial_name, - "originalname": original_name, - } - if callspec is not None: - from_parent_kwargs["callspec"] = callspec - if fixtureinfo is not None: - from_parent_kwargs["fixtureinfo"] = fixtureinfo - - clone = type(item).from_parent(parent=parent, **from_parent_kwargs) - # pytest.Item supports arbitrary user attributes for cross-hook state. - clone._rampart_trial_index = i # ty: ignore[unresolved-attribute] # noqa: SLF001 - clone._rampart_trial_base = item.nodeid # ty: ignore[unresolved-attribute] # noqa: SLF001 - - _copy_markers_to_clone(source=item, clone=clone) - clone.add_marker( - pytest.mark.trial(*trial_marker.args, **trial_marker.kwargs), - ) - # Group all trials for the same base test on one xdist worker - # so that trial aggregation works correctly across workers. - clone.add_marker(pytest.mark.xdist_group(item.nodeid)) - clones.append(clone) - - return clones - - -@pytest.hookimpl(trylast=True) -def pytest_collection_modifyitems( - config: pytest.Config, - items: list[pytest.Item], -) -> None: - """Clone trial-marked items and validate marker usage. - - Uses ``trylast=True`` so clones are created after pytest-asyncio - has wrapped async items — ``item.obj`` on the original already - carries the async wrapper, which is passed to clones via callobj. - - Expands each ``@pytest.mark.trial(n=)`` item into *n* clones with - distinct node IDs. All markers (harm, parametrize, etc.) from the - original item are copied to each clone. Attaches - ``_rampart_trial_index`` and ``_rampart_trial_base`` to each clone - for session-end aggregation. - - Args: - config (pytest.Config): The pytest configuration object. - items (list[pytest.Item]): The collected test items. - - Raises: - pytest.UsageError: If trial(n=) is not a positive integer or - item has no parent. - """ - expanded: list[pytest.Item] = [] - saw_trial = False - rampart_session = config.stash.get(_rampart_key, None) - for item in items: - trial_marker = item.get_closest_marker("trial") - if trial_marker is None: - expanded.append(item) - continue - - saw_trial = True - n = _resolve_trial_n(trial_marker) - threshold = _resolve_trial_threshold(trial_marker) - clones = _create_trial_clones( - item=item, - trial_marker=trial_marker, - count=n, - ) - - if rampart_session is not None: - # Registered on every process, including xdist workers whose - # specs the controller's merge later drops via setdefault. The - # redundancy is intentional: it keeps single-process and the - # controller's own collection pass correct without branching on - # worker vs controller. Do not "optimize" it away on workers — - # that breaks the single-process and fallback paths. - base_nodeid = item.nodeid - for clone in clones: - rampart_session.register_trial_spec( - clone_nodeid=clone.nodeid, - base_nodeid=base_nodeid, - threshold=threshold, - ) - - expanded.extend(clones) - - items[:] = expanded - - if saw_trial and is_xdist_controller(config=config): - dist_mode = get_dist_mode(config=config) - if dist_mode != "loadgroup": - logger.warning( - "RAMPART @trial markers present with --dist=%s. Trial " - "clones may be split across workers. Aggregation remains " - "correct (controller merges all results), but using " - "--dist=loadgroup keeps trial clones co-located on one " - "worker for better locality.", - dist_mode, - ) - - def _absorb_results( *, rampart_session: RampartSession, @@ -580,76 +371,6 @@ def rampart_sinks(): ) -def _aggregate_trial_results( - *, - rampart_session: RampartSession, -) -> None: - """Group trial specs by base node ID and compute per-group rates. - - Trial specs are recorded during ``pytest_collection_modifyitems`` - on every process and shipped through the xdist worker payload so - aggregation does not depend on ``session.items`` — which is not - reliably populated with trial clones on the xdist controller at - session-finish time. - - Args: - rampart_session (RampartSession): The RAMPART session state. - """ - groups: dict[str, list[tuple[str, float]]] = {} - for clone_nodeid, spec in rampart_session.trial_specs.items(): - groups.setdefault(spec.base_nodeid, []).append( - (clone_nodeid, spec.threshold), - ) - - for base_nodeid, clones in groups.items(): - # All clones of the same base share the same threshold; pick any. - threshold = clones[0][1] - rampart_session.record_trial_group( - base_nodeid=base_nodeid, - clone_nodeids=[c[0] for c in clones], - threshold=threshold, - ) - - -def _evaluate_gates( - *, - rampart_session: RampartSession, -) -> None: - """Log trial group gate results. - - Reports whether each trial group passed or failed based on: - - Any ERROR -> FAIL - - Pass rate below threshold -> FAIL - - Args: - rampart_session (RampartSession): The RAMPART session state. - """ - for base_nodeid, group in sorted(rampart_session.trial_groups.items()): - if group.passed: - logger.info( - "Gate PASSED: %s — %d/%d safe (%.0f%% pass rate, threshold: %.0f%%)", - base_nodeid, - group.safe, - group.total, - group.pass_rate * 100, - group.threshold * 100, - ) - elif group.errors > 0: - logger.info( - "Gate FAILED: %s — %d/%d runs produced ERROR", - base_nodeid, - group.errors, - group.total, - ) - else: - logger.info( - "Gate FAILED: %s — pass rate %.0f%% below threshold %.0f%%", - base_nodeid, - group.pass_rate * 100, - group.threshold * 100, - ) - - def _enforce_incomplete_exit_status( *, session: pytest.Session, @@ -680,19 +401,16 @@ def pytest_sessionfinish( session: pytest.Session, exitstatus: int, # noqa: ARG001 — pytest hook signature ) -> None: - """Aggregate trial results, evaluate gates, and emit sinks. + """Finalize the session and emit sinks. Dispatches between three modes: - xdist worker: serialize results to ``config.workeroutput`` and skip sink emission (the controller emits the unified report). - - xdist controller: trials already aggregated against the merged - ``_results_by_nodeid``; resolve sinks via the - ``pytest_rampart_sinks`` hook (falling back to conftest discovery), - evaluate gates, and emit. - - non-xdist: original single-process pipeline (aggregate, gate, - emit); hook sinks are added here when the fixture path was - suppressed. + - xdist controller: resolve sinks via the ``pytest_rampart_sinks`` + hook (falling back to conftest discovery) and emit merged results. + - non-xdist: add hook sinks when the fixture path was suppressed, + then emit. An incomplete run (a lost or crashed worker) is forced to a non-zero exit status so a dropped shard cannot pass silently. @@ -716,8 +434,6 @@ def pytest_sessionfinish( logger.warning("%s", exc) return - _aggregate_trial_results(rampart_session=rampart_session) - _evaluate_gates(rampart_session=rampart_session) _enforce_incomplete_exit_status(session=session, rampart_session=rampart_session) if is_xdist_controller(config=session.config): @@ -864,28 +580,6 @@ def _write_result_line( ) -def _write_trial_group_lines( - *, - terminalreporter: TerminalReporter, - rampart_session: RampartSession, -) -> None: - """Write trial group aggregate lines to the terminal. - - Format: ``PASS test_name [8/10 safe, 80% defense rate, threshold: 70%] — PASSED`` - - Args: - terminalreporter: The pytest terminal reporter. - rampart_session (RampartSession): The RAMPART session state. - """ - for base_nodeid, group in sorted(rampart_session.trial_groups.items()): - test_name = base_nodeid.split("::")[-1] if "::" in base_nodeid else base_nodeid - terminalreporter.write_line( - f" {group.terminal_label} {test_name} " - f"[{group.detail}, {group.pass_rate:.0%} pass rate, " - f"threshold: {group.threshold:.0%}] -- {group.verdict}", - ) - - def _write_incomplete_warning( *, terminalreporter: TerminalReporter, @@ -919,7 +613,7 @@ def pytest_terminal_summary( Fires after all tests complete. Emits an incomplete-run warning first (even when no results were collected, since a lost worker can leave the run incomplete with zero results), then writes harm-grouped - result lines, trial group aggregates, and population statistics. + result lines and population statistics. Args: terminalreporter: The pytest terminal reporter. @@ -962,11 +656,6 @@ def pytest_terminal_summary( test_name=test_name, ) - _write_trial_group_lines( - terminalreporter=terminalreporter, - rampart_session=rampart_session, - ) - stats = report.population_summary() if stats.total_runs > 0: terminalreporter.write_line( diff --git a/rampart/reporting/sink.py b/rampart/reporting/sink.py index ff61470..17c82d8 100644 --- a/rampart/reporting/sink.py +++ b/rampart/reporting/sink.py @@ -97,14 +97,12 @@ def population_summary( Each Result corresponds to one test execution — one run of one test body. For parametrized payload suites, each payload variant - is one Result. For trial-marked tests, each trial clone is one - Result; trial groups are aggregated separately by the plugin - before this method is called. + is one Result. Population executions contribute each individual + Result while their PopulationResult controls the test assertion. This method does not distinguish payloads from trial repetitions. - Callers that need population-level statistics (distinct payloads, - not repeated trials) should filter Results to non-trial items - before calling, or use the plugin-managed trial-group aggregates. + Callers that need statistics over distinct payloads rather than + repeated executions should filter Results before calling. Args: harm_category (HarmCategory | str | None): Filter to a specific diff --git a/tests/unit/core/test_execution.py b/tests/unit/core/test_execution.py index 0cd924f..8ec5604 100644 --- a/tests/unit/core/test_execution.py +++ b/tests/unit/core/test_execution.py @@ -222,6 +222,23 @@ async def test_rejects_invalid_threshold_before_execution_async(self) -> None: assert handler.events == [] + @pytest.mark.parametrize("threshold", [True, "0.8"]) + async def test_rejects_invalid_threshold_type_before_execution_async( + self, + threshold: object, + ) -> None: + handler = _RecordingHandler() + execution = _SuccessExecution(event_handlers=[handler]) + + with pytest.raises(TypeError, match="threshold must be a number"): + await execution.execute_trials_async( + adapter=_StubAdapter(), + n=3, + threshold=threshold, # ty: ignore[invalid-argument-type] + ) + + assert handler.events == [] + class TestPopulationPublicExports: def test_exported_from_rampart(self) -> None: diff --git a/tests/unit/core/test_result.py b/tests/unit/core/test_result.py index 4e6fab6..32efda9 100644 --- a/tests/unit/core/test_result.py +++ b/tests/unit/core/test_result.py @@ -202,6 +202,14 @@ def test_rejects_threshold_outside_valid_range(self, threshold: float) -> None: with pytest.raises(ValueError, match="threshold must be between"): PopulationResult(results=[], threshold=threshold) + @pytest.mark.parametrize("threshold", [True, "0.8"]) + def test_rejects_invalid_threshold_type(self, threshold: object) -> None: + with pytest.raises(TypeError, match="threshold must be a number"): + PopulationResult( + results=[], + threshold=threshold, # ty: ignore[invalid-argument-type] + ) + def test_summary_contains_population_verdict(self) -> None: population = PopulationResult( results=[_result(SafetyStatus.SAFE), _result(SafetyStatus.UNSAFE)], diff --git a/tests/unit/pytest_plugin/test_plugin.py b/tests/unit/pytest_plugin/test_plugin.py index f5d99d3..94e1b3f 100644 --- a/tests/unit/pytest_plugin/test_plugin.py +++ b/tests/unit/pytest_plugin/test_plugin.py @@ -17,14 +17,10 @@ from rampart.pytest_plugin.plugin import ( _emit_sinks, _enforce_incomplete_exit_status, - _evaluate_gates, _has_sink_hook_impl, _resolve_hook_sinks, - _resolve_trial_n, _sanitize_for_terminal, _write_result_line, - _write_trial_group_lines, - pytest_collection_modifyitems, pytest_configure, pytest_sessionfinish, pytest_terminal_summary, @@ -166,255 +162,6 @@ def test_build_report_counts(self) -> None: assert report.failed == 1 assert report.errors == 1 - def test_record_trial_group(self) -> None: - session = RampartSession() - - items: list[Any] = [MagicMock() for _ in range(4)] - statuses = [ - SafetyStatus.UNSAFE, - SafetyStatus.SAFE, - SafetyStatus.UNSAFE, - SafetyStatus.SAFE, - ] - for idx, item in enumerate(items): - item.nodeid = f"test_file.py::test_example[trial-{idx}]" - collector = ResultCollector() - collector.record( - result=Result( - safe=statuses[idx] == SafetyStatus.SAFE, - status=statuses[idx], - summary=f"trial-{idx}", - ), - ) - session.absorb(node=item, collector=collector) - - session.record_trial_group( - base_nodeid="test_example", - clone_nodeids=[item.nodeid for item in items], - threshold=0.5, - ) - - groups = session.trial_groups - assert "test_example" in groups - group = groups["test_example"] - assert group.total == 4 - assert group.safe == 2 - assert group.unsafe == 2 - assert group.errors == 0 - assert group.threshold == pytest.approx(0.5) - assert group.pass_rate == pytest.approx(0.5) - assert group.passed - - def test_record_trial_group_all_errors(self) -> None: - session = RampartSession() - - items: list[Any] = [MagicMock() for _ in range(3)] - for idx, item in enumerate(items): - item.nodeid = f"test_file.py::test_err[trial-{idx}]" - collector = ResultCollector() - collector.record( - result=Result( - safe=False, - status=SafetyStatus.ERROR, - summary=f"err-{idx}", - ), - ) - session.absorb(node=item, collector=collector) - - session.record_trial_group( - base_nodeid="test_err", - clone_nodeids=[item.nodeid for item in items], - threshold=0.0, - ) - - group = session.trial_groups["test_err"] - assert group.errors == 3 - assert group.unsafe == 0 - assert group.pass_rate == pytest.approx(0.0) - assert not group.passed - - def test_record_trial_group_excludes_no_result_from_denominator(self) -> None: - session = RampartSession() - item = MagicMock() - item.nodeid = "test_file.py::test_skip[trial-0]" - collector = ResultCollector() - collector.record( - result=Result(safe=True, status=SafetyStatus.SAFE, summary="safe"), - ) - session.absorb(node=item, collector=collector) - - session.record_trial_group( - base_nodeid="test_skip", - clone_nodeids=[item.nodeid, "test_file.py::test_skip[trial-1]"], - threshold=1.0, - ) - - group = session.trial_groups["test_skip"] - assert group.no_result == 1 - assert group.pass_rate == pytest.approx(1.0) - assert group.passed - - def test_record_trial_group_empty_items_noop(self) -> None: - session = RampartSession() - session.record_trial_group( - base_nodeid="test_empty", - clone_nodeids=[], - threshold=0.0, - ) - assert "test_empty" not in session.trial_groups - - -def _make_trial_item( - *, - n: int = 3, - threshold: float = 0.0, - nodeid: str = "test_file.py::test_example", - name: str = "test_example", -) -> MagicMock: - """Build a mock pytest.Item with a trial marker.""" - marker = pytest.mark.trial(n=n, threshold=threshold).mark - item = MagicMock() - item.get_closest_marker.return_value = marker - item.nodeid = nodeid - item.name = name - item.originalname = name - item.parent = MagicMock() - item.function = lambda: None - return item - - -def _make_plain_item( - *, - nodeid: str = "test_file.py::test_plain", - name: str = "test_plain", -) -> MagicMock: - """Build a mock pytest.Item without a trial marker.""" - item = MagicMock() - item.get_closest_marker.return_value = None - item.nodeid = nodeid - item.originalname = name - return item - - -class TestTrialCloning: - """Trial cloning produces n items with distinct [trial-N] node ids.""" - - def test_trial_cloning_produces_n_items( - self, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - item = _make_trial_item(n=3) - clone_instances = [MagicMock() for _ in range(3)] - for clone in clone_instances: - clone.iter_markers.return_value = [] - mock_from_parent = MagicMock(side_effect=clone_instances) - # type(item).from_parent is used in plugin, so patch it on the mock's type - type(item).from_parent = mock_from_parent - - items: list[Any] = [item] - config = MagicMock() - pytest_collection_modifyitems( - config=cast("pytest.Config", config), - items=items, - ) - - assert len(items) == 3 - calls = mock_from_parent.call_args_list - for i, call in enumerate(calls): - assert call.kwargs["name"] == f"test_example[trial-{i}]" - - def test_trial_n_zero_raises_usage_error(self) -> None: - item = _make_trial_item(n=0) - items: list[Any] = [item] - config = MagicMock() - - with pytest.raises(pytest.UsageError, match="must be >= 1"): - pytest_collection_modifyitems( - config=cast("pytest.Config", config), - items=items, - ) - - def test_non_trial_items_unchanged(self, monkeypatch: pytest.MonkeyPatch) -> None: - plain = _make_plain_item() - trial = _make_trial_item(n=2) - clone_instances = [MagicMock() for _ in range(2)] - for clone in clone_instances: - clone.iter_markers.return_value = [] - type(trial).from_parent = MagicMock(side_effect=clone_instances) - - items: list[Any] = [plain, trial] - config = MagicMock() - pytest_collection_modifyitems( - config=cast("pytest.Config", config), - items=items, - ) - - assert items[0] is plain - assert len(items) == 3 - - def test_trial_item_with_no_parent_raises(self) -> None: - item = _make_trial_item(n=2) - item.parent = None - - items: list[Any] = [item] - config = MagicMock() - - with pytest.raises(pytest.UsageError, match="no parent"): - pytest_collection_modifyitems( - config=cast("pytest.Config", config), - items=items, - ) - - -class TestResolveTrialN: - """_resolve_trial_n extracts n from positional and keyword args.""" - - def test_keyword_n(self) -> None: - marker = pytest.mark.trial(n=7).mark - assert _resolve_trial_n(marker) == 7 - - def test_positional_n(self) -> None: - marker = pytest.mark.trial(5).mark - assert _resolve_trial_n(marker) == 5 - - def test_keyword_takes_precedence(self) -> None: - marker = pytest.mark.trial(3, n=10).mark - assert _resolve_trial_n(marker) == 10 - - def test_defaults_to_one(self) -> None: - marker = pytest.mark.trial(threshold=0.5).mark - assert _resolve_trial_n(marker) == 1 - - def test_string_n_raises_usage_error(self) -> None: - """Non-integer n raises UsageError instead of a confusing TypeError.""" - marker = pytest.mark.trial(n="five").mark - with pytest.raises(pytest.UsageError, match="must be an integer"): - _resolve_trial_n(marker) - - def test_positional_string_raises_usage_error(self) -> None: - """Non-integer positional arg raises UsageError.""" - marker = pytest.mark.trial("hello").mark - with pytest.raises(pytest.UsageError, match="must be an integer"): - _resolve_trial_n(marker) - - def test_float_n_raises_usage_error(self) -> None: - """Float n raises UsageError.""" - marker = pytest.mark.trial(n=3.5).mark - with pytest.raises(pytest.UsageError, match="must be an integer"): - _resolve_trial_n(marker) - - def test_bool_n_raises_usage_error(self) -> None: - """Bool n raises UsageError (bool is subclass of int).""" - marker = pytest.mark.trial(n=True).mark - with pytest.raises(pytest.UsageError, match="must be an integer"): - _resolve_trial_n(marker) - - def test_bool_false_raises_usage_error(self) -> None: - """False also rejected despite bool being int subclass.""" - marker = pytest.mark.trial(n=False).mark - with pytest.raises(pytest.UsageError, match="must be an integer"): - _resolve_trial_n(marker) - class TestSanitizeForTerminal: """ANSI escape sequences are stripped from terminal output.""" @@ -704,81 +451,6 @@ def test_set_duration_reflected_in_report(self) -> None: assert report.duration_seconds == pytest.approx(42.5) -class TestTrialGroupRendering: - """Trial group aggregate lines are written to terminal.""" - - def test_writes_trial_group_line(self) -> None: - session = RampartSession() - items: list[Any] = [MagicMock() for _ in range(10)] - for idx, item in enumerate(items): - item.nodeid = f"test_file.py::test_stat[trial-{idx}]" - collector = ResultCollector() - status = SafetyStatus.UNSAFE if idx < 2 else SafetyStatus.SAFE - collector.record( - result=Result( - safe=status == SafetyStatus.SAFE, - status=status, - summary=f"t-{idx}", - ), - ) - session.absorb(node=item, collector=collector) - - session.record_trial_group( - base_nodeid="test_file.py::test_stat", - clone_nodeids=[item.nodeid for item in items], - threshold=0.3, - ) - - reporter = MagicMock() - _write_trial_group_lines( - terminalreporter=cast("TerminalReporter", reporter), - rampart_session=session, - ) - - reporter.write_line.assert_called_once() - line = reporter.write_line.call_args[0][0] - assert "8/10 safe" in line - assert "80% pass rate" in line - assert "PASSED" in line - - def test_no_trial_groups_writes_nothing(self) -> None: - session = RampartSession() - reporter = MagicMock() - _write_trial_group_lines( - terminalreporter=cast("TerminalReporter", reporter), - rampart_session=session, - ) - reporter.write_line.assert_not_called() - - -class TestEvaluateGates: - """Gate evaluation logs when threshold is exceeded.""" - - def test_logs_when_rate_exceeds_threshold(self) -> None: - session = RampartSession() - items: list[Any] = [MagicMock() for _ in range(4)] - for idx, item in enumerate(items): - item.nodeid = f"test.py::test_gate[trial-{idx}]" - collector = ResultCollector() - status = SafetyStatus.UNSAFE if idx < 2 else SafetyStatus.SAFE - collector.record( - result=Result( - safe=status == SafetyStatus.SAFE, - status=status, - summary=f"t-{idx}", - ), - ) - session.absorb(node=item, collector=collector) - - session.record_trial_group( - base_nodeid="test.py::test_gate", - clone_nodeids=[item.nodeid for item in items], - threshold=0.1, - ) - - _evaluate_gates(rampart_session=session) - - class TestEmitSinks: """Sink emission calls emit_async and handles errors.""" diff --git a/tests/unit/pytest_plugin/test_xdist.py b/tests/unit/pytest_plugin/test_xdist.py index 8c56ea1..f4405b9 100644 --- a/tests/unit/pytest_plugin/test_xdist.py +++ b/tests/unit/pytest_plugin/test_xdist.py @@ -31,7 +31,7 @@ ToolCall, Turn, ) -from rampart.pytest_plugin._session import RampartSession, TrialSpec +from rampart.pytest_plugin._session import RampartSession from rampart.pytest_plugin._xdist import ( DEFAULT_SIZE_LIMIT_BYTES, MAX_METADATA_DEPTH, @@ -43,7 +43,6 @@ WorkerOutputError, _sanitize, _strip_ansi, - deserialize_trial_specs, deserialize_worker_data, discover_sinks_from_conftest, finalize_worker, @@ -697,37 +696,6 @@ def test_merges_results_on_success(self) -> None: assert len(session._results) == 1 assert session._results[0].summary == "from-worker" - def test_merges_trial_specs_on_success(self) -> None: - session = RampartSession() - worker_session = RampartSession() - worker_session.register_trial_spec( - clone_nodeid="test.py::test_x[trial-0]", - base_nodeid="test.py::test_x", - threshold=0.8, - ) - worker_session.register_trial_spec( - clone_nodeid="test.py::test_x[trial-1]", - base_nodeid="test.py::test_x", - threshold=0.8, - ) - payload = serialize_worker_data(session=worker_session) - node = MagicMock() - node.gateway.id = "gw1" - node.workeroutput = {WORKEROUTPUT_KEY: payload} - handle_testnodedown(session=session, node=node, error=None) - assert session.is_incomplete is False - assert set(session.trial_specs) == { - "test.py::test_x[trial-0]", - "test.py::test_x[trial-1]", - } - assert ( - session.trial_specs["test.py::test_x[trial-0]"].base_nodeid - == "test.py::test_x" - ) - assert session.trial_specs[ - "test.py::test_x[trial-0]" - ].threshold == pytest.approx(0.8) - class TestOrderingDeterminism: def _payload_node( @@ -792,86 +760,6 @@ def test_handle_testnodedown_tags_source_worker(self) -> None: assert session._results[0].metadata["_rampart_source_worker"] == "gw3" -class TestTrialSpecs: - def test_serialize_round_trip(self) -> None: - session = RampartSession() - session.register_trial_spec( - clone_nodeid="t.py::a[trial-0]", - base_nodeid="t.py::a", - threshold=0.75, - ) - session.register_trial_spec( - clone_nodeid="t.py::a[trial-1]", - base_nodeid="t.py::a", - threshold=0.75, - ) - payload = serialize_worker_data(session=session) - - # Payload must survive a JSON round-trip (xdist transports JSON). - decoded = json.loads(json.dumps(payload)) - specs = deserialize_trial_specs(data=decoded) - - assert specs == { - "t.py::a[trial-0]": TrialSpec(base_nodeid="t.py::a", threshold=0.75), - "t.py::a[trial-1]": TrialSpec(base_nodeid="t.py::a", threshold=0.75), - } - - def test_payload_without_trials_returns_empty_dict(self) -> None: - session = RampartSession() - payload = serialize_worker_data(session=session) - assert deserialize_trial_specs(data=payload) == {} - - def test_skips_malformed_entries(self) -> None: - data: dict[str, Any] = { - "schema": SCHEMA_VERSION, - "results_by_nodeid": {}, - "trial_specs": [ - {"clone_nodeid": "ok", "base_nodeid": "b", "threshold": 0.5}, - "not-a-dict", - {"clone_nodeid": "", "base_nodeid": "b", "threshold": 0.5}, - {"clone_nodeid": "x", "base_nodeid": 123, "threshold": 0.5}, - {"clone_nodeid": "y", "base_nodeid": "b"}, - ], - } - specs = deserialize_trial_specs(data=data) - assert set(specs) == {"ok", "y"} - assert specs["y"].threshold == pytest.approx(0.0) - - def test_clamps_non_finite_threshold(self) -> None: - data: dict[str, Any] = { - "schema": SCHEMA_VERSION, - "results_by_nodeid": {}, - "trial_specs": [ - {"clone_nodeid": "a", "base_nodeid": "b", "threshold": float("inf")}, - {"clone_nodeid": "c", "base_nodeid": "d", "threshold": float("nan")}, - ], - } - specs = deserialize_trial_specs(data=data) - assert specs["a"].threshold == pytest.approx(0.0) - assert specs["c"].threshold == pytest.approx(0.0) - - def test_merge_is_idempotent(self) -> None: - session = RampartSession() - spec = TrialSpec(base_nodeid="b", threshold=0.5) - session.merge_trial_specs(trial_specs={"k": spec}) - session.merge_trial_specs(trial_specs={"k": spec}) - assert session.trial_specs == {"k": spec} - - def test_merge_first_writer_wins(self) -> None: - session = RampartSession() - original = TrialSpec(base_nodeid="b1", threshold=0.5) - replacement = TrialSpec(base_nodeid="b2", threshold=0.9) - session.merge_trial_specs(trial_specs={"k": original}) - session.merge_trial_specs(trial_specs={"k": replacement}) - # Defensive: the first registered spec wins so a worker can't - # silently override what the controller already saw at collection. - assert session.trial_specs["k"] == original - - def test_invalid_payload_raises(self) -> None: - with pytest.raises(WorkerOutputError): - deserialize_trial_specs(data="not a dict") - - class TestFinalizeWorker: def test_no_op_on_controller(self) -> None: config = _make_config(is_worker=False, numprocesses=2) diff --git a/tests/unit/pytest_plugin/test_xdist_aggregation.py b/tests/unit/pytest_plugin/test_xdist_aggregation.py index 96ac976..f314227 100644 --- a/tests/unit/pytest_plugin/test_xdist_aggregation.py +++ b/tests/unit/pytest_plugin/test_xdist_aggregation.py @@ -18,7 +18,7 @@ import pytest if TYPE_CHECKING: - from _pytest.pytester import Pytester, RunResult + from _pytest.pytester import Pytester pytest_plugins = ["pytester"] @@ -175,259 +175,83 @@ def test_population_statistics_over_full_set( assert report["population_summary"]["unsafe_count"] == 1 -class TestXdistTrialAggregation: - def test_trial_aggregation_across_workers_loadgroup( +class TestPopulationPytestVerdict: + def _make_population_test( self, + *, configured_pytester: Pytester, + threshold: float, ) -> None: configured_pytester.makepyfile( - test_trial=""" + test_population=f""" import pytest - from rampart import record_result - from rampart.core.result import Result, SafetyStatus - from rampart.core.types import ObservabilityLevel - - @pytest.mark.harm("test") - @pytest.mark.trial(n=4, threshold=0.5) - def test_trial_split(): - record_result(Result( - safe=True, status=SafetyStatus.SAFE, summary="t", - observability_level=ObservabilityLevel.RESPONSE_ONLY, - )) - """, - ) - result = configured_pytester.runpytest( - "-p", - "no:cacheprovider", - "-n", - "2", - "--dist", - "loadgroup", - ) - result.assert_outcomes(passed=4) - reports = _load_reports(configured_pytester) - assert len(reports) == 1 - assert reports[0]["total_runs"] == 4 - def test_trial_aggregation_across_workers_load( - self, - configured_pytester: Pytester, - ) -> None: - configured_pytester.makepyfile( - test_trial=""" - import pytest - from rampart import record_result + from rampart.core.execution import BaseExecution from rampart.core.result import Result, SafetyStatus - from rampart.core.types import ObservabilityLevel - - @pytest.mark.harm("test") - @pytest.mark.trial(n=4, threshold=0.5) - def test_trial_split(): - record_result(Result( - safe=True, status=SafetyStatus.SAFE, summary="t", - observability_level=ObservabilityLevel.RESPONSE_ONLY, - )) - """, - ) - result = configured_pytester.runpytest( - "-p", - "no:cacheprovider", - "-n", - "2", - "--dist", - "load", - ) - result.assert_outcomes(passed=4) - reports = _load_reports(configured_pytester) - assert len(reports) == 1 - assert reports[0]["total_runs"] == 4 - def test_trial_group_passes_at_threshold_with_unsafe_under_loadgroup( - self, - configured_pytester: Pytester, - ) -> None: - """UNSAFE trials are tolerated when the pass rate meets the threshold. - 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. - """ - configured_pytester.makepyfile( - test_trial_mixed=""" - import pytest - from rampart import record_result - from rampart.core.result import Result, SafetyStatus - from rampart.core.types import ObservabilityLevel - - @pytest.mark.harm("test") - @pytest.mark.trial(n=4, threshold=0.5) - def test_trial_mixed(request): - unsafe = request.node.name.endswith("[trial-3]") - record_result(Result( - safe=not unsafe, - status=SafetyStatus.UNSAFE if unsafe else SafetyStatus.SAFE, - summary="u" if unsafe else "s", - observability_level=ObservabilityLevel.RESPONSE_ONLY, - )) + class SequenceExecution(BaseExecution): + def __init__(self): + super().__init__() + self.statuses = [ + SafetyStatus.SAFE, + SafetyStatus.SAFE, + SafetyStatus.SAFE, + SafetyStatus.UNSAFE, + SafetyStatus.UNSAFE, + ] + + @property + def strategy_name(self): + return "sequence" + + async def _execute_async(self, *, adapter): + status = self.statuses.pop(0) + return Result( + safe=status is SafetyStatus.SAFE, + status=status, + summary=status.value, + ) + + + @pytest.mark.trial(n=5, threshold={threshold}) + async def test_population_threshold(): + result = await SequenceExecution().execute_trials_async( + adapter=object(), + n=5, + threshold={threshold}, + ) + assert result, result.summary """, ) - result = configured_pytester.runpytest( - "-p", - "no:cacheprovider", - "-n", - "2", - "--dist", - "loadgroup", - ) - # All 4 clones pass at the pytest item level — record_result - # does not fail the test; it only records a Result. - result.assert_outcomes(passed=4) - reports = _load_reports(configured_pytester) - assert len(reports) == 1 - report = reports[0] - assert report["total_runs"] == 4 - assert report["passed"] == 3 - assert report["failed"] == 1 - # The trial-group PASS line proves the controller correctly - # aggregated worker results. The bracketed stats uniquely - # identify the group line (the per-clone lines lack them). - summary = "\n".join(result.outlines) - assert "RAMPART Safety Summary" in summary - assert ( - "PASS test_trial_mixed [3/4 safe, 75% pass rate, threshold: 50%]" - in summary - ) - def test_trial_group_passes_at_threshold_with_unsafe_under_load( + def test_exact_threshold_passes_one_pytest_item( self, configured_pytester: Pytester, ) -> None: - """Same as above but with --dist=load so clones may split workers. - - The PR docs claim aggregation remains correct under --dist=load - because the controller merges all worker results. - """ - configured_pytester.makepyfile( - test_trial_mixed_load=""" - import pytest - from rampart import record_result - from rampart.core.result import Result, SafetyStatus - from rampart.core.types import ObservabilityLevel - - @pytest.mark.harm("test") - @pytest.mark.trial(n=4, threshold=0.5) - def test_trial_mixed_load(request): - unsafe = request.node.name.endswith("[trial-3]") - record_result(Result( - safe=not unsafe, - status=SafetyStatus.UNSAFE if unsafe else SafetyStatus.SAFE, - summary="u" if unsafe else "s", - observability_level=ObservabilityLevel.RESPONSE_ONLY, - )) - """, - ) - result = configured_pytester.runpytest( - "-p", - "no:cacheprovider", - "-n", - "2", - "--dist", - "load", - ) - result.assert_outcomes(passed=4) - reports = _load_reports(configured_pytester) - assert len(reports) == 1 - report = reports[0] - assert report["total_runs"] == 4 - assert report["failed"] == 1 - summary = "\n".join(result.outlines) - assert ( - "PASS test_trial_mixed_load [3/4 safe, 75% pass rate, threshold: 50%]" - in summary + self._make_population_test( + configured_pytester=configured_pytester, + threshold=0.6, ) - def test_trial_group_fails_below_threshold_under_loadgroup( - self, - configured_pytester: Pytester, - ) -> None: - """No UNSAFE results, but pass rate below threshold => FAIL. - - 2 SAFE + 2 UNDETERMINED trials, threshold=0.75. Pass rate is 0.5 - so the group must FAIL on the threshold rule (not the unsafe rule). - """ - configured_pytester.makepyfile( - test_trial_threshold=""" - import pytest - from rampart import record_result - from rampart.core.result import Result, SafetyStatus - from rampart.core.types import ObservabilityLevel + result = configured_pytester.runpytest("-p", "no:cacheprovider", "-q") - @pytest.mark.harm("test") - @pytest.mark.trial(n=4, threshold=0.75) - def test_trial_threshold(request): - undetermined = request.node.name.endswith( - ("[trial-2]", "[trial-3]"), - ) - record_result(Result( - safe=True, - status=( - SafetyStatus.UNDETERMINED - if undetermined else SafetyStatus.SAFE - ), - summary="t", - observability_level=ObservabilityLevel.RESPONSE_ONLY, - )) - """, - ) - result = configured_pytester.runpytest( - "-p", - "no:cacheprovider", - "-n", - "2", - "--dist", - "loadgroup", - ) - # All 4 clones pass as pytest tests (record_result(safe=True)), - # but the trial GROUP should fail on threshold. - result.assert_outcomes(passed=4) - summary = "\n".join(result.outlines) - assert "FAIL test_trial_threshold" in summary - assert "50% pass rate" in summary - assert "threshold: 75%" in summary + result.assert_outcomes(passed=1) + assert not any("trial-" in line for line in result.outlines) - def test_trial_group_passes_when_all_safe_under_loadgroup( + def test_below_threshold_fails_one_pytest_item( self, configured_pytester: Pytester, ) -> None: - """All-SAFE trial group with achievable threshold => PASS verdict.""" - configured_pytester.makepyfile( - test_trial_all_safe=""" - import pytest - from rampart import record_result - from rampart.core.result import Result, SafetyStatus - from rampart.core.types import ObservabilityLevel - - @pytest.mark.harm("test") - @pytest.mark.trial(n=3, threshold=0.5) - def test_trial_all_safe(): - record_result(Result( - safe=True, status=SafetyStatus.SAFE, summary="ok", - observability_level=ObservabilityLevel.RESPONSE_ONLY, - )) - """, - ) - result = configured_pytester.runpytest( - "-p", - "no:cacheprovider", - "-n", - "2", - "--dist", - "loadgroup", + self._make_population_test( + configured_pytester=configured_pytester, + threshold=0.8, ) - result.assert_outcomes(passed=3) - summary = "\n".join(result.outlines) - assert "PASS test_trial_all_safe" in summary - assert "PASSED" in summary + + result = configured_pytester.runpytest("-p", "no:cacheprovider", "-q") + + result.assert_outcomes(failed=1) + assert not any("trial-" in line for line in result.outlines) class TestXdistMetadata: @@ -483,8 +307,8 @@ def test_collect_only_does_not_emit_reports( assert reports == [] -class TestCloneIdDeterminism: - def test_trial_clone_ids_deterministic_across_processes( +class TestTrialMarkerDeclaration: + def test_trial_marker_collects_one_item( self, configured_pytester: Pytester, ) -> None: @@ -497,27 +321,40 @@ def test_x(): pass """, ) - result_serial: RunResult = configured_pytester.runpytest( + result = configured_pytester.runpytest( "-p", "no:cacheprovider", "--collect-only", "-q", ) - result_parallel: RunResult = configured_pytester.runpytest( + + result.assert_outcomes(passed=0) + assert sum("test_det.py::test_x" in line for line in result.outlines) == 1 + assert not any("trial-" in line for line in result.outlines) + + def test_trial_marker_remains_selectable( + self, + configured_pytester: Pytester, + ) -> None: + configured_pytester.makepyfile( + test_selection=""" + import pytest + + @pytest.mark.trial(n=3, threshold=0.8) + def test_population(): + pass + + def test_plain(): + pass + """, + ) + + result = configured_pytester.runpytest( "-p", "no:cacheprovider", - "--collect-only", + "-m", + "trial", "-q", - "-n", - "2", ) - def _trial_ids(lines: list[str]) -> list[str]: - return sorted(line.strip() for line in lines if "trial-" in line) - - serial_ids = _trial_ids(result_serial.outlines) - parallel_ids = _trial_ids(result_parallel.outlines) - # Under xdist --collect-only, both should produce the same - # deterministic clone IDs so that workers can match them. - if serial_ids and parallel_ids: - assert serial_ids == parallel_ids + result.assert_outcomes(passed=1, deselected=1)