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 `async` ¶

Python

on_event_async(*, 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
@override
async def on_event_async(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.

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._trial_specs: dict[str, TrialSpec] = {}
    self._sinks: list[ReportSink] = sinks or []
    self._duration_seconds: float = 0.0
    self._cached_report: TestRunReport | None = None
    self._emitted: bool = False
    self._incomplete: bool = False
    self._incomplete_reasons: list[str] = []
    self._report_metadata: dict[str, object] = {}

sinks `property` ¶

Python

sinks

Configured report sinks.

results_by_nodeid `property` ¶

Python

results_by_nodeid

Read-only view of results grouped by pytest node ID.

is_emitted `property` ¶

Python

is_emitted

True once report emission has been attempted (idempotency guard).

is_incomplete `property` ¶

Python

is_incomplete

True if any worker failed to deliver complete results.

incomplete_reasons `property` ¶

Python

incomplete_reasons

The recorded reasons the run is incomplete (empty if complete).

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.

trial_specs `property` ¶

Python

trial_specs

Read-only view of registered trial specs, keyed by clone 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):
            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), the full node ID, its index within the test, and the harm category from @pytest.mark.harm. The nodeid and index give a total, deterministic ordering for the terminal summary and report regardless of xdist worker completion order.

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), the full node ID, its index within the test, and the
    harm category from ``@pytest.mark.harm``. The nodeid and index
    give a total, deterministic ordering for the terminal summary and
    report regardless of xdist worker completion order.

    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.
    """
    tagged = tag_collected_results(node=node, results=collector.results)
    self._results.extend(tagged)
    self._results_by_nodeid[node.nodeid] = tagged
    self._cached_report = None

record_trial_group ¶

Python

record_trial_group(
    *, base_nodeid, clone_nodeids, 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). - Clones 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
`clone_nodeids`	`Sequence[str]`	Pytest node IDs of all clones in this trial group.	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,
    clone_nodeids: Sequence[str],
    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).
    - Clones 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.
        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

    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,
    )

register_trial_spec ¶

Python

register_trial_spec(
    *, clone_nodeid, base_nodeid, threshold
)

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.

Parameters:

Name	Type	Description	Default
`clone_nodeid`	`str`	Node ID of the cloned item.	required
`base_nodeid`	`str`	Node ID of the original (uncloned) item.	required
`threshold`	`float`	Pass-rate threshold from the trial marker.	required

Source code in rampart/pytest_plugin/_session.py

Python
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,
    )

merge_trial_specs ¶

Python

merge_trial_specs(*, trial_specs)

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.

Parameters:

Name	Type	Description	Default
`trial_specs`	`Mapping[str, TrialSpec]`	Specs keyed by clone node ID.	required

Source code in rampart/pytest_plugin/_session.py

Python
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)

merge_worker_results ¶

Python

merge_worker_results(*, results_by_nodeid)

Merge an xdist worker's results into this session.

Extends both the flat _results list and the _results_by_nodeid mapping. Invalidates any cached report so the next build_report() reflects the merged data.

Parameters:

Name	Type	Description	Default
`results_by_nodeid`	`dict[str, list[Result]]`	Worker results grouped by pytest node ID.	required

Source code in rampart/pytest_plugin/_session.py

Python
def merge_worker_results(
    self,
    *,
    results_by_nodeid: dict[str, list[Result]],
) -> None:
    """Merge an xdist worker's results into this session.

    Extends both the flat ``_results`` list and the
    ``_results_by_nodeid`` mapping. Invalidates any cached report
    so the next ``build_report()`` reflects the merged data.

    Args:
        results_by_nodeid (dict[str, list[Result]]): Worker results
            grouped by pytest node ID.
    """
    for nodeid, results in results_by_nodeid.items():
        self._results.extend(results)
        self._results_by_nodeid.setdefault(nodeid, []).extend(results)
    self._cached_report = None

mark_emitted ¶

Python

mark_emitted()

Mark the session as having attempted report emission.

Source code in rampart/pytest_plugin/_session.py

Python
def mark_emitted(self) -> None:
    """Mark the session as having attempted report emission."""
    self._emitted = True

mark_incomplete ¶

Python

mark_incomplete(*, reason)

Record that a worker failed to deliver complete results.

Parameters:

Name	Type	Description	Default
`reason`	`str`	A short human-readable explanation surfaced in the report metadata.	required

Source code in rampart/pytest_plugin/_session.py

Python
def mark_incomplete(self, *, reason: str) -> None:
    """Record that a worker failed to deliver complete results.

    Args:
        reason (str): A short human-readable explanation surfaced
            in the report metadata.
    """
    self._incomplete = True
    if reason not in self._incomplete_reasons:
        self._incomplete_reasons.append(reason)
    self._cached_report = None

set_report_metadata ¶

Python

set_report_metadata(*, metadata)

Attach run-level metadata that will appear on TestRunReport.

Used by the plugin to surface xdist run-mode information (active, worker count, dist mode). Subsequent calls merge into existing metadata.

Parameters:

Name	Type	Description	Default
`metadata`	`dict[str, object]`	Key/value pairs to attach.	required

Source code in rampart/pytest_plugin/_session.py

Python
def set_report_metadata(self, *, metadata: dict[str, object]) -> None:
    """Attach run-level metadata that will appear on ``TestRunReport``.

    Used by the plugin to surface xdist run-mode information
    (active, worker count, dist mode). Subsequent calls merge into
    existing metadata.

    Args:
        metadata (dict[str, object]): Key/value pairs to attach.
    """
    self._report_metadata.update(metadata)
    self._cached_report = None

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 or merged or when metadata is updated.

Results are sorted by (_pytest_nodeid, _rampart_result_index, _rampart_source_worker) for a total, deterministic ordering across xdist worker completion orders. _pytest_nodeid falls back to _pytest_test_name and _rampart_source_worker is absent (constant) outside xdist, so single-process ordering is unaffected.

These leading-underscore keys are RAMPART scheduling bookkeeping, namespaced to avoid colliding with user-supplied result metadata.

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 or merged
    or when metadata is updated.

    Results are sorted by ``(_pytest_nodeid, _rampart_result_index,
    _rampart_source_worker)`` for a total, deterministic ordering across
    xdist worker completion orders. ``_pytest_nodeid`` falls back to
    ``_pytest_test_name`` and ``_rampart_source_worker`` is absent
    (constant) outside xdist, so single-process ordering is unaffected.

    These leading-underscore keys are RAMPART scheduling bookkeeping,
    namespaced to avoid colliding with user-supplied result metadata.

    Returns:
        TestRunReport: Aggregated test run results.
    """
    if self._cached_report is not None:
        return self._cached_report
    sorted_results = sorted(self._results, key=_result_sort_key)
    counts = Counter(r.status for r in sorted_results)
    metadata: dict[str, Any] = dict(self._report_metadata)
    if self._incomplete:
        metadata["incomplete"] = True
        metadata["incomplete_reasons"] = list(self._incomplete_reasons)
    self._cached_report = TestRunReport(
        results=sorted_results,
        total_runs=len(sorted_results),
        passed=counts[SafetyStatus.SAFE],
        failed=counts[SafetyStatus.UNSAFE],
        undetermined=counts[SafetyStatus.UNDETERMINED],
        errors=counts[SafetyStatus.ERROR],
        duration_seconds=self._duration_seconds,
        metadata=metadata,
    )
    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.

