plesty.lib.test.field_test ========================== .. py:module:: plesty.lib.test.field_test .. autoapi-nested-parse:: Standard on-hardware field test pipeline for PLESTY device modules. :class:`~plesty.lib.test.device_pipeline.DevicePipeline` verifies a device module without hardware. A module can pass every mock gate and still fail on the real instrument: the command-format contract is only exercised by the real traffic manager, driver binding only happens against the real driver, and transport pathologies (wedged USBTMC sessions, a stale response left in the read buffer by a timed-out query) have no mock equivalent. :class:`FieldTestPipeline` is the on-hardware counterpart. It runs a fixed gate sequence against a connected instrument and produces a :class:`FieldTestReport` — per-gate status plus a **timing profile** and **failure statistics** per command, so timeouts and poll intervals become measured values instead of guesses, and flaky commands become visible instead of anecdotal. The markdown and JSONL reports are internal artifacts: they carry the host name and the instrument address. Keep them out of public repositories — the scaffold git-ignores them. What *is* published is ``reports//field-test.json``, the sanitised projection built by :mod:`plesty.lib.test.report_artifact`: what was verified, how it performed, and what the run found wrong with the module, with nothing that says where it ran. Pass ``instrument=`` and ``reports_dir=`` to have :meth:`run` write it. Typical usage in a device module:: # tests/field_test.py — hardware-gated, skipped in CI from plesty.lib.test.field_test import FieldTestPipeline from plesty.lib.test.report_artifact import Instrument PIPELINE = FieldTestPipeline( MyDevice, "USB0::0x1313::0x8078::P0001234::INSTR", repetitions=20, safe_ops=["measure_power"], report_path="reports/field-test-mydevice", instrument=Instrument(model="PM100D", sensor="S130C"), reports_dir="reports", ) if __name__ == "__main__": report = PIPELINE.run() raise SystemExit(0 if report.ok else 1) Gate summary ------------ - Gate 1 ``test_discovery`` — the address resolves to a real resource. - Gate 2 ``test_connect_lifecycle`` — connect, identity, check_errors, disconnect and reconnect, each timed. - Gate 3 ``test_param_roundtrip`` — every writable parameter round-trips on hardware; the value read back is written back, so device state is unchanged. - Gate 4 ``test_param_constraints`` — hardware MIN/MAX and option sets against the schema envelope. - Gate 5 ``test_functions`` — every operation on the safe allow-list, driven by its declared kind: a motion one step out and back, a lifecycle op once, a control op once at rest after the motions. - Gate 6 ``test_buffer_drain`` — a query abandoned mid-flight must not leave its answer behind for the next query to consume. - Gate 7 ``test_error_recovery`` — the device recovers from a forced timeout, and the first command after a reconnect succeeds. - Gate 8 ``test_stability`` — every queryable parameter and safe op repeated ``repetitions`` times to collect the statistics above. Attributes ---------- .. autoapisummary:: plesty.lib.test.field_test.PASS plesty.lib.test.field_test.FAIL plesty.lib.test.field_test.SKIP plesty.lib.test.field_test.HOST plesty.lib.test.field_test.CLIENT plesty.lib.test.field_test.FAILURE_CLASSES plesty.lib.test.field_test._TIMEOUT_MARKERS plesty.lib.test.field_test._DEVICE_ERROR_MARKERS plesty.lib.test.field_test.GATE_SEQUENCE plesty.lib.test.field_test.GATE_NAMES plesty.lib.test.field_test._REPEATABLE_KINDS plesty.lib.test.field_test.LIFECYCLE_COMMANDS plesty.lib.test.field_test._VERIFICATION_RANK Exceptions ---------- .. autoapisummary:: plesty.lib.test.field_test.FieldTestError plesty.lib.test.field_test.WrongResponseError plesty.lib.test.field_test.DeviceReportedError plesty.lib.test.field_test._SkipGate Classes ------- .. autoapisummary:: plesty.lib.test.field_test.LatencyStats plesty.lib.test.field_test.FailureStats plesty.lib.test.field_test.GateResult plesty.lib.test.field_test.FieldTestReport plesty.lib.test.field_test.FieldTestPipeline Functions --------- .. autoapisummary:: plesty.lib.test.field_test.classify_failure plesty.lib.test.field_test._exception_chain plesty.lib.test.field_test._summarise plesty.lib.test.field_test._run_gate plesty.lib.test.field_test.utc_now plesty.lib.test.field_test.environment_info plesty.lib.test.field_test._exposed_methods plesty.lib.test.field_test._release plesty.lib.test.field_test._traffic_manager plesty.lib.test.field_test._as_number plesty.lib.test.field_test._compare_range plesty.lib.test.field_test._values_equal plesty.lib.test.field_test._distinguishable Module Contents --------------- .. py:data:: PASS :value: 'pass' .. py:data:: FAIL :value: 'fail' .. py:data:: SKIP :value: 'skip' .. py:data:: HOST :value: 'host' .. py:data:: CLIENT :value: 'client' .. py:data:: FAILURE_CLASSES :type: tuple[str, Ellipsis] :value: ('timeout', 'wrong_response', 'device_error', 'exception') .. py:data:: _TIMEOUT_MARKERS :value: ('timeout', 'timed out', 'vi_error_tmo') .. py:data:: _DEVICE_ERROR_MARKERS :value: ('device error', 'error queue', 'syst:err', 'not operable') .. py:data:: GATE_SEQUENCE :type: tuple[tuple[str, str, bool], Ellipsis] :value: (('discovery', 'test_discovery', False), ('connect_lifecycle', 'test_connect_lifecycle', False),... .. py:data:: GATE_NAMES :type: tuple[str, Ellipsis] .. py:data:: _REPEATABLE_KINDS :type: frozenset[str] .. py:data:: LIFECYCLE_COMMANDS :type: tuple[str, Ellipsis] :value: ('connect', 'identity', 'check_errors', 'check_operatability', 'disconnect', 'reconnect') .. py:data:: _VERIFICATION_RANK :type: dict[str, int] .. py:exception:: FieldTestError Bases: :py:obj:`RuntimeError` Base class for failures raised by the field test pipeline itself. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: WrongResponseError Bases: :py:obj:`FieldTestError` The instrument answered, but with a value the gate cannot accept. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: DeviceReportedError Bases: :py:obj:`FieldTestError` The instrument reported an error through its own error channel. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: _SkipGate Bases: :py:obj:`Exception` Internal signal: a gate cannot run and is recorded as skipped. Initialize self. See help(type(self)) for accurate signature. .. py:function:: classify_failure(exc: BaseException) -> str Classify an exception into one of :data:`FAILURE_CLASSES`. The whole ``__cause__``/``__context__`` chain is inspected, because the transport layer wraps the original error: a VISA timeout surfaces as a ``RuntimeError("Error sending command …")`` whose cause is the timeout. Classifying only the outermost exception would report every transport failure as a generic exception. :param exc: The exception raised by a device call. :returns: One of ``"timeout"``, ``"wrong_response"``, ``"device_error"`` or ``"exception"``. .. py:function:: _exception_chain(exc: BaseException, limit: int = 8) -> list[BaseException] Return *exc* followed by its causes, outermost first. :param exc: The exception to unwrap. :param limit: Maximum chain length, guarding against cyclic contexts. .. py:class:: LatencyStats Latency distribution of one command over all successful attempts. :ivar count: Number of successful timed calls. :ivar min_s: Fastest observed call, in seconds. :ivar median_s: Median call duration, in seconds. :ivar p95_s: 95th percentile (nearest-rank), in seconds. :ivar max_s: Slowest observed call, in seconds. :ivar mean_s: Arithmetic mean, in seconds. .. py:attribute:: count :type: int .. py:attribute:: min_s :type: float .. py:attribute:: median_s :type: float .. py:attribute:: p95_s :type: float .. py:attribute:: max_s :type: float .. py:attribute:: mean_s :type: float .. py:method:: from_samples(samples: collections.abc.Sequence[float]) -> LatencyStats :classmethod: Build statistics from a sequence of durations in seconds. :param samples: Measured durations. An empty sequence yields an all-zero record with ``count == 0``. .. py:method:: to_dict() -> dict[str, Any] Return a JSON-serialisable mapping of the statistics. .. py:class:: FailureStats Attempt and failure counts of one command, broken down by failure class. :ivar attempts: Total number of calls, successful or not. :ivar by_class: Failure count per entry of :data:`FAILURE_CLASSES`. :ivar last_error: String form of the most recent failure, if any. .. py:attribute:: attempts :type: int :value: 0 .. py:attribute:: by_class :type: dict[str, int] .. py:attribute:: last_error :type: str | None :value: None .. py:property:: failures :type: int Total number of failed calls across all classes. .. py:property:: failure_rate :type: float Failed calls divided by attempts; ``0.0`` when never attempted. .. py:method:: record_success() -> None Count one successful call. .. py:method:: record_failure(exc: BaseException) -> str Count one failed call and return the failure class it was assigned. :param exc: The exception raised by the call. .. py:method:: to_dict() -> dict[str, Any] Return a JSON-serialisable mapping of the statistics. .. py:class:: GateResult Outcome of a single field test gate. :ivar name: Gate name, e.g. ``"buffer_drain"``. :ivar status: One of :data:`PASS`, :data:`FAIL` or :data:`SKIP`. :ivar duration_s: Wall-clock duration of the gate. :ivar evidence: Gate-specific observations kept for the report. :ivar error: Failure or skip reason, if any. .. py:attribute:: name :type: str .. py:attribute:: status :type: str .. py:attribute:: duration_s :type: float .. py:attribute:: evidence :type: dict[str, Any] .. py:attribute:: error :type: str | None :value: None .. py:method:: to_dict() -> dict[str, Any] Return a JSON-serialisable mapping of the gate result. .. py:class:: FieldTestReport Structured result of a field test: gates, timings, failure statistics. :ivar device: Device class name under test. :ivar address: Instrument address the pipeline was pointed at. :ivar label: Short name for this run, used when several runs are compared. :ivar tier: ``"host"`` or ``"client"`` — which path the run exercised. A module can hold a green host report and fail the first time an experiment talks to it: five of the 2026-08-04 findings live above the device layer. :ivar started_utc: ISO-8601 UTC timestamp of the first gate. :ivar finished_utc: ISO-8601 UTC timestamp after the last gate. :ivar environment: Host and interpreter identification. :ivar identity: Instrument identity string, once known. :ivar gates: Gate results in execution order. :ivar timing_profile: Per command: min / median / p95 / max latency. This is the empirical basis for the module's timeouts and poll intervals. :ivar failure_stats: Per command: attempts and failures by class, so flaky commands are visible instead of anecdotal. :ivar coverage: What the module declares, and how much of it the run exercised. Keyed by the declaration rather than by the calls made, so a parameter nobody touched is visible as untested instead of absent. :ivar findings: What the run found wrong with the *module*, each with the action that fixes it. A gate that failed says which gate; a finding says what to change. .. py:attribute:: device :type: str .. py:attribute:: address :type: str .. py:attribute:: label :type: str :value: '' .. py:attribute:: tier :type: str :value: 'host' .. py:attribute:: started_utc :type: str :value: '' .. py:attribute:: finished_utc :type: str :value: '' .. py:attribute:: environment :type: dict[str, Any] .. py:attribute:: identity :type: str | None :value: None .. py:attribute:: gates :type: list[GateResult] :value: [] .. py:attribute:: timing_profile :type: dict[str, LatencyStats] .. py:attribute:: failure_stats :type: dict[str, FailureStats] .. py:attribute:: coverage :type: plesty.lib.test.report_artifact.Coverage .. py:attribute:: findings :type: list[plesty.lib.test.report_artifact.Finding] :value: [] .. py:property:: ok :type: bool ``True`` when no gate failed. Skipped gates do not fail a run. .. py:property:: counts :type: dict[str, int] Number of gates per status. .. py:property:: errors :type: list[plesty.lib.test.report_artifact.Finding] Findings that make the module wrong, not merely imperfect. A module declaring a parameter the instrument refuses is telling every caller — the rig, a remote client, the manager's API summary — that something exists which does not. That is worth failing a publish over, which is what ``plesty check`` does with this list. .. py:method:: gate(name: str) -> GateResult | None Return the result of gate *name*, or ``None`` if it did not run. :param name: Gate name as recorded in :attr:`gates`. .. py:method:: to_dict() -> dict[str, Any] Return the whole report as a JSON-serialisable mapping. .. py:method:: to_public_document(module: plesty.lib.test.report_artifact.ModuleInfo, instrument: plesty.lib.test.report_artifact.Instrument) -> dict[str, Any] Build the published ``field-test.json`` document. The internal report identifies the rig; this one identifies the *module*. The host name, the instrument address, the identity string and every exception message are left behind — see :mod:`plesty.lib.test.report_artifact` for why each one is a hazard in a world-readable repository. :param module: Module identification, usually :meth:`ModuleInfo.from_project`. :param instrument: The hardware this run was pointed at, as far as it may be published. :returns: The document, already checked by :func:`~plesty.lib.test.report_artifact.validate_document`. .. py:method:: write_public(reports_dir: str | pathlib.Path, module: plesty.lib.test.report_artifact.ModuleInfo, instrument: plesty.lib.test.report_artifact.Instrument) -> pathlib.Path Publish ``//field-test.json``. :param reports_dir: The module's ``reports/`` directory. :param module: Module identification. :param instrument: The hardware this run used. :returns: The path written. :raises ReportSchemaError: If the document fails its own schema. Nothing is written — a malformed artifact in a public repository is not something a later fix takes back. .. py:method:: to_jsonl() -> str Return the report as JSON Lines: one ``run`` record, then one per gate. The line-oriented form appends cleanly across sessions, so successive field tests of the same module stay comparable in one file. .. py:method:: to_markdown() -> str Return a human-readable summary for module docs and reports. .. py:method:: _coverage_line() -> str Summarise how much of the declared surface the run exercised. .. py:method:: _findings_markdown() -> list[str] Render the findings section, or a line saying there are none. Findings come last and are the part to act on: a gate name tells you where to look, a finding tells you what to change. .. py:method:: write(report_path: str | pathlib.Path) -> list[pathlib.Path] Write the report as ``.jsonl`` and ``.md``. :param report_path: Base path. A ``.jsonl`` or ``.md`` suffix is stripped so both files land beside each other. Parent directories are created as needed. :returns: The paths written, JSONL first. .. py:class:: FieldTestPipeline(device_cls: type, *args: Any, address: str | None = None, label: str | None = None, repetitions: int = 20, report_path: str | None = None, instrument: plesty.lib.test.report_artifact.Instrument | None = None, reports_dir: str | pathlib.Path | None = None, archive_dir: str | pathlib.Path | None = None, project_root: str | pathlib.Path = '.', safe_ops: collections.abc.Sequence[str] | None = None, unsafe_ops: collections.abc.Sequence[str] | None = None, op_payloads: dict[str, dict[str, Any]] | None = None, resolve_unsafe_ops: collections.abc.Callable[[Any], collections.abc.Sequence[str]] | None = None, op_kinds: dict[str, plesty.lib.device.device_utils.OpKind | dict[str, Any]] | None = None, ignore_keys: collections.abc.Sequence[str] | None = None, only_keys: collections.abc.Sequence[str] | None = None, param_schema: str | None = None, drain_probe: tuple[str, str] | None = None, slow_op: str | None = None, slow_op_payload: dict[str, Any] | None = None, drain_orphan: collections.abc.Callable[[Any], None] | None = None, provoke_restore_keys: collections.abc.Sequence[str] = (), drain_timeout_ms: int = 1, limit_probe_timeout_ms: int = 500, settle_s: float = 0.0, value_tolerance: float = 0.001, max_failure_rate: float = 0.0, max_retries: int = 0, retry_classes: collections.abc.Sequence[str] = ('wrong_response', 'device_error', 'exception'), expensive_connect: bool = False, stop_on_failure: bool = False, seed: int = 123, **kwargs: Any) Standard on-hardware test pipeline for a PLESTY device API module. Each gate method can be called individually from a hardware-marked pytest test, or the whole sequence can be run with :meth:`run`. Gates raise on failure so a standalone call fails its test; :meth:`run` catches per gate and always returns a complete report. :param device_cls: The device class under test. :param \*args: Positional arguments forwarded to ``device_cls.__init__``. For most modules the first one is the instrument address. :param address: Instrument address recorded in the report. Defaults to the first positional argument. :param label: Short name for this run, used when comparing several runs. :param repetitions: Per-command repeat count for the stability gate. :param report_path: Base path for the JSONL and markdown report. When ``None``, :meth:`run` returns the report without writing it. These two carry the host, the address and the identity string, so they belong in a git-ignored directory. :param instrument: The hardware this run is pointed at, as far as it may be published — model, transport and the interchangeable part fitted. Required to publish ``field-test.json``; without it the run still writes the internal report and says why it published nothing. :param reports_dir: The module's ``reports/`` directory. When given (and an ``instrument`` is known) the run publishes ``//field-test.json`` — the committed, sanitised projection described in :mod:`plesty.lib.test.report_artifact`. :param archive_dir: A shared location every run is also copied to — a lab NAS, typically set from the environment rather than written into the module, since the path names internal infrastructure. The local reports stay authoritative; this second copy is what lets someone read the full markdown report, with the host and the address, without it ever entering git. Each run lands in its own timestamped folder so successive runs stay comparable. A share that is unmounted logs a warning and does not fail the run. :param project_root: Directory holding the module's ``pyproject.toml``, used to name the module in the published document. Defaults to the current working directory. :param safe_ops: Operations the function gate is allowed to call. Operations registered with ``safe=True`` are added automatically. When the resulting set is empty the function gate is skipped — no motion or emission operation ever runs without an explicit opt-in. :param unsafe_ops: Operations removed from the allow-list. :param op_kinds: What each operation does, per name, overriding what the module declared with ``@expose_to_api(kind=...)`` or a schema ``kind``. Values are :class:`~plesty.lib.device.device_utils.OpKind` or the same fields as a mapping. The gates act by kind: a ``motion`` operation is driven from the position its ``position_key`` reports (target = start ± ``step``), alternated in the stability gate and put back where it started; a ``lifecycle`` operation (home, reset) runs once when allow-listed and never on repeat; a ``control`` operation (stop, abort) runs once, at rest, after a motion; a ``configure`` operation is never called — the round-trip gate covers settings. A ``read`` is called as it is. :param resolve_unsafe_ops: Called once with the *connected* device to work out which operations it cannot perform — what an instrument supports often depends on what is plugged into it. Answered from the open session rather than by opening another one: every extra connect/disconnect cycle is another chance to strand a lock or wedge the instrument. :param op_payloads: Explicit payload per operation. Operations without an entry get a payload generated from their ``FuncParam`` metadata. :param ignore_keys: Configuration keys excluded from every parameter gate. :param only_keys: Configuration keys to test, to the exclusion of all others. Exclusion stops scaling somewhere around a few dozen parameters: a spectrometer exposing two hundred and fifty of them would need almost all of them listed as exceptions, and the one that matters is whichever was forgotten. Naming what to touch inverts that — a parameter is left alone unless someone decided otherwise, which is the right default for a device where writing the save directory or the trigger mode has consequences an unattended run cannot judge. :param drain_probe: The two distinguishable parameter keys used by the buffer drain gate: a query on the first is abandoned, then the second is queried. Auto-selected when ``None``. :param drain_op: An operation to abandon instead of a parameter query. A parameter read that returns in two milliseconds cannot be abandoned — the shortest timeout the transport accepts is longer than the call — so the drain gate needs something deliberately slow. An averaged measurement is ideal: it takes as long as you ask it to. :param drain_op_payload: Payload for ``drain_op``, kept separate from ``op_payloads`` so making one call slow does not slow the function and stability gates too. :param drain_orphan: Leave a reply unread on the transport, without racing a timeout. Shortening the timeout is not a lever on every instrument: a USBTMC read measured at 676 ms returned normally inside a 100 ms budget, because the driver does not enforce short timeouts — the same reason the transport needs a watchdog at all. Given a callable that transmits a query and never reads its answer, the orphaned response is created deliberately instead, and ``drain_timeout_ms`` is not used. How to do that is device-specific, so the module supplies it. :param provoke_restore_keys: Parameters read before the provoking gates and written back after them. A slow measurement gets its slowness from a setting, and the gate must not leave that setting behind. The write-back is verified by reading the value again, and a parameter that will not go back fails the gate — a warning was not enough, because a green report is a reason not to look at the bench. A key found already at the value ``slow_op_payload`` would set is restored to its schema default instead of to itself: that is a previous run's leak, and preserving it is how one failed restore became the instrument's permanent state. :param drain_timeout_ms: Transport timeout used to force the abandoned query. :param limit_probe_timeout_ms: Transport timeout while asking for hardware limits. MIN/MAX is optional per parameter in SCPI, so a parameter that does not implement it never answers; a supported limit comes back in milliseconds, so waiting the full command timeout for each unsupported one only makes the gate slow. :param settle_s: Pause between stability repetitions. :param value_tolerance: Relative tolerance when comparing numeric round-trips. :param max_failure_rate: Highest per-command failure rate the stability gate accepts before failing. :param max_retries: How many times a failing gate call is attempted in total. ``0`` (the default) means a single attempt and no retry. Retrying sounds prudent and measured badly. The transport already retries a failed command three times, escalating through a device clear and a session reopen; retrying on top of that turned one unanswerable query into nine attempts and three session teardowns, and left the lab PM100D wedged. On an instrument that is already struggling, another attempt is not a second chance — it is more load on the thing that is failing. Raise it only for a device with a demonstrated transient fault. The stability gate never retries regardless: measuring how often a command fails is its whole purpose. :param retry_classes: Which failure classes are worth retrying. Timeouts are excluded by default: the transport's own recovery ladder already retries a timed-out command three times, escalating through a device clear and a session reopen, so retrying again here makes nine attempts and three teardowns. Against a command the instrument does not implement — an optional SCPI query, a measurement its sensor cannot make — that escalation was observed to end in the watchdog racing the read and locking the resource. The ladder manufactures the wedge it exists to repair. :param expensive_connect: Connecting costs more than the gates that need their own connection are worth. True for a device whose connect starts an application rather than opening a port — LightField takes tens of seconds, and a run that opened four sessions spent almost all of its time launching software. Those gates are skipped with that stated as the reason, and everything else runs on one connection. :param stop_on_failure: Abort the run at the first failed gate instead of carrying on. On real hardware a failure often leaves the instrument in a state where the remaining gates measure nothing. :param seed: RNG seed for generated operation payloads. :param \*\*kwargs: Keyword arguments forwarded to ``device_cls.__init__``. Store the device class, construction arguments and gate settings. .. py:attribute:: device_cls .. py:attribute:: args :value: () .. py:attribute:: kwargs .. py:attribute:: address .. py:attribute:: label .. py:attribute:: repetitions :value: 20 .. py:attribute:: report_path :value: None .. py:attribute:: instrument :value: None .. py:attribute:: reports_dir .. py:attribute:: archive_dir .. py:attribute:: project_root .. py:attribute:: safe_ops .. py:attribute:: unsafe_ops .. py:attribute:: op_payloads .. py:attribute:: resolve_unsafe_ops :value: None .. py:attribute:: op_kinds :type: dict[str, plesty.lib.device.device_utils.OpKind] .. py:attribute:: ignore_keys .. py:attribute:: only_keys .. py:attribute:: param_schema :value: None .. py:attribute:: drain_probe :value: None .. py:attribute:: slow_op :value: None .. py:attribute:: slow_op_payload .. py:attribute:: drain_orphan :value: None .. py:attribute:: provoke_restore_keys :value: () .. py:attribute:: drain_timeout_ms :value: 1 .. py:attribute:: limit_probe_timeout_ms :value: 500 .. py:attribute:: settle_s :value: 0.0 .. py:attribute:: value_tolerance :value: 0.001 .. py:attribute:: max_failure_rate :value: 0.0 .. py:attribute:: max_retries :value: 0 .. py:attribute:: retry_classes :value: ('wrong_response', 'device_error', 'exception') .. py:attribute:: expensive_connect :value: False .. py:attribute:: stop_on_failure :value: False .. py:attribute:: seed :value: 123 .. py:attribute:: report .. py:attribute:: _samples :type: dict[str, list[float]] .. py:attribute:: _first_seen :type: dict[str, Any] .. py:attribute:: _device :type: Any | None :value: None .. py:attribute:: _resolved_unsafe :type: list[str] | None :value: None .. py:attribute:: _kind_exclusions :type: dict[str, str] .. py:attribute:: _motion_in_flight :type: dict[str, Any] | None :value: None .. py:attribute:: _prepared_payload :type: dict[str, Any] | None :value: None .. py:attribute:: started_stamp :type: str :value: '' .. py:attribute:: observations :type: dict[str, dict[str, Any]] .. py:method:: open() -> Any Connect the device and keep the session open for subsequent gates. :returns: The connected device instance. .. py:method:: close() -> None Close a session previously opened with :meth:`open`. .. py:method:: _session() -> collections.abc.Iterator[Any] Yield a connected device, reusing an open session when there is one. .. py:method:: _capture_declarations(device: Any) -> None Record everything the module declares, before any gate narrows it. Coverage is keyed by what the module *declares*, not by what the run happened to call — ``only_keys`` and ``ignore_keys`` decide what is exercised, not what exists. A module declaring 262 parameters of which 11 are verified should say so; the alternative presents a schema as though every entry were equally trustworthy. :param device: The freshly connected device. .. py:method:: _mark_verified(name: str, state: str) -> None Raise how thoroughly *name* was verified, never lower it. The stability gate queries every parameter the round-trip gate already wrote to; without the ranking it would report those as merely queried. :param name: Parameter key or operation name. :param state: One of ``queried``, ``roundtrip``, ``called``. .. py:method:: _measure(name: str, call: collections.abc.Callable[Ellipsis, Any], *args: Any, **kwargs: Any) -> Any Time one device call, retrying a failure up to ``max_retries`` times. A single unlucky reading should not condemn a gate — especially with ``stop_on_failure``, where it would end the whole run. Every attempt is still counted in the failure statistics, so retrying never flatters the numbers: it only changes whether the gate gives up. :param name: Key under which the call is recorded, e.g. ``"query:power"``. :param call: The callable to invoke. :param \*args: Positional arguments for *call*. :param \*\*kwargs: Keyword arguments for *call*. :returns: Whatever *call* returned. :raises BaseException: The last failure, once every attempt is spent. .. py:method:: _measure_once(name: str, call: collections.abc.Callable[Ellipsis, Any], *args: Any, **kwargs: Any) -> Any Time one device call, recording its latency or its failure class. Failed calls contribute an attempt to the failure statistics but no sample to the timing profile: a timeout measures the timeout setting, not the instrument. :param name: Key under which the call is recorded, e.g. ``"query:power"``. :param call: The callable to invoke. :param \*args: Positional arguments for *call*. :param \*\*kwargs: Keyword arguments for *call*. :returns: Whatever *call* returned. .. py:method:: _try(name: str, call: collections.abc.Callable[Ellipsis, Any], *args: Any, **kwargs: Any) -> Any Measure a call, swallowing the exception and returning ``None``. Used by the stability gate, which deliberately does **not** retry: measuring how often a command fails is the entire point, and a retry would report a flaky instrument as a healthy one. :param name: Key under which the call is recorded. :param call: The callable to invoke. :param \*args: Positional arguments for *call*. :param \*\*kwargs: Keyword arguments for *call*. .. py:method:: finalize() -> FieldTestReport Fold the collected samples into the timing profile and derive findings. :returns: The report, with :attr:`FieldTestReport.timing_profile` and :attr:`FieldTestReport.findings` populated. .. py:method:: _derive_findings() -> list[plesty.lib.test.report_artifact.Finding] Turn what the gates observed into what the developer must change. Every finding is built from a *classification* the gate recorded, never from an exception message: the message is the driver's, and publishing it would put a resource string or a lab path in a committed file. :returns: The findings, in the order the gates produced them. .. py:method:: _roundtrip_findings() -> list[plesty.lib.test.report_artifact.Finding] Findings from gate 3: what the module declares but the hardware refuses. A parameter the instrument rejects is a defect in the module, not a coverage state. The module is telling every caller — the rig, a remote client, the manager's API summary — that a parameter exists which does not, so it is an ``error``. .. py:method:: _constraint_findings() -> list[plesty.lib.test.report_artifact.Finding] Findings from gate 4: where the schema envelope and the hardware disagree. A schema wider than the hardware is a finding even when the gate fails, and the finding is the useful half: the gate says which gate, the finding says which parameter and what to do about it. .. py:method:: _gate(name: str) -> collections.abc.Iterator[dict[str, Any]] Record one gate's status, duration and evidence. Yields a mutable evidence dictionary. A :class:`_SkipGate` raised inside the block records the gate as skipped; any other exception records it as failed and is re-raised so a standalone pytest call fails. :param name: Gate name. .. py:method:: _record(name: str, status: str, start: float, evidence: dict[str, Any], error: str | None) -> None Append a gate result to the report and echo it to stdout. .. py:method:: test_discovery() -> None Gate 1: the configured address resolves to a real instrument. Constructs and initialises the device without connecting, then asks its traffic manager to resolve the address. Modules whose transport has no discovery step skip this gate. ``init()`` is not always the cheap, local step the name suggests: on one module it starts an application and claims the instrument. This gate used to leave that device behind, so the next gate's connect was refused by the run's own leftovers — the gate that resolves an address was making the address unusable. .. py:method:: test_connect_lifecycle() -> None Gate 2: connect, identity, error check, disconnect and reconnect. Runs on its own connections so the resource lock is genuinely released and re-acquired. Identity must be a non-empty string and must be identical across the reconnect. .. py:method:: test_param_roundtrip() -> None Gate 3: every writable parameter round-trips on real hardware. Each parameter is queried, the value read back is written again, and the result is queried once more. Writing back the instrument's own value exercises the full write path without changing device state. Read-only parameters are queried only; write-only parameters are skipped. .. py:method:: test_param_constraints() -> None Gate 4: hardware limits against the schema envelope. For every numeric parameter the instrument's own MIN/MAX is compared with the schema range, and for every categorical parameter the instrument's option set is compared with the schema options. A schema envelope *wider* than the hardware fails the gate: it would accept a value the instrument rejects. A narrower envelope is conservative and only reported. .. py:method:: test_functions() -> None Gate 5: every operation on the safe allow-list can be called. Operations are only called when they appear in ``safe_ops`` or were registered with ``safe=True``. Without an allow-list the gate is skipped rather than guessing which operations move a stage or fire a laser. How each is called depends on its kind: a ``motion`` is driven one step from where it is and put back, a ``control`` is issued once at rest after the motions, a ``lifecycle`` once; ``read`` and ``acquire`` are called with their declared payload. .. py:method:: test_buffer_drain() -> None Gate 6: an abandoned query must not answer the next one. A timed-out query can leave its response unread in the transport buffer; the next query then silently consumes the orphaned answer and every reading after it is off by one. The gate abandons a query on one parameter, then queries a second parameter with a distinguishable value: if that second query returns the first parameter's value, the transport has no drain strategy. The gate is skipped when the transport exposes no timeout, or when no pair of parameters currently holds distinguishable values. .. py:method:: test_error_recovery() -> None Gate 7: the device recovers from a forced timeout and a reconnect. Provokes a real failure — a timed-out ``slow_op``, the same mechanism the drain gate uses — then checks that the instrument is still usable in place, that its error channel answers, and that the first command issued after a full disconnect/reconnect succeeds. Runs on its own connection. .. py:method:: test_stability() -> None Gate 8: repeat every queryable parameter and safe op for statistics. Every call is timed and every failure classified, so the report carries the latency distribution and failure rate this module's timeouts should be derived from. Failures do not stop the loop — collecting the rate is the point — but a per-command rate above ``max_failure_rate`` fails the gate at the end. .. py:method:: reset(label: str | None = None) -> None Discard collected results so this pipeline can be run again. Used by :class:`~plesty.lib.test.field_test_concurrency.ConcurrentFieldTest` to run the same pipeline first alone and then under load without mixing the two sets of statistics. :param label: New label for the next run. The current label is kept when omitted. .. py:method:: run(gates: collections.abc.Sequence[str] | None = None, stop_on_failure: bool | None = None) -> FieldTestReport Run the gate sequence and return the completed report. By default a gate failure is recorded and the sequence continues, so one broken gate still yields the timing profile and failure statistics of the others. With *stop_on_failure* the run aborts at the first failure instead, and every gate that did not run is recorded as skipped with the reason — useful on real hardware, where a failure often leaves the instrument in a state that makes later gates measure nothing. Consecutive gates that share a session run on one connection; gates that exercise connect/disconnect open their own. The report is written to ``report_path`` when one was given. :param gates: Gate names to run, from :data:`GATE_NAMES`. Runs every gate when ``None``. :param stop_on_failure: Abort at the first failed gate. Falls back to the value given at construction time. :returns: The finalised :class:`FieldTestReport`. .. py:method:: _publish(report: FieldTestReport) -> pathlib.Path | None Write the committed ``field-test.json``, or say why nothing was written. The run ends by naming every ``error`` finding with its action: the person who just ran it is the person who can fix it, and a finding recorded but never said out loud is a finding nobody acts on. :param report: The finalised report. :returns: The path published, or ``None`` when nothing was. .. py:method:: _archive(written: collections.abc.Sequence[pathlib.Path]) -> pathlib.Path | None Copy this run's reports to the shared archive, if one is configured. The local files stay authoritative; this is a second copy, in a place other people and other tools can read. The markdown is the point of it: that is the form with the host, the address, the instrument's serial and the error messages the driver actually produced, and it is exactly what git refuses and what someone debugging needs. Each run gets its own folder, named for when it started. Overwriting would destroy the comparison the archive exists for — "the p95 doubled since the last run" needs the last run to still be there. A failure here never fails the run. The share is unmounted often enough, and a hardware run that just passed eight gates must not be reported as broken because a copy did not land. :param written: The files this run produced. :returns: The archive folder, or ``None`` when nothing was archived. .. py:method:: _last_gate_failed() -> bool Return ``True`` when the most recently recorded gate failed. .. py:method:: _drain_errors(device: Any, limit: int = 16) -> list[str] Empty the instrument's error queue, returning what was in it. :param device: The connected device. :param limit: Maximum reads, so a device that always reports an error cannot spin here forever. :returns: The errors that were already queued, oldest first. .. py:method:: _probing(device: Any) -> collections.abc.Iterator[None] Ask for optional features cheaply, without provoking wedge recovery. A parameter that does not implement MIN/MAX simply never answers. With the normal command timeout and the recovery ladder active, learning that costs three timeouts and a session reopen per parameter — and the reopen tears down a session that was perfectly healthy. Inside this block a probe fails fast and stays failed. Failing fast leaves a mess, though: every abandoned query is a response the instrument may still be holding, and the next command reads it instead of its own answer. So the block ends with one device clear — one for the whole gate rather than one per failed probe, which is the cost the ladder was imposing. :param device: The connected device. .. py:method:: _drain_transport(manager: Any) -> None Clear whatever abandoned queries left behind, best effort. :param manager: The device's traffic manager. .. py:method:: _observe(key: str, **readings: Any) -> None Record what the instrument reported about one parameter. Gates 3 and 4 already read every value and every hardware limit, so the schema proposal costs no extra device traffic. :param key: Configuration key. :param \*\*readings: Any of ``value``, ``min_value``, ``max_value``, ``options``. ``None`` readings are ignored. .. py:method:: propose_schema_update(param_schema: str | None = None) -> plesty.lib.test.schema_refresh.SchemaProposal Compare the schema against what the instrument reported. Returns a proposal — nothing is written. Applying it requires an explicit confirmation through :meth:`~plesty.lib.test.schema_refresh.SchemaProposal.apply`, because a schema is the module's published contract and a value read during one experiment is not automatically the right thing to ship. :param param_schema: Path to the parameter schema. Falls back to the path given at construction time. :returns: The :class:`~plesty.lib.test.schema_refresh.SchemaProposal`. :raises ValueError: If no schema path is available. .. py:method:: _param_keys(device: Any) -> list[str] Return the configuration keys in scope for parameter gates. :param device: The connected device. .. py:method:: _allowed_ops(device: Any) -> list[str] Return the operations the function and stability gates may call. Combines the explicit ``safe_ops`` allow-list with operations registered as ``safe=True``, then removes ``unsafe_ops``. :param device: The connected device. .. py:method:: _op_kind(device: Any, op_name: str) -> plesty.lib.device.device_utils.OpKind Return what *op_name* does, from the test configuration or the module. The configuration (``op_kinds``) wins wholesale; otherwise the schema registration options or the ``@expose_to_api`` marker. ``target`` defaults to the operation's first required argument. :param device: The connected device. :param op_name: Operation or method name. .. py:method:: _first_required_arg(device: Any, op_name: str) -> str | None :staticmethod: Return the name of the first required argument of *op_name*, if any. .. py:method:: _motion_plan(device: Any, op_name: str, kind: plesty.lib.device.device_utils.OpKind) -> dict[str, Any] Read where a moving part is and build the payloads that move it out and back. :param device: The connected device. :param op_name: The motion operation. :param kind: Its declaration; ``position_key``, ``target`` and ``step`` must all be known — a motion the test cannot place or bound is a configuration error, not something to guess at. :raises FieldTestError: If the declaration is incomplete. .. py:method:: _motion_once(device: Any, op_name: str, kind: plesty.lib.device.device_utils.OpKind) -> dict[str, Any] Move one step out and back, and report both, for the functions gate. :param device: The connected device. :param op_name: The motion operation. :param kind: Its declaration. .. py:method:: _payload_for(device: Any, op_name: str, generator: plesty.lib.sim.data_generator.DataGenerator) -> dict[str, Any] Build the call payload for one operation. :param device: The connected device. :param op_name: Operation name. :param generator: Value generator used for parameters without a default. .. py:method:: _hardware_range(device: Any, key: str) -> tuple[Any, Any] Return the instrument's MIN/MAX for *key*, or ``(None, None)``. :param device: The connected device. :param key: Configuration key. .. py:method:: _hardware_options(device: Any, key: str) -> list[Any] | None Return the instrument's option set for *key*, or ``None``. :param device: The connected device. :param key: Configuration key. .. py:method:: _orphan_and_verify(device: Any, key_a: str, key_b: str, baseline_a: Any, baseline_b: Any, restore: dict[str, Any], evidence: dict[str, Any], orphan: collections.abc.Callable[[Any], None]) -> None Leave a reply unread, then check the next query is not answered with it. Deterministic where the timeout route is not: nothing is raced, the answer is simply never read. :param device: The connected device. :param key_a: The parameter whose reply is orphaned. :param key_b: The parameter queried afterwards. :param baseline_a: ``key_a``'s value, which the stale reply would carry. :param baseline_b: ``key_b``'s value before the orphan. :param restore: Parameters to write back on the way out. :param evidence: The gate's evidence mapping. :param orphan: What leaves the reply unread — the module's own hook, or the transport's when it did not supply one. :raises FieldTestError: If the reply could not be orphaned. :raises WrongResponseError: If the next query is answered with the orphan. .. py:method:: _provoking(manager: Any) -> collections.abc.Iterator[None] Cause a failure on purpose, without the transport fighting back. A gate that abandons a call wants it to fail. The transport wants to rescue it, and both of its mechanisms make things worse here: The recovery ladder reads the timeout as a wedge and escalates — clear, retry, reopen, retry — so the failure this gate depends on is undone, and each retry re-arms a watchdog. The watchdog then fires by design, because the call was made to exceed its timeout, and issues a device clear from its own thread while this one is still inside the read. VISA refuses that with ``VI_ERROR_RSRC_LOCKED`` every time, and the failed clear is left pending at the instrument — which is how this gate hung the lab meter rather than testing it. :param manager: The device's traffic manager. .. py:method:: _call_op(device: Any, op_name: str, payload: dict[str, Any]) -> Any Call an operation, whichever way the module declares it. A schema-registered operation takes a single payload dict; an ``@expose_to_api`` method takes ordinary keyword arguments. :param device: The connected device. :param op_name: Operation or method name. :param payload: Arguments for the call. :returns: Whatever the operation returned. .. py:method:: _restore_params(device: Any, snapshot: dict[str, Any]) -> None Write remembered parameter values back to the instrument. :param device: The connected device. :param snapshot: Values captured before the gate disturbed them. :raises FieldTestError: If any value could not be written back and read back. This fails the gate on purpose. A restore that only warned is how the lab meter was left averaging 2000 samples for several runs: the report was green, so nobody looked at the bench, and every later measurement paid 670 ms for it. .. py:method:: _restore_one(device: Any, key: str, value: Any) -> str | None Write one value back and confirm the instrument took it. A write that returns success has not necessarily landed — the transport was deliberately failed moments earlier, and the first command after that can be swallowed. So success is the read-back, not the write. One retry, because asking twice costs a round trip and getting this wrong costs every measurement in every later run. :param device: The connected device. :param key: Parameter to restore. :param value: Value it held before the gate disturbed it. :returns: What is still wrong, or ``None`` once the instrument confirms it. .. py:method:: _provocation_snapshot(device: Any, evidence: dict[str, Any]) -> dict[str, Any] Decide what a provoking gate must put the instrument back to. Three sources, in decreasing order of authority. **The first value seen this run.** A gate's own entry read happens after the gates before it, so it inherits anything they left behind — ``error_recovery`` runs last and would snapshot ``buffer_drain``'s provocation as if the operator had set it. The first read of the run predates every provocation, and it is the operator's actual working state rather than a schema's idea of it. **The entry read**, when nothing read the key earlier — running one gate on its own is a legitimate way to use this pipeline. **The schema default**, when the value on offer is the one this gate's own provocation would set. The snapshot is written back, so preserving a leak is how one failed restore became permanent: every later run read 2000 and wrote 2000 back. Provocation values are chosen to be extreme, so finding one already in place is evidence of a leak, not a coincidence. :param device: The connected device. :param evidence: The gate's evidence mapping, annotated with where each value came from and with any leak found. :returns: What to write back when the gate is done. .. py:method:: _default_orphan(device: Any, key: str) -> collections.abc.Callable[[Any], None] | None Build the orphan from the device and its transport, or return ``None``. Every module used to hand-write this, and the hand-written ones were all the same two lines: ask the device for the command that queries a parameter, and hand it to the transport to transmit without reading. Neither half is module knowledge — the command comes from the bound solver, the not-reading from the transport — so neither should have been asked for per module. ``None`` when either half is missing: a device with no solver cannot name a command, and a transport that cannot transmit without reading cannot leave the reply behind. The gate then falls back to racing a timeout, and skips saying so if that does not work either. :param device: The connected device. :param key: The parameter whose reply is to be orphaned. :returns: A callable taking the device, or ``None``. .. py:method:: _abandon(device: Any, key: str) -> Any Make the call the drain gate intends to abandon mid-flight. :param device: The connected device. :param key: Parameter to query when no ``drain_op`` is configured. :returns: Whatever the call returned, on the rare occasion it completes. .. py:method:: _prepare_provocation(device: Any) -> dict[str, Any] Return the payload that makes ``slow_op`` slow, reading what it needs first. Called before the transport's timeout is shortened: a motion slow-op is a long move *from where the part is*, and reading that position under the provoking timeout would itself time out. The payload names how far, as a delta — an absolute target already reached would be a no-op and provoke nothing. :param device: The connected device. :raises _SkipGate: If a motion slow-op is not declared well enough to drive. .. py:method:: _restore_motion(device: Any, evidence: dict[str, Any]) -> None Move a motion slow-op's part back to where the provocation found it. Runs with the transport's timeout restored, after the provoking block: the interrupted move may still be in progress, and the module's own motion method is what knows how to wait for it. :param device: The connected device. :param evidence: The gate's evidence mapping, told what was put back. .. py:method:: _resolve_drain_probe(device: Any) -> tuple[str, str] Return the two parameter keys used by the buffer drain gate. :param device: The connected device. :raises _SkipGate: When fewer than two queryable parameters exist. .. py:method:: _recovery_key(device: Any) -> str Return the parameter key the error-recovery gate probes with. :param device: The connected device. :raises _SkipGate: When the device has no queryable parameter. .. py:method:: _values_match(expected: Any, actual: Any) -> bool Return ``True`` when *actual* is the same value as *expected*. Numeric values are compared with the pipeline's relative tolerance so that an instrument rounding a written value does not count as a mismatch. :param expected: The value written or read before. :param actual: The value read back. .. py:function:: _summarise(report: FieldTestReport) -> str Render the end-of-run summary: what failed, what skipped, what passed. Named rather than counted. A run that says "1 failed" sends the reader back through several hundred lines of transport logging to find out which gate it was, when the name and the reason are already recorded here. :param report: The finalised report. :returns: The summary block, failures first. .. py:function:: _run_gate(gate: collections.abc.Callable[[], None]) -> None Call one gate, letting :meth:`FieldTestPipeline._gate` record the outcome. .. py:function:: utc_now() -> str Return the current UTC time as an ISO-8601 string with second precision. .. py:function:: environment_info() -> dict[str, Any] Return host, interpreter and library identification for the report. The library version matters as much as the host. A report that reproduces a symptom already fixed usually means the fix is not installed, and without the version in the report there is no way to tell that from the fix not working — which costs a great deal more than recording it does. .. py:function:: _exposed_methods(device: Any) -> set[str] Return the names of the device's ``@expose_to_api`` methods. :param device: The device to inspect. .. py:function:: _release(device: Any) -> None Undo an ``init()`` that a gate performed but never connected. Best effort by design: this runs while a gate is already failing or skipping, and a cleanup that raises would replace the real reason with its own. :param device: The device to release. .. py:function:: _traffic_manager(device: Any) -> plesty.lib.traffic.TrafficManager | None Return the device's traffic manager, if it has one. :param device: A device instance, initialised or connected. .. py:function:: _as_number(value: Any) -> float | None Coerce a raw instrument response to a float, or ``None`` when it is not one. .. py:function:: _compare_range(param: Any, hw_min: float | None, hw_max: float | None) -> str Compare a schema range against the instrument's own limits. :param param: The :class:`~plesty.lib.device.params.ConfigParameter`. :param hw_min: Instrument minimum, or ``None`` when unavailable. :param hw_max: Instrument maximum, or ``None`` when unavailable. :returns: ``"schema_wider"``, ``"schema_narrower"``, ``"ok"``, ``"unbounded"`` or ``"unavailable"``. .. py:function:: _values_equal(left: Any, right: Any) -> bool Return ``True`` when two raw readings are the same value. .. py:function:: _distinguishable(left: Any, right: Any) -> bool Return ``True`` when two readings differ enough to tell a stale one apart.