API Reference — pytest Plugin¶

RAMPART's pytest integration. Activates automatically when installed.

_collection ¶

Result collection infrastructure for the pytest plugin.

Provides the ContextVar-based mechanism for collecting Result objects produced during test execution. The pytest plugin activates a collector per test; execution event handlers write into it automatically.

ResultCollectionHandler ¶

Bases: ExecutionEventHandler

Default ExecutionEventHandler installed on every BaseExecution.

Writes the Result into the active per-test collector on ON_POST_EXECUTE. No-op for all other events. No-op when no collector is active (safe to use outside pytest).

on_event `async` ¶

Python

on_event(*, event_data)

Record result on post-execute. Ignore all other events.

Parameters:

Name	Type	Description	Default
`event_data`	`ExecutionEventData`	The event data.	required

Source code in rampart/pytest_plugin/_collection.py

Python
async def on_event(self, *, event_data: ExecutionEventData) -> None:
    """Record result on post-execute. Ignore all other events.

    Args:
        event_data (ExecutionEventData): The event data.
    """
    if event_data.event is not ExecutionEvent.ON_POST_EXECUTE:
        return
    if event_data.result is None:
        return
    collector = _active_collector.get()
    if collector is not None:
        collector.record(result=event_data.result)

ResultCollector ¶

Python

ResultCollector()

Accumulates Result objects produced during a single test.

Framework-internal. Never referenced by test authors.

Source code in rampart/pytest_plugin/_collection.py

Python
def __init__(self) -> None:
    self._results: list[Result] = []

results `property` ¶

Python

results

All results recorded so far.

record ¶

Python

record(*, result)

Record a result.

Parameters:

Name	Type	Description	Default
`result`	`Result`	The result to record.	required

Source code in rampart/pytest_plugin/_collection.py

Python
def record(self, *, result: Result) -> None:
    """Record a result.

    Args:
        result (Result): The result to record.
    """
    self._results.append(result)

record_result ¶

Python

record_result(result)

Record a Result into the active test's collector.

For building-block tests that construct Results manually rather than via Attacks. or Probes. factories. No-op when called outside a pytest test context (e.g., in library usage or scripts).

Re-exported from rampart at the top level — consumers import it as:

Text Only

from rampart import record_result

Parameters:

Name	Type	Description	Default
`result`	`Result`	The result to record.	required

Source code in rampart/pytest_plugin/_collection.py

Python
def record_result(result: Result) -> None:
    """Record a Result into the active test's collector.

    For building-block tests that construct Results manually rather
    than via Attacks.* or Probes.* factories. No-op when called
    outside a pytest test context (e.g., in library usage or scripts).

    Re-exported from rampart at the top level — consumers import it as:

        from rampart import record_result

    Args:
        result (Result): The result to record.
    """
    collector = _active_collector.get()
    if collector is not None:
        collector.record(result=result)

_session ¶

Session-scoped state for the RAMPART pytest plugin.

Accumulates Result objects, computes trial group aggregates, and builds the final TestRunReport.

Note: The architecture places RampartSession in plugin.py. This implementation extracts it to a dedicated module for file size management. This is a documented deviation from the architecture.

RampartSession ¶

Python

RampartSession(*, sinks=None)

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.

Parameters:

Name	Type	Description	Default
`sinks`	`list[ReportSink]`	Report sinks to emit to at session end. Defaults to an empty list (terminal-only output).	`None`

Source code in rampart/pytest_plugin/_session.py

Python
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._sinks: list[ReportSink] = sinks or []
    self._duration_seconds: float = 0.0
    self._cached_report: TestRunReport | None = None

sinks `property` ¶

Python

sinks

Configured report sinks.

has_results `property` ¶

Python

has_results

True if any results have been collected.

trial_groups `property` ¶

Python

trial_groups

Trial group aggregates, keyed by base node ID.

add_sinks ¶

Python

add_sinks(*, sinks)

Register additional sinks for report emission.