Parallel Execution Hooks¶

When pytest-xdist is installed, the plugin streams Result envelopes on call reports, with a non-passing setup fallback when no call report will occur, through pytest_runtest_logreport. It uses the optional pytest_testnodedown hook to reconcile per-worker Result counts. See Parallel Execution for the data flow and trust boundary.

_xdist ¶

xdist support for RAMPART's pytest plugin.

Provides serialization, deserialization, and controller-side merge logic for running RAMPART under pytest-xdist. Workers stream Result objects on call-phase test reports; the controller merges each report incrementally and emits a single unified report at session end.

Trust boundary: worker payloads may contain attacker-controlled content (agent responses, payload text). Serialization is strictly JSON-safe primitives; deserialization validates schema version, enum values, and metadata depth; ANSI escapes are stripped from free text as defense-in-depth.

SCHEMA_VERSION `module-attribute` ¶

Python

SCHEMA_VERSION = 'rampart.xdist.v2'

WORKEROUTPUT_KEY `module-attribute` ¶

Python

WORKEROUTPUT_KEY = 'rampart_xdist_v2'

SIZE_LIMIT_OPTION `module-attribute` ¶

Python

SIZE_LIMIT_OPTION = 'rampart_xdist_max_bytes'

DEFAULT_SIZE_LIMIT_BYTES `module-attribute` ¶

