Plesty Documentation

Test It

The five contract tests gate e1 runs, a stand-in rig for the step methods, and a full run and resume — all without hardware.

An experiment is tested with no instrument at all — not even the demo servers. The scaffold writes the tests; what you change is what they are given once the template's example became your measurement.

The contract tests

ExperimentPipeline, from plesty.lib.test.experiment_pipeline, builds your experiment without instruments and checks the contract. Gate e1 looks for all five tests by name:

Test Checks
test_experiment_subclass one public Experiment subclass, built without devices
test_plan_deterministic build_plan() gives the same plan twice
test_plan_ops_resolve every step's op is a method of the class
test_plan_serializable the plan survives saving and loading unchanged
test_lifecycle_hooks setup and teardown are async

The pipeline passes its extra arguments to your class, so give it settings your build_plan() can use — three rows are enough:

from plesty.lib.test.experiment_pipeline import ExperimentPipeline
from plesty.demo_pow_pl.experiment import Experiment

CONFIG = {"start_deg": 0.0, "stop_deg": 20.0, "step_deg": 10.0, "averaging": 1}
PIPELINE = ExperimentPipeline(Experiment, config=CONFIG)


def test_plan_deterministic() -> None:
    PIPELINE.test_plan_deterministic()

# … and the other four, the same way

The scaffold's other tests still speak of its example: points in test_load_reads_the_config_file is only a value read back, and can stay; test_build_devices_declares_the_rig passes --device-address and checks the device sub-device, both removed in The Rig — parse no arguments there, and check one of your own instruments instead:

    args = cli.build_parser().parse_args([])
    …
    assert config["hwp"]["requires"] == list(RIG["hwp"])

A stand-in for the rig

To run the steps without instruments, give the experiment an object that answers the calls the steps make — like the stage, the spectrometer and the power meter would:

class FakeComposite:
    connected = False

    def connect_all(self) -> None:
        self.connected = True

    def disconnect_all(self) -> None:
        self.connected = False

    def release(self) -> None:
        self.connected = False

    def set_data_path(self, path, subdirs=False, devices=None, **kwargs) -> dict:
        """Accept the run's raw folder and answer each device's, as the rig would."""
        return {name: f"{path}/{name}" for name in devices or []}

    def call(self, name, operation, *args, **kwargs):
        if operation == "move_absolute":
            return args[0]                 # the stage reached the angle asked for
        if operation == "measure_power":
            return {"power": 1.5e-5}
        if operation == "get_recent_file":
            return "frame.spe"             # the spectrometer's newest file
        return True

A whole run, and a resume

The most useful tests run the plan for real. A row names its spectrum relative to the shared disk, so the test gives the experiment one — a temporary folder:

def test_run_completes(tmp_path) -> None:
    devices = FakeComposite()
    experiment = Experiment(devices=devices, config=CONFIG, data_root=tmp_path, run_root="runs")
    run_id = asyncio.run(experiment.run())
    assert "run_completed" in (tmp_path / "runs" / run_id / "journal.jsonl").read_text()
    assert devices.connected is False            # teardown ran


def test_resume_skips_completed_steps(tmp_path) -> None:
    first = Experiment(devices=FakeComposite(), config=CONFIG, data_root=tmp_path, run_root="runs")
    run_id = asyncio.run(first.run())
    again = Experiment(devices=FakeComposite(), config=CONFIG, data_root=tmp_path, run_root="runs")
    assert asyncio.run(again.run(resume=run_id)) == run_id

The analysis step runs too, on the stand-in's constant power — so the tests also prove that MalusFit is called with what the rows hold and that its result and report are saved.

The viewer

viz.py is code like the rest, and counts toward coverage. A plot is a function of rows, so it is tested on rows written by hand:

def test_viewer_draws_the_power_normalised(tmp_path: Path) -> None:
    """The custom plot scales the power to its highest reading; the analysis row is skipped."""
    import numpy as np
    from plesty.lib.monitor import RunData

    from plesty.demo_pow_pl.viz import normalised, read_spectrum

    assert read_spectrum({"mean_power_w": 1.0}, None) == {}  # a row without a spectrum
    rows = [
        {"hwp_deg": 0.0, "power_w": 1.0},
        {"hwp_deg": 2.0, "power_w": 4.0},
        {"modulation": 1.0},
    ]
    (line,) = normalised(RunData(str(tmp_path), rows=rows))
    assert np.allclose(line.y[:2], [0.25, 1.0])
$ uv run pytest -q
............                                                             [100%]
12 passed in 0.11s
$ uv run plesty check --standard quantum
…
All checks passed.

Gate 7 asks for 80 % coverage once the module is released at the quantum standard: the contract tests, the step methods through the stand-in, a run and resume, and the viewer test reach 94 % here. Without the viewer test the module stays at 76 % and the gate fails. MalusFit deserves a test of its own too — a curve with a known axis, fitted back.

Next: Run It from the Bench.