Called by the fixture-based bootstrap to add team-provided sinks.

Parameters:

Name	Type	Description	Default
`sinks`	`list[ReportSink]`	Sinks to append.	required

Raises:

Type	Description
`TypeError`	If any item does not satisfy ReportSink.

Source code in rampart/pytest_plugin/_session.py

Python
def add_sinks(self, *, sinks: list[ReportSink]) -> None:
    """Register additional sinks for report emission.

    Called by the fixture-based bootstrap to add team-provided
    sinks.

    Args:
        sinks (list[ReportSink]): Sinks to append.

    Raises:
        TypeError: If any item does not satisfy ReportSink.
    """
    for sink in sinks:
        if not isinstance(sink, ReportSink):  # pyright: ignore[reportUnnecessaryIsInstance]
            msg = (
                f"Expected ReportSink, got {type(sink).__name__}. "
                "Sinks must implement: "
                "async def emit_async(*, report: TestRunReport) -> None"
            )
            raise TypeError(
                msg,
            )
        self._sinks.append(sink)

set_duration ¶

Python

set_duration(*, duration_seconds)

Set the total session duration.

Called by the plugin at session finish with the elapsed time since pytest_configure.

Parameters:

Name	Type	Description	Default
`duration_seconds`	`float`	Total wall-clock seconds.	required

Source code in rampart/pytest_plugin/_session.py

Python
def set_duration(self, *, duration_seconds: float) -> None:
    """Set the total session duration.

    Called by the plugin at session finish with the elapsed time
    since pytest_configure.

    Args:
        duration_seconds (float): Total wall-clock seconds.
    """
    self._duration_seconds = duration_seconds

absorb ¶

Python

absorb(*, node, collector)

Absorb results from a completed test's collector.

Tags each result with the short test name (extracted from the node ID) and the harm category from @pytest.mark.harm so the terminal summary can group and display results.

Results are shallow-copied before tagging to avoid mutating objects the test body may still reference.

Parameters:

Name	Type	Description	Default
`node`	`Item`	The test item that just completed.	required
`collector`	`ResultCollector`	The test's result collector.	required

Source code in rampart/pytest_plugin/_session.py

Python
def absorb(self, *, node: pytest.Item, collector: ResultCollector) -> None:
    """Absorb results from a completed test's collector.

    Tags each result with the short test name (extracted from
    the node ID) and the harm category from ``@pytest.mark.harm``
    so the terminal summary can group and display results.

    Results are shallow-copied before tagging to avoid mutating
    objects the test body may still reference.

    Args:
        node (pytest.Item): The test item that just completed.
        collector (ResultCollector): The test's result collector.
    """
    test_name = node.nodeid.split("::")[-1] if "::" in node.nodeid else node.nodeid
    harm_marker = node.get_closest_marker("harm")
    harm_category = (
        harm_marker.args[0] if harm_marker and harm_marker.args else None
    )

    collected = collector.results
    tagged: list[Result] = []
    for original_result in collected:
        # Shallow copy is sufficient because we reconstruct all
        # mutable fields we modify (currently metadata and harm_category).
        result = copy.copy(original_result)
        result.metadata = {**result.metadata, "test_name": test_name}
        if harm_category is not None and result.harm_category is None:
            result.harm_category = harm_category
        tagged.append(result)
    self._results.extend(tagged)
    self._results_by_nodeid[node.nodeid] = tagged
    self._cached_report = None

record_trial_group ¶

Python

record_trial_group(*, base_nodeid, trial_items, threshold)

Record aggregate statistics for a trial group.

