Skip to content

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 declares population configuration but does not create pytest items, so it does not change xdist scheduling. A marked test runs on one worker like any other test and receives its effective values through trial_config.

Use --rampart-trials=N to change the population depth supplied to selected tests. Parallelizing the executions within a test is the responsibility of that test or its population-execution helper.

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.