Python

DEFAULT_SIZE_LIMIT_BYTES = 16 * 1024 * 1024

MIN_RESULT_SIZE_LIMIT_BYTES `module-attribute` ¶

Python

MIN_RESULT_SIZE_LIMIT_BYTES = 4 * 1024

WorkerOutputError ¶

Bases: Exception

Base error for xdist worker output processing failures.

SchemaVersionError ¶

Bases: WorkerOutputError

Raised when a worker payload has missing or unknown schema version.

SizeLimitError ¶

Bases: WorkerOutputError

Raised when one serialized Result exceeds the configured size cap.

is_xdist_worker ¶

Python

is_xdist_worker(*, config)

Return True when this process is a pytest-xdist worker.

Detection is attribute-based; no xdist import required, so this function is safe to call when pytest-xdist is not installed.

Parameters:

Name	Type	Description	Default
`config`	`Config`	The pytest configuration object.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if running in an xdist worker process.

Source code in rampart/pytest_plugin/_xdist.py

Python
def is_xdist_worker(*, config: pytest.Config) -> bool:
    """Return True when this process is a pytest-xdist worker.

    Detection is attribute-based; no xdist import required, so this
    function is safe to call when pytest-xdist is not installed.

    Args:
        config (pytest.Config): The pytest configuration object.

    Returns:
        bool: True if running in an xdist worker process.
    """
    return hasattr(config, "workerinput")

is_xdist_controller ¶

Python

is_xdist_controller(*, config)

Return True when this process is the pytest-xdist controller.

The controller is the non-worker process that owns an active distribution: a --dist mode other than "no" plus at least one way of spawning execution endpoints (--numprocesses workers or explicit --tx gateways). Keying off distribution rather than the worker count alone keeps -d/--tx runs (no -n) on the controller path while excluding a bare --dist with no endpoints.

Parameters:

Name	Type	Description	Default
`config`	`Config`	The pytest configuration object.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if running in the xdist controller process.

Source code in rampart/pytest_plugin/_xdist.py

Python
def is_xdist_controller(*, config: pytest.Config) -> bool:
    """Return True when this process is the pytest-xdist controller.

    The controller is the non-worker process that owns an active
    distribution: a ``--dist`` mode other than ``"no"`` plus at least one
    way of spawning execution endpoints (``--numprocesses`` workers or
    explicit ``--tx`` gateways). Keying off distribution rather than the
    worker count alone keeps ``-d``/``--tx`` runs (no ``-n``) on the
    controller path while excluding a bare ``--dist`` with no endpoints.

    Args:
        config (pytest.Config): The pytest configuration object.

    Returns:
        bool: True if running in the xdist controller process.
    """
    if is_xdist_worker(config=config):
        return False
    if get_dist_mode(config=config) == "no":
        return False
    numprocesses = getattr(config.option, "numprocesses", None)
    tx = getattr(config.option, "tx", None)
    return bool(numprocesses) or bool(tx)

get_dist_mode ¶

Python

get_dist_mode(*, config)

Return the active --dist mode string.

Parameters:

Name	Type	Description	Default
`config`	`Config`	The pytest configuration object.	required

Returns:

Name	Type	Description
`str`	`str`	The dist mode (e.g., `"load"`, `"loadgroup"`, `"no"`).

Source code in rampart/pytest_plugin/_xdist.py