Semantics: - Any UNSAFE result across all trials -> group FAILS - threshold is the minimum pass rate (SAFE / total). e.g. 0.8 means at least 80% of runs must be SAFE. - ERROR results count against the pass rate (they're not SAFE). - Items with zero results (skipped or crashed before producing a Result) are tracked as no_result and count against the pass rate.

Parameters:

Name	Type	Description	Default
`base_nodeid`	`str`	The original test's node ID.	required
`trial_items`	`Sequence[Item]`	All trial clone items.	required
`threshold`	`float`	Minimum pass rate required.	required

Source code in rampart/pytest_plugin/_session.py

Python
def record_trial_group(
    self,
    *,
    base_nodeid: str,
    trial_items: Sequence[pytest.Item],
    threshold: float,
) -> None:
    """Record aggregate statistics for a trial group.

    Semantics:
    - Any UNSAFE result across all trials -> group FAILS
    - threshold is the minimum pass rate (SAFE / total).
      e.g. 0.8 means at least 80% of runs must be SAFE.
    - ERROR results count against the pass rate (they're not SAFE).
    - Items with zero results (skipped or crashed before producing
      a Result) are tracked as ``no_result`` and count against
      the pass rate.

    Args:
        base_nodeid (str): The original test's node ID.
        trial_items (Sequence[pytest.Item]): All trial clone items.
        threshold (float): Minimum pass rate required.
    """
    if not trial_items:
        return

    total = len(trial_items)
    unsafe_count = 0
    error_count = 0
    safe_count = 0
    no_result_count = 0

    for item in trial_items:
        node_results = self._results_by_nodeid.get(item.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

    pass_rate = safe_count / total if total > 0 else 0.0
    passed = unsafe_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,
    )

build_report ¶

Python

build_report()

Build a TestRunReport from all collected results.

The report is cached and reused on subsequent calls. The cache is invalidated when new results are absorbed.

Returns:

Name	Type	Description
`TestRunReport`	`TestRunReport`	Aggregated test run results.

Source code in rampart/pytest_plugin/_session.py

Python
def build_report(self) -> TestRunReport:
    """Build a TestRunReport from all collected results.

    The report is cached and reused on subsequent calls. The
    cache is invalidated when new results are absorbed.

    Returns:
        TestRunReport: Aggregated test run results.
    """
    if self._cached_report is not None:
        return self._cached_report
    counts = Counter(r.status for r in self._results)
    self._cached_report = TestRunReport(
        results=list(self._results),
        total_runs=len(self._results),
        passed=counts[SafetyStatus.SAFE],
        failed=counts[SafetyStatus.UNSAFE],
        undetermined=counts[SafetyStatus.UNDETERMINED],
        errors=counts[SafetyStatus.ERROR],
        duration_seconds=self._duration_seconds,
    )
    return self._cached_report

TrialGroupResult `dataclass` ¶

Python

TrialGroupResult(
    *,
    total,
    safe,
    unsafe,
    errors,
    no_result,
    threshold,
    pass_rate,
    passed,
)

Aggregate statistics for a trial group.

verdict `property` ¶

Python

verdict

Human-readable verdict: PASSED or FAILED.

terminal_label `property` ¶

Python

terminal_label

Short label for terminal output: PASS or FAIL.

detail `property` ¶

Python

detail

Summary detail string for terminal output (e.g. '8/10 safe, 2 no-result').

has_unsafe `property` ¶

Python

has_unsafe

True if any trial produced an UNSAFE result.

API Reference — pytest Plugin¶

_collection ¶

ResultCollectionHandler ¶

on_event async ¶

ResultCollector ¶

results property ¶

record ¶

record_result ¶

_session ¶

RampartSession ¶

sinks property ¶

has_results property ¶

trial_groups property ¶

add_sinks ¶

set_duration ¶

absorb ¶

record_trial_group ¶

build_report ¶

TrialGroupResult dataclass ¶

verdict property ¶

terminal_label property ¶

detail property ¶

has_unsafe property ¶

on_event `async` ¶

results `property` ¶

sinks `property` ¶

has_results `property` ¶

trial_groups `property` ¶

TrialGroupResult `dataclass` ¶

verdict `property` ¶

terminal_label `property` ¶

detail `property` ¶

has_unsafe `property` ¶