From c953665f3a1779550e5ad0e50dc481affb7dbfa8 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 26 Aug 2026 08:23:54 -0500 Subject: [PATCH] Check a beamline descriptor's declared addresses against the live control system A descriptor names control addresses that were reverse-engineered from a controls config or a design document, and nothing has ever compared them to the running control system. The two sides are independently maintained, so a disagreement is real evidence rather than a restatement: this is the property project_independent_check_principle asks for, and re-reading the YAML cannot supply it. The drift it catches is the class that has already shipped three times on the capture-watch PV set: an address that moved, got renamed by an IOC rebuild, or reads as a wire shape its author did not expect. Named preflight, not probe, and address, not handle, to stay inside the house vocabulary. A probe in CORA is a periodic re-affirmation read that writes a coverage-trail row (CaptureProbe, the three *_probe_tick_seconds settings); this is a one-shot operator sweep that records nothing, and it is the same act capture_watch_preflight performs. An address is what ControlPort.read takes and what NoAdapterForAddressError reports, while handle already means an opaque id (TransferHandle) and is the descriptor's own word for the unrelated epics_handle provenance field, which NOT_ADDRESS_FIELDS excludes. Read-only, and more strictly than the deployment: build_control_port takes writes_enabled=False as a literal rather than inheriting Settings.control_writes_enabled, because this command has no write path at any setting. _EXIT_MISMATCH follows record_fidelity_check's 1-means-disagreement contract rather than reusing capture_watch_preflight's _EXIT_PROBLEM, which is the same symbol name for a different value. Ranging the field walk over the whole deployments/ corpus rather than over 2-BM alone found two defects in it before it ran once. Five address-bearing fields (readback_pv, slot_labels_pv, camera_rotation_pv, per_lens_focus_pv, temperature_pv) were never reached, and an address field whose value is a bare list was dropped silently. Both are pinned by a guard that fails on any address-shaped descriptor key classified as neither readable nor deliberately skipped, so the next one is a decision rather than a silent gap. epics_handle is the interesting exclusion: it carries crate provenance, so its values are ranges, globs and hostnames that would read as dead addresses. Three shapes are reported as skipped rather than dropped (a bare device prefix, a confirm: marker, a non-address token like 2-BM's GV1 valve labels), and the pass rate counts only what was read. A descriptor full of confirm: markers must not be able to report a clean sweep having contacted nothing. PyYAML moves from the dev group to a runtime dependency. It was dev-only while the descriptor was read exclusively by scripts/ and the docs build, and this command is run on a deployment host where that group may be absent. Co-Authored-By: Claude Opus 5 (1M context) --- apps/api/pyproject.toml | 15 +- apps/api/src/cora/api/descriptor_preflight.py | 475 ++++++++++++++++++ ...st_descriptor_preflight_against_softioc.py | 141 ++++++ .../unit/api/test_descriptor_preflight.py | 403 +++++++++++++++ apps/api/uv.lock | 4 +- 5 files changed, 1031 insertions(+), 7 deletions(-) create mode 100644 apps/api/src/cora/api/descriptor_preflight.py create mode 100644 apps/api/tests/integration/test_descriptor_preflight_against_softioc.py create mode 100644 apps/api/tests/unit/api/test_descriptor_preflight.py diff --git a/apps/api/pyproject.toml b/apps/api/pyproject.toml index 3935ebf9d3e..06fcac5cb60 100644 --- a/apps/api/pyproject.toml +++ b/apps/api/pyproject.toml @@ -33,6 +33,12 @@ dependencies = [ "opentelemetry-instrumentation-asyncpg>=0.62b1,<0.63", "opentelemetry-exporter-otlp-proto-http>=1.41,<2", "jsonschema-rs>=0.49.6,<1", + # PyYAML: parses a beamline descriptor (deployments//beamline.yaml). + # Runtime rather than dev because `cora.api.descriptor_probe` is an + # operator command run ON a deployment host, where the dev group may not + # be installed. Nothing on the request path imports it: the descriptor is + # a design artifact, and the seed ceremony still reads none of it. + "pyyaml>=6,<7", # Anthropic SDK: used by `AnthropicLLM`, the production implementor of # `cora.infrastructure.ports.LLM`. No other code imports this; # subscribers + tests use the port + `FakeLLM`. Pin <1 to flag a future @@ -228,11 +234,10 @@ dev = [ # Configuration in `[tool.mutmut]` below scopes mutation to pure # decider/evolver modules only. "mutmut>=3.7.0,<4", - # PyYAML: parses the beamline descriptor (deployments//beamline.yaml). - # Used by scripts/beamline_descriptor.py, which the docs build imports and - # the deployment round-trip test dynamic-imports. types-PyYAML supplies the - # type stubs pyright wants for `import yaml`. - "pyyaml>=6,<7", + # types-PyYAML: stubs only, for `import yaml`. The runtime package is a + # top-level dependency (see `dependencies`), not this group: it was dev-only + # while the descriptor was read exclusively by scripts/ and the docs build, + # and stopped being so when `cora.api.descriptor_probe` landed. "types-PyYAML>=6,<7", ] diff --git a/apps/api/src/cora/api/descriptor_preflight.py b/apps/api/src/cora/api/descriptor_preflight.py new file mode 100644 index 00000000000..846c9b1b245 --- /dev/null +++ b/apps/api/src/cora/api/descriptor_preflight.py @@ -0,0 +1,475 @@ +"""Preflight read: check a beamline descriptor's declared channels against +the live control system. + +`python -m cora.api.descriptor_preflight deployments/2-bm/beamline.yaml` walks a +descriptor for every control address it declares, anywhere in the tree (a +device, one of its constituents, or an enclosure), reads each one exactly +once, and reports whether it connects and what shape `ControlPort` sees it +as. `ADDRESS_FIELDS` below is the list of fields that carry one, and +`NOT_ADDRESS_FIELDS` the address-shaped ones deliberately left alone. + +## Why this exists, and what makes it a real check + +A descriptor is hand-authored prose plus addresses, reverse-engineered from a +controls config or a design document. The control system is a separate, +independently-maintained thing. When the descriptor says `2bma:m14` and the +IOC has no such record, the two genuinely disagree, and neither derives from +the other: that is the property [[project_independent_check_principle]] asks +for, and it is why this command is worth more than re-reading the YAML. + +The drift it catches is the one that has already shipped three times on the +capture-watch PV set (see `cora.api.capture_watch_preflight`, "Why this +exists"): an address that moved, got renamed by an IOC rebuild, or reads as a +different wire shape than the descriptor's author assumed. + +## What a green row does NOT mean + +A connected channel proves a record answers to that name. It says nothing +about the claim the descriptor actually makes, which is that the record +drives the named device, on the named axis, in the named Family. A renumbered +motor crate answers on every one of its old names while every one of them now +points at different hardware, and this command reports a clean sweep. + +So: a red row is evidence of drift, a green row is the absence of one narrow +kind of evidence. Confirming device identity stays an operator conversation, +which is what a descriptor's `confirm:` markers are for. + +## Read-only by construction, and more strictly than the deployment + +`build_control_port` is called with `writes_enabled=False` as a literal, NOT +with `Settings.control_writes_enabled`. `capture_watch_preflight` inherits the +deployment's write posture because it is a rehearsal for a path that runs +under it; this command has no write path at any setting, so inheriting a +writable posture would widen its blast radius for no gain. + +## Footprint on the wire, which is not zero + +Channel Access search is broadcast, and an unresolvable address is the noisy +case: it never answers, so the client keeps searching until the read times +out. A descriptor whose addresses are unconfirmed is exactly the input that +generates the most search traffic, and pointing this command at a beamline +whose staff have not agreed to it puts that traffic on their subnet and CORA's +host in their IOC client logs. + +For a descriptor outside the deployment's own beamline, pin +`EPICS_CA_AUTO_ADDR_LIST=NO` and name a specific gateway in +`EPICS_CA_ADDR_LIST` at process start (per `EpicsCaControlPort`'s connection +model, those are read once and never revisited). Every search is then a +unicast to a named target and nothing reaches a subnet you were not invited +onto. +""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Literal, cast + +import yaml + +from cora.infrastructure.config import Settings +from cora.operation.adapters.control_port_config import build_control_port +from cora.operation.ports.control_port import ( + ControlAccessDeniedError, + ControlNotConnectedError, + ControlTimeoutError, + ControlValueCoercionError, + NoAdapterForAddressError, +) + +if TYPE_CHECKING: + from collections.abc import Mapping + + from cora.operation.ports.control_port import ControlPort + +_EXIT_CLEAN = 0 +_EXIT_MISMATCH = 1 + +# The descriptor fields that carry a readable control-system address. +# `Device.pv` and `Enclosure.permit_signal` are declared in the schema +# (scripts/beamline_descriptor.py); the rest are open key-specs living in +# `model_extra`, so the schema cannot enumerate them and this set is the only +# place they are written down. Guarded by +# `test_address_fields_cover_the_descriptor_corpus`, which fails when a +# descriptor introduces an address-shaped key that is in neither this set nor +# `NOT_ADDRESS_FIELDS`. +ADDRESS_FIELDS: frozenset[str] = frozenset( + { + "pv", + "virtual_pv", + "axis_channels", + "permit_signal", + "readback_pv", + "slot_labels_pv", + "camera_rotation_pv", + "per_lens_focus_pv", + "temperature_pv", + } +) + +# Address-SHAPED descriptor keys that are deliberately never read, each for its +# own reason. Listed rather than omitted so the corpus guard can tell a +# considered exclusion from a field nobody has looked at yet. +# +# epics_handle provenance, not an address: the controls section uses it to +# say which crate or IOC serves a device, so its values are +# ranges (`2bmb:m100-m102`), globs (`2bmb:m*`) and hostnames +# (`JenaNV200D`). Several would pass `_shape_of` as addresses +# and fail at read, reporting drift that is not there. +# ip_addresses network endpoints for a device's own controller, not +# Channel Access addresses. Nothing here can read them, and +# reaching for them would be a port scan. +# channels an integer count of a bimorph mirror's electrodes. +NOT_ADDRESS_FIELDS: frozenset[str] = frozenset({"epics_handle", "ip_addresses", "channels"}) + +# A descriptor's address-shaped dict uses this key to say "none yet, ask +# an operator". It carries prose, never an address. +_CONFIRM_KEY = "confirm" + +ChannelShape = Literal["address", "prefix", "unresolved", "opaque"] + + +@dataclass(frozen=True) +class DescriptorChannel: + """One address the descriptor declares, classified before any read. + + `shape` decides whether this entry is readable at all, and the three + non-readable shapes are reported rather than dropped: a silently skipped + entry makes a partial sweep look like a complete one. + + - `address`: an address-shaped token, the only shape that gets read. + - `prefix`: a trailing-colon device prefix (`usxLAX:`). Names a family + of records, is not itself one, so there is nothing to `read`. + - `unresolved`: a `confirm:` marker standing in for an address nobody has + supplied yet. + - `opaque`: a token in an address field that does not look like one + (2-BM's `gate_valves: [GV1, GV2, GV3]` names valves, not records). + Reported so a genuine typo cannot hide behind the same shape. + """ + + location: str + field_name: str + key: str | None + token: str | None + shape: ChannelShape + + @property + def label(self) -> str: + """`field[key]` when the address is one entry in a named-axis map, + bare `field` otherwise.""" + return f"{self.field_name}[{self.key}]" if self.key is not None else self.field_name + + +@dataclass(frozen=True) +class ChannelReading: + """One channel's read outcome. `detail` carries the adapter's own + message on a failure and stays empty on success.""" + + channel: DescriptorChannel + ok: bool + connected: bool + kind: str | None = None + value: object = None + units: str | None = None + element_count: int | None = None + detail: str = "" + + def render(self) -> str: + head = f"{self.channel.token} ({self.channel.location}, {self.channel.label})" + if not self.connected: + return f"BAD {head}: {self.detail}" + shape = self.kind or "?" + if self.element_count is not None: + shape = f"{shape}[{self.element_count}]" + units = f" {self.units}" if self.units else "" + return f"OK {head}: {shape} = {self.value!r}{units}" + + +@dataclass +class PreflightReport: + """A descriptor's full sweep: what was read, and what was not readable. + + `skipped` is carried beside `readings` rather than folded into it because + the two answer different questions. A skipped entry is a fact about the + DESCRIPTOR (it declares no address here), a reading is a fact about the + CONTROL SYSTEM. Counting them together would let a descriptor full of + `confirm:` markers report a high pass rate having contacted almost + nothing, which is the specific way this command could mislead. + """ + + readings: list[ChannelReading] = field(default_factory=list[ChannelReading]) + skipped: list[DescriptorChannel] = field(default_factory=list[DescriptorChannel]) + + @property + def problem(self) -> bool: + return any(not reading.ok for reading in self.readings) + + +def _walk(node: object, location: str, out: list[DescriptorChannel]) -> None: + """Collect address fields from anywhere in the descriptor tree. + + Recursive over the raw YAML rather than over the Pydantic model: the + schema lives in `scripts/`, which is deliberately free of `cora.*` + imports, and the address fields nest (a device's `constituents` are + devices). Structure-blind recursion needs to know only the field names, + which `ADDRESS_FIELDS` pins and a fitness test holds to the schema. + + The two `cast`s carry parsed-YAML shape, which the loader cannot type: + `safe_load` returns `Any`, and narrowing it by `isinstance` alone leaves + the element types unknown. Keys in a descriptor are always strings; + values stay `object` and every use site narrows before touching them. + """ + if isinstance(node, dict): + mapping = cast("dict[str, object]", node) + here = mapping.get("name") + current = here if isinstance(here, str) and here else location + for key, value in mapping.items(): + if key in ADDRESS_FIELDS: + _classify(value, current, key, out) + else: + _walk(value, current, out) + elif isinstance(node, list): + for item in cast("list[object]", node): + _walk(item, location, out) + + +def _classify(value: object, location: str, field_name: str, out: list[DescriptorChannel]) -> None: + """Turn one address field's value into zero or more classified entries. + + Three value shapes occur, and all three must be reached: a bare string + (`pv: "2bma:m44"`), a named-axis map (`pv: {x_in: ..., y_top: ...}`), and + a bare list (`slot_labels_pv: [...]`). The list case is easy to miss + because it is rare, and missing it drops addresses silently, which is the + one failure this command must not have. + """ + if isinstance(value, str): + out.append(_entry(location, field_name, None, value)) + return + if isinstance(value, list): + for item in cast("list[object]", value): + if isinstance(item, str): + out.append(_entry(location, field_name, None, item)) + return + if isinstance(value, dict): + for key, inner in cast("dict[str, object]", value).items(): + if key == _CONFIRM_KEY: + out.append( + DescriptorChannel( + location=location, + field_name=field_name, + key=None, + token=None, + shape="unresolved", + ) + ) + elif isinstance(inner, str): + out.append(_entry(location, field_name, key, inner)) + elif isinstance(inner, list): + for item in cast("list[object]", inner): + if isinstance(item, str): + out.append(_entry(location, field_name, key, item)) + + +def _entry(location: str, field_name: str, key: str | None, token: str) -> DescriptorChannel: + return DescriptorChannel( + location=location, + field_name=field_name, + key=key, + token=token, + shape=_shape_of(token), + ) + + +def _shape_of(token: str) -> ChannelShape: + """Classify an address token by its own text. + + A colon is the discriminator: every EPICS address in the fleet's + descriptors carries at least one, and the tokens that carry none are + device labels sharing an address field (`GV1`), not addresses. This is a + TEXT test, not a reachability one; a well-formed address for a record + that does not exist still classifies `address` and fails at read, which + is the outcome that carries information. + """ + stripped = token.strip() + if not stripped: + return "opaque" + if stripped.endswith(":"): + return "prefix" + return "address" if ":" in stripped else "opaque" + + +def descriptor_channels(raw: Mapping[str, object]) -> list[DescriptorChannel]: + """Every address a parsed descriptor declares, in document order.""" + out: list[DescriptorChannel] = [] + _walk(raw, location="(descriptor)", out=out) + return out + + +def load_descriptor(path: Path) -> dict[str, object]: + """Read a descriptor's raw YAML mapping. + + Deliberately does NOT go through `scripts/beamline_descriptor.load`: this + command reads addresses, and a descriptor whose schema has drifted is + exactly one worth probing rather than refusing. + """ + raw: object = yaml.safe_load(path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise ValueError(f"{path}: top level must be a mapping") + return cast("dict[str, object]", raw) + + +async def preflight_read_channels( + *, + control_port: ControlPort, + channels: list[DescriptorChannel], +) -> PreflightReport: + """Read every `address`-shaped token once, reporting the rest as skipped. + + Deduplicated by token: a descriptor names the same prefix or record from + several devices, and re-reading it would multiply this command's wire + footprint for no extra information. Every occurrence still gets its own + reported row, so a shared address's failure is visible at each site that + depends on it. + """ + report = PreflightReport() + seen: dict[str, ChannelReading] = {} + for channel in channels: + if channel.shape != "address" or channel.token is None: + report.skipped.append(channel) + continue + cached = seen.get(channel.token) + reading = ( + _rebind(cached, channel) + if cached is not None + else await _read_one(control_port, channel) + ) + seen.setdefault(channel.token, reading) + report.readings.append(reading) + return report + + +def _rebind(reading: ChannelReading, channel: DescriptorChannel) -> ChannelReading: + """Re-attach a cached reading to a second declaration site.""" + return ChannelReading( + channel=channel, + ok=reading.ok, + connected=reading.connected, + kind=reading.kind, + value=reading.value, + units=reading.units, + element_count=reading.element_count, + detail=reading.detail, + ) + + +async def _read_one(control_port: ControlPort, channel: DescriptorChannel) -> ChannelReading: + assert channel.token is not None + try: + measurement = await control_port.read(channel.token) + except NoAdapterForAddressError: + return ChannelReading( + channel=channel, + ok=False, + connected=False, + detail="no CONTROL_PORT_ROUTES prefix covers this address", + ) + except (ControlNotConnectedError, ControlTimeoutError, ControlAccessDeniedError) as exc: + return ChannelReading(channel=channel, ok=False, connected=False, detail=str(exc)) + except ControlValueCoercionError as exc: + return ChannelReading( + channel=channel, + ok=False, + connected=True, + detail=f"adapter could not decode the reading: {exc}", + ) + element_count = ( + len(measurement.value) + if measurement.kind == "Array" and hasattr(measurement.value, "__len__") + else None + ) + return ChannelReading( + channel=channel, + ok=True, + connected=True, + kind=measurement.kind, + value=measurement.value, + units=measurement.units, + element_count=element_count, + ) + + +def render_report(report: PreflightReport, *, path: Path) -> list[str]: + """The printed report, as lines, so a test can assert on it without + capturing stdout.""" + lines = [f"descriptor preflight: {path}"] + if not report.readings and not report.skipped: + lines.append(" (descriptor declares no control-system addresses)") + return lines + for reading in report.readings: + lines.append(f" {reading.render()}") + for shape in ("prefix", "unresolved", "opaque"): + group = [entry for entry in report.skipped if entry.shape == shape] + if not group: + continue + lines.append(f" {len(group)} skipped ({shape}):") + for entry in group: + token = entry.token if entry.token is not None else "(no address declared)" + lines.append(f" {token} ({entry.location}, {entry.label})") + ok_count = sum(1 for reading in report.readings if reading.ok) + lines.append( + f"{ok_count}/{len(report.readings)} channels read, {len(report.skipped)} not readable" + ) + return lines + + +def build_parser() -> argparse.ArgumentParser: + """The CLI surface, separate from `main` so tests can invoke it without + building a real `ControlPort`.""" + parser = argparse.ArgumentParser( + prog="python -m cora.api.descriptor_preflight", + description=( + "Read every control-system address a beamline descriptor declares, " + "once, and report whether it connects and what shape CORA's " + "ControlPort sees it as. Read-only; changes nothing, at any " + "CONTROL_WRITES_ENABLED setting. A failed row is evidence the " + "descriptor has drifted from the control system; a clean row " + "proves only that a record answers to that name, never that it " + "drives the device the descriptor names." + ), + ) + parser.add_argument("descriptor", type=Path, help="path to a deployments//beamline.yaml") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + path: Path = args.descriptor + channels = descriptor_channels(load_descriptor(path)) + # writes_enabled=False as a literal, not Settings.control_writes_enabled: + # this command has no write path to enable. See the module docstring. + control_port = build_control_port(Settings().control_port_routes, writes_enabled=False) + + async def _run() -> int: + try: + report = await preflight_read_channels(control_port=control_port, channels=channels) + for line in render_report(report, path=path): + print(line) + return _EXIT_MISMATCH if report.problem else _EXIT_CLEAN + finally: + # ControlPort Protocol does not declare aclose (it's + # adapter-optional); getattr + suppress mirrors + # capture_watch_preflight's own teardown for the same port. + aclose = getattr(control_port, "aclose", None) + if aclose is not None: + with contextlib.suppress(Exception): + await aclose() + + return asyncio.run(_run()) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/apps/api/tests/integration/test_descriptor_preflight_against_softioc.py b/apps/api/tests/integration/test_descriptor_preflight_against_softioc.py new file mode 100644 index 00000000000..a962cc6b65b --- /dev/null +++ b/apps/api/tests/integration/test_descriptor_preflight_against_softioc.py @@ -0,0 +1,141 @@ +"""Integration tests: the descriptor preflight against a real softIOC subprocess. + +The unit tier drives `preflight_read_channels` against a scripted fake, which proves +the reporting logic but cannot prove the command reaches a control system at +all. That gap matters more here than usual: this command's normal output on a +host with no route to the beamline is every row BAD, which is exactly what a +completely broken probe also prints. A green row has to be shown reachable +through the real `EpicsCaControlPort`, or the red ones carry no information. + +Uses the shared `softioc` fixture (`tests/integration/conftest.py`): the +session-scoped `_pin_epics_env` autouse has already locked EPICS env vars to +the per-worker loopback port, so nothing here reaches a network. +""" + +from __future__ import annotations + +import contextlib +from typing import TYPE_CHECKING, Any + +import pytest + +from cora.api.descriptor_preflight import ( + descriptor_channels, + preflight_read_channels, + render_report, +) +from cora.infrastructure.control_port_route import ControlPortRoute +from cora.operation.adapters.control_port_config import build_control_port + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from cora.operation.ports.control_port import ControlPort + + +@pytest.fixture +async def control_port(softioc: str) -> AsyncIterator[ControlPort]: + """A real CA-backed port routed at the softIOC's prefix, read-only, the + same posture `main` builds.""" + port = build_control_port( + [ControlPortRoute(prefix=softioc, substrate="epics_ca", is_simulated=True)], + writes_enabled=False, + ) + try: + yield port + finally: + aclose = getattr(port, "aclose", None) + if aclose is not None: + with contextlib.suppress(Exception): + await aclose() + + +def _descriptor(prefix: str) -> dict[str, Any]: + """A synthetic descriptor exercising every address shape the walk + supports, against records the softIOC actually serves.""" + return { + "enclosures": [{"name": "TEST-A", "permit_signal": f"{prefix}long_value"}], + "optics": { + "stage": "source", + "devices": [ + {"name": "Analog", "pv": f"{prefix}double_value"}, + {"name": "Selector", "pv": f"{prefix}enum_value"}, + {"name": "Trace", "pv": f"{prefix}waveform"}, + { + "name": "Slit", + "pv": {"x_in": f"{prefix}double_value", "y_top": f"{prefix}long_value"}, + }, + {"name": "Lens", "slot_labels_pv": [f"{prefix}string_value"]}, + { + "name": "Tower", + "pv": f"{prefix}long_value", + "constituents": [{"name": "Hexapod", "pv": f"{prefix}cam1:AcquireTime"}], + }, + ], + }, + } + + +async def test_preflight_reads_every_live_address_shape_against_a_real_ioc( + control_port: ControlPort, softioc: str +) -> None: + channels = descriptor_channels(_descriptor(softioc)) + report = await preflight_read_channels(control_port=control_port, channels=channels) + + assert not report.problem, [r.render() for r in report.readings if not r.ok] + assert len(report.readings) == 9 + by_location = {r.channel.location: r for r in report.readings} + assert by_location["Analog"].kind == "Scalar" + assert by_location["Selector"].kind == "Categorical" + assert by_location["Trace"].kind == "Array" + assert by_location["Trace"].element_count is not None + assert by_location["Hexapod"].ok + assert by_location["TEST-A"].channel.field_name == "permit_signal" + + +async def test_preflight_reports_an_address_the_ioc_does_not_serve( + control_port: ControlPort, softioc: str +) -> None: + """The drift case this command exists for: a descriptor naming a record + that is not there. The route matches, so this is a dead address, not a + configuration gap.""" + raw = { + "g": { + "devices": [ + {"name": "Live", "pv": f"{softioc}double_value"}, + {"name": "Ghost", "pv": f"{softioc}no_such_record"}, + ] + } + } + channels = descriptor_channels(raw) + report = await preflight_read_channels(control_port=control_port, channels=channels) + + assert report.problem + outcomes = {r.channel.location: (r.ok, r.connected) for r in report.readings} + assert outcomes == {"Live": (True, True), "Ghost": (False, False)} + text = "\n".join(render_report(report, path=softioc)) # type: ignore[arg-type] + assert "1/2 channels read" in text + + +async def test_preflight_names_an_address_no_configured_route_covers( + control_port: ControlPort, softioc: str +) -> None: + """An address outside every route prefix is a CONTROL_PORT_ROUTES gap, + and must not be reported as a beamline that failed to answer.""" + raw = {"g": {"devices": [{"name": "Elsewhere", "pv": "unrouted:m1"}]}} + channels = descriptor_channels(raw) + report = await preflight_read_channels(control_port=control_port, channels=channels) + assert "CONTROL_PORT_ROUTES" in report.readings[0].detail + + +async def test_preflight_never_writes_to_the_ioc(control_port: ControlPort, softioc: str) -> None: + """Read the value, run a full probe, read it again. The command's whole + safety claim is that it changes nothing, and the claim is worth what a + test of it is worth.""" + address = f"{softioc}double_value" + before = await control_port.read(address) + await preflight_read_channels( + control_port=control_port, channels=descriptor_channels(_descriptor(softioc)) + ) + after = await control_port.read(address) + assert before.value == after.value diff --git a/apps/api/tests/unit/api/test_descriptor_preflight.py b/apps/api/tests/unit/api/test_descriptor_preflight.py new file mode 100644 index 00000000000..5d3efacc49a --- /dev/null +++ b/apps/api/tests/unit/api/test_descriptor_preflight.py @@ -0,0 +1,403 @@ +"""Unit tests for the descriptor preflight command. + +Two halves, and they fail for different reasons. The extraction half is pure +and runs against inline descriptor fragments plus the real +`deployments/*/beamline.yaml` corpus: it fails when an address the descriptor +declares is not reached. The read half drives `preflight_read_channels` against a +scripted fake `ControlPort` (read-only; this command never calls `.write()` +or `.subscribe()`, so the fake implements `.read()` only): it fails when a +substrate outcome is reported as something other than what it was. + +The corpus guard (`test_address_fields_cover_the_descriptor_corpus`) is the +one that earns its place over time. Both of the extraction defects found +while writing this command -- five address fields the walk never reached, and +a bare-list value shape it dropped silently -- were found by ranging over the +real corpus rather than over an example, per +[[project_aggregate_coverage_blindness]]. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING, cast + +import pytest +import yaml + +from cora.api.descriptor_preflight import ( + _EXIT_CLEAN, # pyright: ignore[reportPrivateUsage] + _EXIT_MISMATCH, # pyright: ignore[reportPrivateUsage] + ADDRESS_FIELDS, + NOT_ADDRESS_FIELDS, + DescriptorChannel, + PreflightReport, + build_parser, + descriptor_channels, + load_descriptor, + preflight_read_channels, + render_report, +) +from cora.operation.ports.control_port import ( + ControlAccessDeniedError, + ControlNotConnectedError, + ControlTimeoutError, + ControlValueCoercionError, + Measurement, + NoAdapterForAddressError, +) + +if TYPE_CHECKING: + from collections.abc import Mapping + +_T = datetime(2026, 8, 26, 12, 0, 0, tzinfo=UTC) + +_REPO_ROOT = Path(__file__).resolve().parents[5] +_DEPLOYMENTS = _REPO_ROOT / "deployments" + +# Keys whose name looks like it could carry a control address. Intentionally +# broad: the guard's job is to make a new address-bearing field impossible to +# add without a human classifying it, so over-matching costs one line in +# NOT_ADDRESS_FIELDS and under-matching costs a silently unprobed channel. +_ADDRESS_SHAPED = ("pv", "channel", "signal", "address", "attr", "handle", "tango", "epics") + + +def _reading(value: object, kind: str = "Scalar", units: str | None = None) -> Measurement: + return Measurement( + value=value, + kind=kind, # type: ignore[arg-type] + quality="Good", + produced_at=_T, + units=units, + ) + + +class _FakeControlPort: + """Scripted `read()`-only fake. The probe never writes or subscribes.""" + + def __init__(self, script: dict[str, Measurement | Exception]) -> None: + self._script = script + self.reads: list[str] = [] + + async def read(self, address: str) -> Measurement: + self.reads.append(address) + outcome = self._script[address] + if isinstance(outcome, Exception): + raise outcome + return outcome + + +def _channels(raw: Mapping[str, object]) -> list[DescriptorChannel]: + return descriptor_channels(raw) + + +def _descriptors() -> list[Path]: + return sorted(_DEPLOYMENTS.glob("*/beamline.yaml")) + + +def _corpus_keys(paths: list[Path]) -> set[str]: + """Every mapping key appearing anywhere in the given descriptors.""" + keys: set[str] = set() + + def walk(node: object) -> None: + if isinstance(node, dict): + for key, value in cast("dict[str, object]", node).items(): + keys.add(key) + walk(value) + elif isinstance(node, list): + for item in cast("list[object]", node): + walk(item) + + for path in paths: + walk(yaml.safe_load(path.read_text(encoding="utf-8"))) + return keys + + +# --- extraction ------------------------------------------------------- + + +def test_descriptor_channels_reads_a_bare_string_address() -> None: + found = _channels({"optics": {"devices": [{"name": "Flag", "pv": "2bma:m44"}]}}) + assert [(c.token, c.shape, c.location) for c in found] == [("2bma:m44", "address", "Flag")] + + +def test_descriptor_channels_reads_each_axis_of_a_named_axis_map() -> None: + found = _channels( + {"optics": {"devices": [{"name": "Slit", "pv": {"x_in": "2bma:m14", "y_top": "2bma:m15"}}]}} + ) + assert [(c.key, c.token) for c in found] == [("x_in", "2bma:m14"), ("y_top", "2bma:m15")] + + +def test_descriptor_channels_reads_a_bare_list_address() -> None: + """The shape that was silently dropped: an address field whose value is a + list, with no axis map around it.""" + found = _channels( + {"det": {"devices": [{"name": "Sel", "slot_labels_pv": ["2bm:X:L0", "2bm:X:L1"]}]}} + ) + assert [c.token for c in found] == ["2bm:X:L0", "2bm:X:L1"] + + +def test_descriptor_channels_reads_a_list_nested_in_an_axis_map() -> None: + found = _channels({"g": {"devices": [{"name": "Slit", "pv": {"x": ["2bma:m11", "2bma:m12"]}}]}}) + assert [(c.key, c.token) for c in found] == [("x", "2bma:m11"), ("x", "2bma:m12")] + + +def test_descriptor_channels_descends_into_constituents() -> None: + found = _channels( + { + "g": { + "devices": [ + { + "name": "Tower", + "pv": "2bmb:m24", + "constituents": [{"name": "Hexapod", "pv": "2bmHXP:m1"}], + } + ] + } + } + ) + assert [(c.location, c.token) for c in found] == [ + ("Tower", "2bmb:m24"), + ("Hexapod", "2bmHXP:m1"), + ] + + +def test_descriptor_channels_reads_an_enclosure_permit_signal() -> None: + found = _channels({"enclosures": [{"name": "2-BM-A", "permit_signal": "S02BM-PSS:StaA:Sec"}]}) + assert [(c.location, c.field_name, c.shape) for c in found] == [ + ("2-BM-A", "permit_signal", "address") + ] + + +@pytest.mark.parametrize( + ("token", "shape"), + [ + ("2bma:m14", "address"), + ("2bmb:table3.X", "address"), + ("usxLAX:", "prefix"), + ("2bm:MCTOptics:", "prefix"), + ("GV1", "opaque"), + ("", "opaque"), + (" ", "opaque"), + ], +) +def test_descriptor_channels_classifies_an_address_by_its_own_text(token: str, shape: str) -> None: + found = _channels({"g": {"devices": [{"name": "D", "pv": token}]}}) + assert found[0].shape == shape + + +def test_descriptor_channels_marks_a_confirm_marker_unresolved_carrying_no_address() -> None: + found = _channels({"g": {"devices": [{"name": "Mono", "pv": {"confirm": "MONO-1"}}]}}) + assert [(c.shape, c.token) for c in found] == [("unresolved", None)] + + +def test_descriptor_channels_ignores_a_field_that_is_not_an_address_field() -> None: + found = _channels({"g": {"devices": [{"name": "D", "epics_handle": "2bmb:m*", "note": "x"}]}}) + assert found == [] + + +def test_descriptor_channels_attributes_an_address_to_the_nearest_named_ancestor() -> None: + found = _channels({"grp": {"stage": "sample", "devices": [{"name": "Stage", "pv": "2bmb:m1"}]}}) + assert found[0].location == "Stage" + + +# --- the corpus guard ------------------------------------------------- + + +def test_address_fields_cover_the_descriptor_corpus() -> None: + """Every address-shaped key in every descriptor is classified. + + Ranges over the whole corpus, not one descriptor: a field used only by + i22 is exactly the one a 2-BM-shaped example would miss. A new key here + is not a failure to fix by widening the pattern; it is a question -- + is this a readable channel or not -- answered by putting the key in + ADDRESS_FIELDS or in NOT_ADDRESS_FIELDS with its reason. + """ + descriptors = _descriptors() + assert descriptors, "no descriptors found; the guard would pass vacuously" + keys = _corpus_keys(descriptors) + + address_shaped = {key for key in keys if any(tok in key.lower() for tok in _ADDRESS_SHAPED)} + unclassified = address_shaped - ADDRESS_FIELDS - NOT_ADDRESS_FIELDS + assert not unclassified, ( + f"descriptor key(s) {sorted(unclassified)} look like control addresses but are in neither " + "ADDRESS_FIELDS nor NOT_ADDRESS_FIELDS. Decide which, in cora.api.descriptor_preflight." + ) + + +def test_address_fields_and_not_address_fields_are_disjoint() -> None: + assert not (ADDRESS_FIELDS & NOT_ADDRESS_FIELDS) + + +def test_every_declared_address_field_is_used_by_some_descriptor() -> None: + """An address field nobody declares any more is dead weight in the walk.""" + keys = _corpus_keys(_descriptors()) + assert not (ADDRESS_FIELDS - keys), f"unused address field(s): {sorted(ADDRESS_FIELDS - keys)}" + + +def test_the_pilot_descriptor_yields_readable_channels() -> None: + """2-BM is the live pilot: if the walk reaches nothing there, every green + report this command prints is green because it probed nothing.""" + found = descriptor_channels(load_descriptor(_DEPLOYMENTS / "2-bm" / "beamline.yaml")) + readable = [c for c in found if c.shape == "address"] + assert len(readable) > 50 + assert all(c.token for c in readable) + + +# --- reading ---------------------------------------------------------- + + +async def test_preflight_read_channels_reports_a_healthy_channel_ok() -> None: + channel = _channels({"g": {"devices": [{"name": "D", "pv": "2bma:m1"}]}}) + port = _FakeControlPort({"2bma:m1": _reading(1.5, units="mm")}) + report = await preflight_read_channels(control_port=port, channels=channel) # type: ignore[arg-type] + assert [(r.ok, r.kind, r.value, r.units) for r in report.readings] == [ + (True, "Scalar", 1.5, "mm") + ] + assert not report.problem + + +@pytest.mark.parametrize( + "error", + [ + ControlNotConnectedError("2bma:m1"), + ControlTimeoutError("2bma:m1", 5.0), + ControlAccessDeniedError("2bma:m1"), + ], +) +async def test_preflight_read_channels_reports_an_unreachable_channel_as_disconnected( + error: Exception, +) -> None: + channel = _channels({"g": {"devices": [{"name": "D", "pv": "2bma:m1"}]}}) + port = _FakeControlPort({"2bma:m1": error}) + report = await preflight_read_channels(control_port=port, channels=channel) # type: ignore[arg-type] + reading = report.readings[0] + assert (reading.ok, reading.connected) == (False, False) + assert str(error) in reading.detail + assert report.problem + + +async def test_preflight_read_channels_reports_a_coercion_failure_as_connected_but_bad() -> None: + """The distinction matters: the record exists and answered, so this is a + wire-shape problem, not the descriptor naming something absent.""" + channel = _channels({"g": {"devices": [{"name": "D", "pv": "2bma:m1"}]}}) + port = _FakeControlPort({"2bma:m1": ControlValueCoercionError("2bma:m1", "bytes", "Scalar")}) + report = await preflight_read_channels(control_port=port, channels=channel) # type: ignore[arg-type] + reading = report.readings[0] + assert (reading.ok, reading.connected) == (False, True) + assert "could not decode" in reading.detail + + +async def test_preflight_read_channels_names_a_missing_route_rather_than_a_dead_pv() -> None: + channel = _channels({"g": {"devices": [{"name": "D", "pv": "other:m1"}]}}) + port = _FakeControlPort({"other:m1": NoAdapterForAddressError("other:m1")}) + report = await preflight_read_channels(control_port=port, channels=channel) # type: ignore[arg-type] + assert "CONTROL_PORT_ROUTES" in report.readings[0].detail + + +async def test_preflight_read_channels_carries_an_arrays_element_count() -> None: + channel = _channels({"g": {"devices": [{"name": "D", "pv": "2bma:m1"}]}}) + port = _FakeControlPort({"2bma:m1": _reading((1, 2, 3), kind="Array")}) + report = await preflight_read_channels(control_port=port, channels=channel) # type: ignore[arg-type] + assert report.readings[0].element_count == 3 + + +async def test_preflight_read_channels_skips_every_non_channel_shape_without_reading() -> None: + raw = { + "g": { + "devices": [ + {"name": "A", "pv": "2bma:m1"}, + {"name": "B", "pv": "2bmSP1:"}, + {"name": "C", "pv": {"confirm": "OPT-1"}}, + {"name": "D", "pv": {"gate_valves": ["GV1"]}}, + ] + } + } + port = _FakeControlPort({"2bma:m1": _reading(1.0)}) + report = await preflight_read_channels(control_port=port, channels=_channels(raw)) # type: ignore[arg-type] + assert port.reads == ["2bma:m1"] + assert sorted(c.shape for c in report.skipped) == ["opaque", "prefix", "unresolved"] + + +async def test_preflight_reads_a_repeated_address_once_but_reports_each_site() -> None: + raw = { + "g": { + "devices": [ + {"name": "A", "pv": "2bma:m1"}, + {"name": "B", "pv": "2bma:m1"}, + ] + } + } + port = _FakeControlPort({"2bma:m1": _reading(7.0)}) + report = await preflight_read_channels(control_port=port, channels=_channels(raw)) # type: ignore[arg-type] + assert port.reads == ["2bma:m1"] + assert [r.channel.location for r in report.readings] == ["A", "B"] + assert all(r.ok and r.value == 7.0 for r in report.readings) + + +async def test_preflight_read_channels_propagates_a_cached_failure_to_each_site() -> None: + raw = {"g": {"devices": [{"name": "A", "pv": "x:1"}, {"name": "B", "pv": "x:1"}]}} + port = _FakeControlPort({"x:1": ControlNotConnectedError("down")}) + report = await preflight_read_channels(control_port=port, channels=_channels(raw)) # type: ignore[arg-type] + assert [r.ok for r in report.readings] == [False, False] + assert port.reads == ["x:1"] + + +# --- reporting -------------------------------------------------------- + + +async def test_render_counts_readings_and_skips_separately() -> None: + """A descriptor of pure `confirm:` markers must not report a clean sweep: + the pass rate is over what was READ, and the skips are stated beside it.""" + raw = {"g": {"devices": [{"name": "A", "pv": {"confirm": "Q-1"}}, {"name": "B", "pv": "s:"}]}} + port = _FakeControlPort({}) + report = await preflight_read_channels(control_port=port, channels=_channels(raw)) # type: ignore[arg-type] + lines = render_report(report, path=Path("d.yaml")) + assert "0/0 channels read, 2 not readable" in lines[-1] + assert not report.problem + + +def test_render_report_states_a_descriptor_declared_no_addresses() -> None: + lines = render_report(PreflightReport(), path=Path("d.yaml")) + assert "declares no control-system addresses" in lines[1] + + +async def test_render_groups_skips_by_shape() -> None: + raw = {"g": {"devices": [{"name": "A", "pv": "s:"}, {"name": "B", "pv": {"gv": ["GV1"]}}]}} + port = _FakeControlPort({}) + report = await preflight_read_channels(control_port=port, channels=_channels(raw)) # type: ignore[arg-type] + text = "\n".join(render_report(report, path=Path("d.yaml"))) + assert "1 skipped (prefix):" in text + assert "1 skipped (opaque):" in text + + +async def test_render_labels_a_named_axis_address_with_its_axis() -> None: + raw = {"g": {"devices": [{"name": "Slit", "pv": {"x_in": "2bma:m14"}}]}} + port = _FakeControlPort({"2bma:m14": _reading(1.0)}) + report = await preflight_read_channels(control_port=port, channels=_channels(raw)) # type: ignore[arg-type] + assert "pv[x_in]" in report.readings[0].render() + + +# --- CLI -------------------------------------------------------------- + + +def test_build_parser_requires_a_descriptor_path() -> None: + with pytest.raises(SystemExit): + build_parser().parse_args([]) + + +def test_build_parser_accepts_a_descriptor_path() -> None: + assert build_parser().parse_args(["deployments/2-bm/beamline.yaml"]).descriptor == Path( + "deployments/2-bm/beamline.yaml" + ) + + +def test_load_descriptor_rejects_a_non_mapping(tmp_path: Path) -> None: + path = tmp_path / "d.yaml" + path.write_text("- a\n- b\n", encoding="utf-8") + with pytest.raises(ValueError, match="top level must be a mapping"): + load_descriptor(path) + + +def test_exit_codes_are_distinct() -> None: + assert _EXIT_CLEAN != _EXIT_MISMATCH diff --git a/apps/api/uv.lock b/apps/api/uv.lock index 92d32545b11..feebd1b45d2 100644 --- a/apps/api/uv.lock +++ b/apps/api/uv.lock @@ -351,6 +351,7 @@ dependencies = [ { name = "pydantic" }, { name = "pydantic-settings" }, { name = "pyjwt", extra = ["crypto"] }, + { name = "pyyaml" }, { name = "structlog" }, { name = "uuid-utils" }, { name = "uvicorn", extra = ["standard"] }, @@ -386,7 +387,6 @@ dev = [ { name = "pytest-split" }, { name = "pytest-timeout" }, { name = "pytest-xdist" }, - { name = "pyyaml" }, { name = "ruff" }, { name = "tach" }, { name = "testcontainers" }, @@ -418,6 +418,7 @@ requires-dist = [ { name = "pydantic-settings", specifier = ">=2.14.2,<3" }, { name = "pyjwt", extras = ["crypto"], specifier = ">=2.13.0,<3" }, { name = "pytango", marker = "extra == 'tango'", specifier = ">=10.3.1,<11" }, + { name = "pyyaml", specifier = ">=6,<7" }, { name = "structlog", specifier = ">=26.1.0" }, { name = "uuid-utils", specifier = ">=0.17.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.52.1" }, @@ -446,7 +447,6 @@ dev = [ { name = "pytest-split", specifier = ">=0.11.0,<1" }, { name = "pytest-timeout", specifier = ">=2.4.0,<3" }, { name = "pytest-xdist", specifier = ">=3.8.0,<4" }, - { name = "pyyaml", specifier = ">=6,<7" }, { name = "ruff", specifier = ">=0.16.1,<0.17" }, { name = "tach", specifier = ">=0.35.0" }, { name = "testcontainers", extras = ["postgres"], specifier = ">=4.14.2,<5" },