Python
def get_dist_mode(*, config: pytest.Config) -> str:
    """Return the active ``--dist`` mode string.

    Args:
        config (pytest.Config): The pytest configuration object.

    Returns:
        str: The dist mode (e.g., ``"load"``, ``"loadgroup"``, ``"no"``).
    """
    return cast("str", getattr(config.option, "dist", "no"))

get_worker_count ¶

Python

get_worker_count(*, config)

Return the number of xdist workers configured.

Parameters:

Name	Type	Description	Default
`config`	`Config`	The pytest configuration object.	required

Returns:

Name	Type	Description
`int`	`int`	Number of workers (0 when xdist is not active).

Source code in rampart/pytest_plugin/_xdist.py

Python
def get_worker_count(*, config: pytest.Config) -> int:
    """Return the number of xdist workers configured.

    Args:
        config (pytest.Config): The pytest configuration object.

    Returns:
        int: Number of workers (0 when xdist is not active).
    """
    numprocesses = getattr(config.option, "numprocesses", 0)
    return int(numprocesses) if numprocesses else 0

attach_report_results ¶

Python

attach_report_results(*, config, report, results)

Attach serialized Results to a call-phase worker report.

Returns:

Name	Type	Description
`int`	`int`	Number of Result representations attached.

Source code in rampart/pytest_plugin/_xdist.py

Python
def attach_report_results(
    *,
    config: pytest.Config,
    report: pytest.TestReport,
    results: list[Result],
) -> int:
    """Attach serialized Results to a call-phase worker report.

    Returns:
        int: Number of Result representations attached.
    """
    if not results:
        return 0
    data = serialize_report_data(
        config=config,
        nodeid=report.nodeid,
        results=results,
    )
    setattr(report, REPORT_RESULTS_ATTR, data)
    return len(results)

serialize_report_data ¶

Python

serialize_report_data(*, config, nodeid, results)

Serialize call-phase Results into an execnet-safe report envelope.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: JSON-safe report envelope.

Source code in rampart/pytest_plugin/_xdist.py

Python
def serialize_report_data(
    *,
    config: pytest.Config,
    nodeid: str,
    results: list[Result],
) -> dict[str, Any]:
    """Serialize call-phase Results into an execnet-safe report envelope.

    Returns:
        dict[str, Any]: JSON-safe report envelope.
    """
    limit_bytes = _size_limit(config=config)
    return {
        "schema": SCHEMA_VERSION,
        "nodeid": nodeid,
        "results": [
            _serialize_capped_result(
                limit_bytes=limit_bytes,
                result=result,
                nodeid=nodeid,
            )
            for result in results
        ],
    }

deserialize_report_data ¶

Python

deserialize_report_data(*, data, report_nodeid)

Deserialize one call-phase report envelope.

Performs strict schema validation: missing schema key, unknown versions, and malformed enum values all raise WorkerOutputError (or subclass). The report nodeid and envelope nodeid must agree.

Each result's metadata["_pytest_nodeid"] and metadata["_rampart_result_index"] are set authoritatively from the envelope nodeid and list position so cross-worker ordering is total and independent of any (untrusted) serialized values.

Parameters:

Name	Type	Description	Default
`data`	`object`	The deserialized private TestReport attribute.	required
`report_nodeid`	`str`	The nodeid from the owning TestReport.	required

Returns:

Type	Description
`tuple[dict[str, list[Result]], bool]`	tuple[dict[str, list[Result]], bool]: Results grouped by nodeid and whether any Result is a truncation marker.

Raises:

Type	Description
`SchemaVersionError`	Missing or unknown schema version.
`WorkerOutputError`	Malformed payload (type errors, bad enums).

Source code in rampart/pytest_plugin/_xdist.py

