Testing Experiments Without Hardware

Experiments are tested entirely without instruments: the contract gates verify the experiment's structure, and step operations run against mocked device clients.

The ExperimentPipeline contract gates

plesty.lib.test.experiment_pipeline.ExperimentPipeline is the standard test pipeline for experiment modules — the experiment counterpart of the device DevicePipeline; its five contract tests are enforced by gate e1 when module_type = "experiment". It instantiates your experiment class (without devices — plan construction is device-free by contract) and checks the framework contract:

Wire them into your test suite one per test function, as the demo experiment does:

from pathlib import Path
from plesty.lib.test.experiment_pipeline import ExperimentPipeline
from plesty.scan_exp import ScanExperiment


def _pipeline(tmp_path: Path) -> ExperimentPipeline:
    return ExperimentPipeline(ScanExperiment, points=3, run_root=tmp_path)


def test_plan_deterministic(tmp_path: Path) -> None:
    _pipeline(tmp_path).test_plan_deterministic()


def test_plan_ops_resolve(tmp_path: Path) -> None:
    _pipeline(tmp_path).test_plan_ops_resolve()


def test_mock_pipeline_runs(tmp_path: Path) -> None:
    """One-shot: executes the full run without hardware."""
    _pipeline(tmp_path).run_mock_pipeline()

Testing step operations with mocked devices

Step operations that drive devices are tested by handing the experiment a CompositeDevice of mocks:

import asyncio
from unittest.mock import MagicMock
from plesty.lib.device.composite_device import CompositeDevice
from plesty.scan_exp import ScanExperiment


def make_mock_device(query_value: float = 0.005) -> MagicMock:
    mock = MagicMock()
    mock.query.return_value = query_value
    mock.write.return_value = True
    mock.identity.return_value = "Mock Device v1.0"
    return mock


def test_measure_point_reads_power(tmp_path) -> None:
    devices = CompositeDevice({
        "stage": make_mock_device(),
        "powermeter": make_mock_device(query_value=0.005),
    })
    experiment = ScanExperiment(devices=devices, points=3, run_root=tmp_path)

    result = asyncio.run(experiment.measure_point(x=1))

    devices.stage.write.assert_called_with("POSITION", 1)
    assert result.data[0] == 0.005

Async step operations run in tests via asyncio.run(...) — no extra pytest plugin needed.

End-to-end: run and resume

The most valuable test executes a full run against the real persistence layer (in tmp_path) and verifies resume semantics:

def test_run_persists_plan_journal_and_data(tmp_path) -> None:
    experiment = ScanExperiment(points=3, run_root=tmp_path)
    run_id = asyncio.run(experiment.run())

    run_dir = tmp_path / run_id
    assert (run_dir / "plan.json").exists()
    assert (run_dir / "journal.jsonl").exists()
    assert len(list((run_dir / "data").glob("*.json"))) == 3


def test_resume_skips_completed_steps(tmp_path) -> None:
    run_id = asyncio.run(ScanExperiment(points=3, run_root=tmp_path).run())
    # resuming a completed run re-measures nothing
    asyncio.run(ScanExperiment(points=3, run_root=tmp_path).run(resume=run_id))

Coverage requirement (Gate 7)

Gate 7 requires ≥ 80% test coverage. For experiment modules, cover:

1. The ExperimentPipeline contract gates
2. Step operations with mocked instruments (including error paths — a device raising mid-step)
3. A full run + resume against tmp_path

uv run pytest tests/ -v
uv run pytest tests/ --cov=plesty/scan_exp --cov-report=term-missing

Gate d1 itself does not apply to experiments — it checks the device DevicePipeline tests and is activated only when module_type = "device".