Parallel Execution with pytest-xdist¶

RAMPART supports parallel test execution via pytest-xdist, producing a single unified report even when tests run across multiple worker processes.

Quick Start¶

Bash

pip install pytest-xdist
pytest -n 4

With -n 4, pytest spawns 4 worker processes that execute tests in parallel. RAMPART intercepts each worker's results, ships them to the controller process, and emits one consolidated report at the end of the session.

How It Works¶

Text Only

Worker 1                    Worker 2                    Controller
─────────                   ─────────                   ──────────
eligible Results             eligible Results
    │                           │
serialize → TestReport      serialize → TestReport
    │                           │
    └───────────┬───────────────┘
                ▼
        pytest_runtest_logreport
        validate + merge incrementally
                │
                ▼
        pytest_testnodedown
        reconcile streamed Result counts
                │
                ▼
        pytest_sessionfinish (controller)
        aggregate trials → evaluate gates → emit sinks
                │
                ▼
        Single unified TestRunReport

Workers attach JSON-safe serialized Result objects to each call-phase TestReport, or to a non-passing setup report when setup recorded Results and no call report will occur. Workers do not emit RAMPART report sinks.
Controller receives each envelope through pytest_runtest_logreport, validates and merges it into its RampartSession, and emits sinks once at session end.
Worker shutdown carries only trial specifications and the expected number of streamed Result representations in config.workeroutput. The controller reconciles that count in pytest_testnodedown; Results are never delivered through both paths.

The normal transport boundary is the call phase, which includes Results recorded during successful fixture setup. A failed or skipped setup that recorded Results uses its setup report as a fallback because no call report follows. Results recorded during fixture teardown are not streamed.

The result: one JsonFileReportSink output file, one call to MyCustomSink.emit_async, and accurate population statistics over the full result set.

Trial 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.

`--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 Only

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.

Registering Sinks: the `pytest_rampart_sinks` hook¶

The recommended way to register report sinks is the pytest_rampart_sinks hook. It is resolved on the controller — which never executes fixtures — so it behaves identically in single-process and xdist runs, and (unlike the fixture path) supports sinks that need configuration.

Implement it in your conftest.py:

Python

# conftest.py
from pathlib import Path

from rampart.reporting import JsonFileReportSink


def pytest_rampart_sinks(config):
    return [JsonFileReportSink(output_dir=Path(".report"))]

Multiple implementations are supported; RAMPART emits to the union of every returned sink.
An implementation may return an empty list to contribute none.
Non-ReportSink items (or a non-list return) are dropped with a warning, so one malformed implementation cannot break emission.

If your sinks need dependencies, build them inside the hook — it receives the pytest.Config and runs on the controller, so you can build sinks from config values or environment variables there.

Trust Boundary & Security¶

Worker payloads cross a process boundary via execnet and may contain attacker-controlled content (agent responses, payload text, evaluator rationale). RAMPART's serialization defends against:

Arbitrary code execution — strict JSON-safe primitives only; no pickle, marshal, or custom __reduce__.
Schema drift — payloads with missing or unknown schema versions are rejected fail-closed.
Memory exhaustion — each serialized Result is capped at 16 MiB by default.
Terminal/log injection — ANSI escape sequences are stripped from free-form text at the deserialization boundary.
Path traversal — worker-local artifact paths are stored as opaque strings in metadata; the controller never accesses worker files.

Size cap¶

The default 16 MiB cap can be overridden via the pytest CLI option or an ini setting:

Bash

pytest -n 4 --rampart-xdist-max-bytes=134217728

Or in pytest.ini / pyproject.toml:

INI

[pytest]
rampart_xdist_max_bytes = 134217728

An oversized Result is replaced by an attributed ERROR/truncation marker while normal Results from the same worker continue to stream. The controller records the run as incomplete in TestRunReport.metadata. Configured limits below 4 KiB use a 4 KiB effective minimum so the marker itself always fits.

Incomplete Runs¶

If a worker crashes, drops a streamed Result, omits its final count, or hits the size cap, the controller marks the run as incomplete:

Python

report.metadata["incomplete"]            # True if any worker failed
report.metadata["incomplete_reasons"]    # list[str] — one per failure

Reports are still emitted with whatever data was collected. For safety-critical CI, sinks or post-processing should check the incomplete flag and fail the build accordingly.

Run-Mode Metadata¶

Reports produced under xdist include:

Python

report.metadata["xdist_active"]   # True
report.metadata["worker_count"]   # int
report.metadata["dist_mode"]      # "load", "loadgroup", etc.

Durability behavior¶

Each eligible report is merged as it reaches the controller. If a worker is killed mid-run, every Result already delivered remains in the final report and the run is marked incomplete because the worker cannot provide a final expected count. A clean worker shutdown publishes its emitted Result count; any mismatch with the controller's received count detects a silent drop and also marks the run incomplete.

The cap applies independently to each Result. An oversized transcript therefore does not discard normal Results from that worker.

Limitations¶

Results recorded only during fixture teardown are outside the report-streaming boundary and are not included.
A worker that dies can lose Results whose eligible reports had not reached the controller; already-streamed Results are retained and the run is marked incomplete (see Durability behavior).
Mixed RAMPART versions across controller and workers are unsupported; install the same version everywhere.
pytest-xdist itself does not support interactive debugging (--pdb, --trace); use single-process mode for debugging.

Parallel Execution with pytest-xdist¶

Quick Start¶

How It Works¶

Trial Tests with xdist¶

Choosing loadgroup vs load¶

Registering Sinks: the pytest_rampart_sinks hook¶

Trust Boundary & Security¶

Size cap¶

Incomplete Runs¶

Run-Mode Metadata¶

Durability behavior¶

Limitations¶

Choosing `loadgroup` vs `load`¶

Registering Sinks: the `pytest_rampart_sinks` hook¶