Python
def deserialize_report_data(
    *,
    data: object,
    report_nodeid: str,
) -> tuple[dict[str, list[Result]], bool]:
    """Deserialize one call-phase report envelope.

    Performs strict schema validation: missing ``schema`` key, unknown
    versions, and malformed enum values all raise ``WorkerOutputError``
    (or subclass). The report nodeid and envelope nodeid must agree.

    Each result's ``metadata["_pytest_nodeid"]`` and
    ``metadata["_rampart_result_index"]`` are set authoritatively from the
    envelope nodeid and list position so cross-worker ordering is total
    and independent of any (untrusted) serialized values.

    Args:
        data (object): The deserialized private TestReport attribute.
        report_nodeid (str): The nodeid from the owning TestReport.

    Returns:
        tuple[dict[str, list[Result]], bool]: Results grouped by nodeid
            and whether any Result is a truncation marker.

    Raises:
        SchemaVersionError: Missing or unknown schema version.
        WorkerOutputError: Malformed payload (type errors, bad enums).
    """
    typed = _validate_schema(data=data)
    nodeid = typed.get("nodeid")
    if not isinstance(nodeid, str) or not nodeid:
        msg = "Streamed report envelope has an invalid 'nodeid'."
        raise WorkerOutputError(msg)
    if nodeid != report_nodeid:
        msg = (
            f"Streamed envelope nodeid {nodeid!r} does not match "
            f"TestReport nodeid {report_nodeid!r}."
        )
        raise WorkerOutputError(msg)
    raw_results = typed.get("results")
    if not isinstance(raw_results, list):
        msg = f"Expected list for results, got {type(raw_results).__name__}."
        raise WorkerOutputError(msg)
    deserialized: list[Result] = []
    truncated = False
    for index, raw_result in enumerate(cast("list[Any]", raw_results)):
        result = _deserialize_result(data=raw_result)
        result.metadata["_pytest_nodeid"] = nodeid
        result.metadata["_rampart_result_index"] = index
        deserialized.append(result)
        if isinstance(raw_result, dict) and raw_result.get(_TRUNCATED_MARKER) is True:
            truncated = True
    return {nodeid: deserialized}, truncated

merge_report_results ¶

Python

merge_report_results(*, session, report)

Validate and incrementally merge one streamed report envelope.

Returns:

Type	Description
`tuple[str, int] \| None`	tuple[str, int] \| None: Source worker and merged Result count, or None when the report carries no RAMPART envelope.

Source code in rampart/pytest_plugin/_xdist.py

Python
def merge_report_results(
    *,
    session: RampartSession,
    report: pytest.TestReport,
) -> tuple[str, int] | None:
    """Validate and incrementally merge one streamed report envelope.

    Returns:
        tuple[str, int] | None: Source worker and merged Result count,
            or None when the report carries no RAMPART envelope.
    """
    payload = getattr(report, REPORT_RESULTS_ATTR, None)
    if payload is None:
        return None
    worker_id = _report_worker_id(report=report)
    results_by_nodeid, truncated = deserialize_report_data(
        data=payload,
        report_nodeid=report.nodeid,
    )
    _tag_source_worker(
        results_by_nodeid=results_by_nodeid,
        worker_id_str=worker_id,
    )
    session.merge_worker_results(results_by_nodeid=results_by_nodeid)
    result_count = sum(len(results) for results in results_by_nodeid.values())
    if truncated:
        session.mark_incomplete(
            reason=f"worker {worker_id} streamed a truncated Result (size cap)",
        )
    return worker_id, result_count

serialize_worker_data ¶

Python

serialize_worker_data(*, session, streamed_result_count)

Serialize slim session-level worker data for the controller.

Results are deliberately absent because call-phase reports are the sole Result transport. Workeroutput retains trial specs and the expected streamed Result count for completeness reconciliation.

Parameters:

Name	Type	Description	Default
`session`	`RampartSession`	The worker's session state.	required
`streamed_result_count`	`int`	Result representations attached to reports by this worker.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: A JSON-safe payload ready to write to `config.workeroutput`.

Source code in rampart/pytest_plugin/_xdist.py

Python
def serialize_worker_data(
    *,
    session: RampartSession,
    streamed_result_count: int,
) -> dict[str, Any]:
    """Serialize slim session-level worker data for the controller.

    Results are deliberately absent because call-phase reports are the
    sole Result transport. Workeroutput retains trial specs and the
    expected streamed Result count for completeness reconciliation.

    Args:
        session (RampartSession): The worker's session state.
        streamed_result_count (int): Result representations attached to
            reports by this worker.

    Returns:
        dict[str, Any]: A JSON-safe payload ready to write to
            ``config.workeroutput``.
    """
    return {
        "schema": SCHEMA_VERSION,
        _STREAMED_RESULT_COUNT: streamed_result_count,
        "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()
        ],
    }

deserialize_trial_specs ¶

Python

deserialize_trial_specs(*, data)

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.

Parameters:

Name	Type	Description	Default
`data`	`object`	The deserialized JSON object from `node.workeroutput`.	required

Returns:

Type	Description
`dict[str, TrialSpec]`	dict[str, TrialSpec]: Trial specs keyed by clone node ID.

Raises:

Type	Description
`SchemaVersionError`	Missing or unknown schema version.
`WorkerOutputError`	`data` is not a dict payload.

Source code in rampart/pytest_plugin/_xdist.py

Python
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

finalize_worker ¶

Python

finalize_worker(*, config, session, streamed_result_count)

Serialize slim worker session state into config.workeroutput.

Called from pytest_sessionfinish on each xdist worker. The worker skips sink emission entirely; the controller is responsible for the final report.

Parameters:

Name	Type	Description	Default
`config`	`Config`	The pytest configuration object.	required
`session`	`RampartSession`	The worker's session state.	required
`streamed_result_count`	`int`	Number of Result representations attached to test reports by this worker.	required

Source code in rampart/pytest_plugin/_xdist.py

Python
def finalize_worker(
    *,
    config: pytest.Config,
    session: RampartSession,
    streamed_result_count: int,
) -> None:
    """Serialize slim worker session state into ``config.workeroutput``.

    Called from ``pytest_sessionfinish`` on each xdist worker. The
    worker skips sink emission entirely; the controller is responsible
    for the final report.

    Args:
        config (pytest.Config): The pytest configuration object.
        session (RampartSession): The worker's session state.
        streamed_result_count (int): Number of Result representations
            attached to test reports by this worker.
    """
    if not is_xdist_worker(config=config):
        return
    workeroutput = cast(
        "dict[str, Any]",
        config.workeroutput,  # ty: ignore[unresolved-attribute]
    )
    workeroutput[WORKEROUTPUT_KEY] = serialize_worker_data(
        session=session,
        streamed_result_count=streamed_result_count,
    )

handle_testnodedown ¶

Python

handle_testnodedown(
    *, session, node, error, received_result_count
)

Reconcile a finished worker's streamed Result count.

Called from pytest_testnodedown on the controller for each worker that completes. Failures (missing payload, deserialization errors, worker crashes) are recorded via mark_incomplete rather than raised, so a single bad worker does not abort report emission.

Parameters:

Name	Type	Description	Default
`session`	`RampartSession`	The controller's session state.	required
`node`	`object`	The xdist node object (has `workeroutput` attribute).	required
`error`	`object`	The shutdown error from xdist, or None on clean exit.	required
`received_result_count`	`int`	Results already merged from this worker.	required

Source code in rampart/pytest_plugin/_xdist.py

Python
def handle_testnodedown(
    *,
    session: RampartSession,
    node: object,
    error: object,
    received_result_count: int,
) -> None:
    """Reconcile a finished worker's streamed Result count.

    Called from ``pytest_testnodedown`` on the controller for each
    worker that completes. Failures (missing payload, deserialization
    errors, worker crashes) are recorded via ``mark_incomplete`` rather
    than raised, so a single bad worker does not abort report emission.

    Args:
        session (RampartSession): The controller's session state.
        node: The xdist node object (has ``workeroutput`` attribute).
        error: The shutdown error from xdist, or None on clean exit.
        received_result_count (int): Results already merged from this worker.
    """
    worker_id_str = get_worker_id(node)
    if error is not None:
        logger.warning(
            "Worker %s reported shutdown error; report will be incomplete: %s",
            worker_id_str,
            error,
        )
        session.mark_incomplete(reason=f"worker {worker_id_str} error: {error}")
        return
    workeroutput = getattr(node, "workeroutput", None)
    if not isinstance(workeroutput, dict):
        logger.warning(
            "Worker %s exited without workeroutput; report will be incomplete.",
            worker_id_str,
        )
        session.mark_incomplete(reason=f"worker {worker_id_str} missing workeroutput")
        return
    payload: Any = cast("dict[str, Any]", workeroutput).get(WORKEROUTPUT_KEY)
    if payload is None:
        logger.warning(
            "Worker %s did not produce RAMPART output; report will be incomplete.",
            worker_id_str,
        )
        session.mark_incomplete(reason=f"worker {worker_id_str} missing RAMPART output")
        return
    trial_specs = _safe_deserialize_trial_specs(
        payload=cast("object", payload),
        worker_id_str=worker_id_str,
    )
    if trial_specs:
        session.merge_trial_specs(trial_specs=trial_specs)
    try:
        expected_result_count = _deserialize_streamed_result_count(payload=payload)
    except WorkerOutputError:
        logger.exception(
            "Worker %s streamed Result count is invalid; report will be incomplete.",
            worker_id_str,
        )
        session.mark_incomplete(
            reason=f"worker {worker_id_str} missing streamed Result count",
        )
        return
    if expected_result_count != received_result_count:
        logger.error(
            "Worker %s streamed Result count mismatch: expected %d, received %d.",
            worker_id_str,
            expected_result_count,
            received_result_count,
        )
        session.mark_incomplete(
            reason=(
                f"worker {worker_id_str} streamed Result count mismatch "
                f"(expected {expected_result_count}, received {received_result_count})"
            ),
        )

API Reference — pytest Plugin¶

_collection ¶

ResultCollectionHandler ¶

on_event_async async ¶

ResultCollector ¶

results property ¶

record ¶

record_result ¶

_session ¶

RampartSession ¶

sinks property ¶

results_by_nodeid property ¶

is_emitted property ¶

is_incomplete property ¶

incomplete_reasons property ¶

has_results property ¶

trial_groups property ¶

trial_specs property ¶

add_sinks ¶

set_duration ¶

absorb ¶

record_trial_group ¶

register_trial_spec ¶

merge_trial_specs ¶

merge_worker_results ¶

mark_emitted ¶

mark_incomplete ¶

set_report_metadata ¶

build_report ¶

TrialGroupResult dataclass ¶

verdict property ¶

terminal_label property ¶

detail property ¶

has_unsafe property ¶

Parallel Execution Hooks¶

_xdist ¶

SCHEMA_VERSION module-attribute ¶

WORKEROUTPUT_KEY module-attribute ¶

SIZE_LIMIT_OPTION module-attribute ¶

DEFAULT_SIZE_LIMIT_BYTES module-attribute ¶

MIN_RESULT_SIZE_LIMIT_BYTES module-attribute ¶

WorkerOutputError ¶

SchemaVersionError ¶

SizeLimitError ¶

is_xdist_worker ¶

is_xdist_controller ¶

get_dist_mode ¶

get_worker_count ¶

attach_report_results ¶

serialize_report_data ¶

deserialize_report_data ¶

merge_report_results ¶

serialize_worker_data ¶

deserialize_trial_specs ¶

finalize_worker ¶

handle_testnodedown ¶

on_event_async `async` ¶

results `property` ¶

sinks `property` ¶

results_by_nodeid `property` ¶

is_emitted `property` ¶

is_incomplete `property` ¶

incomplete_reasons `property` ¶

has_results `property` ¶

trial_groups `property` ¶

trial_specs `property` ¶

TrialGroupResult `dataclass` ¶

verdict `property` ¶

terminal_label `property` ¶

detail `property` ¶

has_unsafe `property` ¶

SCHEMA_VERSION `module-attribute` ¶

WORKEROUTPUT_KEY `module-attribute` ¶

SIZE_LIMIT_OPTION `module-attribute` ¶

DEFAULT_SIZE_LIMIT_BYTES `module-attribute` ¶

MIN_RESULT_SIZE_LIMIT_BYTES `module-attribute` ¶