diff --git a/CLAUDE.md b/CLAUDE.md index 562fbc4..77cbd21 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -112,7 +112,22 @@ Engine → VideoComposer and engine → DmxPlayer are inter-component protocols ## Field notes / gotchas -- **Editing `network_map.xml` or `default_mappings.xml` requires restarting BOTH `cuems-controller-engine` AND `cuems-editor`** (plus `cuems-node-engine` if active) — both daemons load topology **once at startup** via `ConfigManager(load_all=True)` and hold it in memory; there is no per-project-load reload. Edit → `xmllint --noout --schema /etc/cuems/.xsd ` → confirm no project loaded → restart. Verify: editor logs `"number_of_nodes": N`; engine logs correct `Controller IP` + adopted set. +- **Node adoption from the UI (`nodelist_modify`) is a three-hop chain, and the engine hop is new.** browser → editor (WS `:9092`) → engine (`/tmp/editor.ipc`) → nodeconf (`/tmp/nodeconf.ipc`). The engine end lives in `ControllerEngine.nodelist_modify` and is **deliberately not in `command_dict`**: every entry there is invoked as `(value, context)`, a two-argument contract with nowhere to put `modify_action`, so a table-registered handler would send `modify_action=None` and nodeconf would answer `Invalid modify_action: None` to *every* click. The message is bound before the table and dispatched directly. Three more things that bite: + - **A missing nodeconf is the common case** (it ships disabled on most of the fleet), and it is *not* cheap to discover: measured on the rig, with `/tmp/nodeconf.ipc` absent `request_to_nodeconf` does **not** return quickly — it burns the full 15 s IPC timeout and then raises `TimeoutError`. Editor commands are serialized by the engine's single listener, so a second click queues behind the first and can blow the editor's own 25 s timeout, turning a clear refusal into *"Engine did not respond"*. Hence the `os.path.exists(NODECONF_IPC_PATH)` pre-check (instant, actionable refusal — measured 0.00 s) and the explicit 5 s `timeout=` on the call. The `None` return is still handled: `Communicator.send_request` swallows exceptions and returns it on other failures. + - Failures must reply **exactly once and never by raising**: `editor_command_callback`'s `except` also replies, so a raise reaches the operator as `Engine reports error: Command : …` and a handler that replies *and* raises puts two replies on one NNG Rep context. Use `_nodelist_refuse`. + - It is **refused while a project is running or merely loaded**. `_resolve_cluster_state` computes `_required_nodes` once, at `load_project`; a REMOVE accepted mid-load would leave the GO gate waiting on a node the operator just removed. +- **A successful adoption re-reads `network_map.xml` in place** (`_reload_network_map` → `cm.load_network_map()` + `_register_node_osc_handlers()`), so the new node reaches the GO gate without an engine restart. ⚠ Order is load-bearing: `get_nodes_by_adoption()` mutates `online`/`adopted` from `str` to `bool` **in place** and survives only one pass, so the re-registration is valid only on the freshly reloaded dict — never call it as a standalone "shortcut" refresh. If the reload fails the operator is told it is a **partial success** (map changed, this process stale) instead of getting an OK. +- **A project node that is missing or silent WARNS — it never blocks GO.** `_resolve_cluster_state` computes `required = (adopted & alive & project) | {controller}`, so a project node that is not adopted, or adopted but did not pong, is simply **excluded** from the arm gate: the load succeeds, GO is enabled, and that node's cues never fire. That is a deliberate operator decision (a mute cue beats a dead show), and the diagnosis is published rather than left in the journal: + - `missing` = project nodes not in the map at all; `unreachable` = adopted, used, did not answer. The controller's own UUID is excluded from both — it is in `required` unconditionally, and it is not always flagged `adopted` in `network_map.xml`, so including it would raise a false alarm on a healthy load. + - Three surfaces, all carrying the same two lists: pushed at load as `/engine/status/cluster_warning` (a **JSON string**, since OSC cannot carry lists), replayed to every new WebSocket client in `_on_ws_client_connect` (with boot auto-load the browser is normally opened *after* the load), and readable on demand in `get_cluster_status` → the editor's `node_status`. + - **Always sent, empty lists included** — a clean load has to actively erase the previous warning. Cleared on `unload_project` and at the top of `load_project`, so a load that fails early does not leave the previous project's warning on screen. Not cleared by STOP: a stopped project is still loaded, and its missing nodes are still missing. + - `load_id` increments per load so a UI can tell an operator's retry (new id, identical lists) from a reconnect replay (same id). + - ⚠ It is only as good as the project's output mapping: `_collect_project_nodes` derives node UUIDs from `output_name[:36]`, so a project whose outputs are unmapped or malformed yields an empty `project` set and the alert says nothing is wrong. +- **`project_status` has three states, and `loaded` is not `armed`.** `get_project_status` answers `running` (loaded and playing), `loaded` (loaded, not playing) or `none`, with `project_uuid` **and** `project_unix_name` populated in the first two. Until 869dr2kzh it collapsed loaded-but-stopped into `none`, so a stopped project was indistinguishable from no project — while `/engine/status/load` went on naming it, which is the contradiction `cuems-power-bridge`'s field notes recorded as "engine says `project_status=none` but keeps broadcasting a stale loaded". Two traps to keep in mind: + - **`loaded` says nothing about readiness.** The GO gate waits on `armed` (`/engine/status/armed`); a project whose nodes never answer sits in `loaded` for the whole 120 s arm watchdog. `armed` is deliberately *not* mirrored into this reply — one pollable source per fact. + - **The load state is read from two attributes** (`self.script` and the `load` status) that a concurrent load can be mutating. They are ANDed, so a torn read falls to `none`, never to a `loaded` naming a project the engine does not hold. +- **`/engine/status/load` broadcasts the `unix_name`, on purpose — do not "fix" it to a uuid.** The frontend prints this string in the transport bar and matches it against `project.unix_name`, so a uuid there shows the operator a raw uuid and breaks the match. Consumers needing the uuid ask `project_status`, which returns both (that is how 869dr2k2h was closed — additively, not by converting the broadcast). ⚠️ The status is cleared at the **top** of `load_project`, right after `reset_script()`, so it dies with `self.script`: before that, a failed load left the engine holding no project while still broadcasting the previous project's name — the transport bar showed it, and power-bridge's `project_loaded()` believed it, accepting `/setnextcue` and `/gocue` against a project that was gone. The visible cost, accepted: during a successful load the bar reads `—` until the new name arrives, instead of the stale old one. +- **Editing `network_map.xml` or `default_mappings.xml` requires restarting BOTH `cuems-controller-engine` AND `cuems-editor`** (plus `cuems-node-engine` if active) — both daemons load topology **once at startup** via `ConfigManager(load_all=True)` and hold it in memory; there is no per-project-load reload (the one exception is the targeted re-read after `nodelist_modify`, above). Edit → `xmllint --noout --schema /etc/cuems/.xsd ` → confirm no project loaded → restart. Verify: editor logs `"number_of_nodes": N`; engine logs correct `Controller IP` + adopted set. - **`cuems-node-engine` has two non-obvious hard startup blockers** (both hit on formitgo 10.16.1.111): 1. `Requires=jack-alsa-bridges.service`, which runs `zita-j2a -d hw:PCH,0`. If that ALSA device doesn't exist (onboard codec is HDMI-only, or a USB DAC is the real card) zita fails → dependency failure → node-engine never starts. Fix: repoint `-d` to the real card (`/proc/asound/cards`; USB DAC often `hw:HID,0`), `daemon-reload` + `reset-failed` + restart. 2. `set_video_outputs` reads `/run/cuems/display.conf` and dies with `DisplayConfNotFoundError` if there are no `[output:*]` sections with a valid `canvas_region`. A hand-authored override that only sets `resolution_policy=native` (no sections) is tolerated by videocomposer but **breaks node-engine** — keep the `[output:*]` sections too. diff --git a/src/cuemsengine/ControllerEngine.py b/src/cuemsengine/ControllerEngine.py index 7ce48ad..36921de 100644 --- a/src/cuemsengine/ControllerEngine.py +++ b/src/cuemsengine/ControllerEngine.py @@ -4,7 +4,9 @@ # SPDX-FileContributor: Ion Reguera import asyncio +import json import math +import os import threading import time from functools import partial @@ -17,6 +19,15 @@ from .core.BaseEngine import BaseEngine from .core.libmtc import libmtcmaster +# Where cuems-nodeconf binds its NNG responder. Checked before an adopt so a +# disabled daemon is an instant, explanatory refusal instead of a 15 s stall. +NODECONF_IPC_PATH = "/tmp/nodeconf.ipc" +# Adopt/un-adopt is a local XML rewrite; it has no business taking longer. +NODECONF_TIMEOUT_S = 5.0 +# Shortest gap between two real cluster probes served to the UI. Each probe +# pings every adopted node, and the settings panel polls while it is open. +CLUSTER_STATUS_CLAMP_S = 2.0 + class ControllerEngine(BaseEngine): """ @@ -101,6 +112,16 @@ def __init__(self, **kwargs): self._pong_expected: set[str] = set() self._pong_event = threading.Event() self._cluster_lock = threading.Lock() + # What the last load found wrong with the cluster, kept so it can be + # pushed at load time AND replayed to a browser that connects later + # (with boot auto-load the UI is normally opened after the load). + # Initialised here on purpose: _on_ws_client_connect reads it, and a + # client connecting before the first load of a freshly restarted engine + # would otherwise raise AttributeError mid-coroutine and take the rest + # of the late-join dump down with it. _load_id is bumped on every + # resolve so a UI can tell an operator's retry from a reconnect replay. + self._load_diagnosis: dict | None = None + self._load_id: int = 0 # One-shot watchdog: fires N seconds after _resolve_cluster_state if # _armed_nodes still doesn't cover _required_nodes (e.g. a node ponged # alive but then died mid-rsync). Logs an error listing the pending @@ -248,14 +269,30 @@ def _register_osc_command_handlers(self): "/engine/players/*", self._handle_player_osc_message ) - # Register direct player handler for every adopted node in the network - # map. - # UI sends /{node_uuid}//... for both controller and worker - # nodes; - # without per-node registration the WS dispatcher silently drops the - # message and the NNG forward never happens. - # The set deduplicates so the controller's own UUID isn't registered - # twice (it appears in both node_conf and network_map['node_list']). + self._register_node_osc_handlers() + + Logger.info("OSC command handlers registered for WebSocket receiving") + + def _register_node_osc_handlers(self) -> None: + """Register the direct player OSC route for every adopted node. + + UI sends /{node_uuid}//... for both controller and worker nodes; + without per-node registration the WS dispatcher silently drops the + message and the NNG forward never happens. The set deduplicates so the + controller's own UUID isn't registered twice (it appears in both + node_conf and network_map['node_list']). + + Called at startup and again after a nodelist_modify, so a node adopted + from the UI gets its route without an engine restart. Re-registering is + safe: register_osc_handler is dict-keyed, so a repeat overwrites. + + ⚠ get_nodes_by_adoption() mutates online/adopted from str to bool IN + PLACE, so it only survives one pass over a given dict. Call this either + at startup or immediately after ConfigManager.load_network_map() has + installed a fresh dict — see _reload_network_map(). A REMOVE leaves the + old route registered on purpose: traffic from a non-adopted node is + already filtered by _adopted_nodes at the callbacks. + """ node_uuids: set[str] = set() own_uuid = ( self.cm.node_conf.get("uuid", "") if hasattr(self, "cm") and self.cm else "" @@ -278,8 +315,6 @@ def _register_osc_command_handlers(self): ) Logger.info(f"Registered direct player OSC handler for /{nuuid}/*") - Logger.info("OSC command handlers registered for WebSocket receiving") - def _handle_direct_player_osc_message(self, address: str, args: list): """Handle direct player OSC messages from UI (///...). @@ -703,7 +738,10 @@ def editor_command_callback(self, item: dict, context): try: self.handle_editor_command( - action=item["action"], value=item["value"], context=context + action=item["action"], + value=item["value"], + context=context, + modify_action=item.get("modify_action"), ) except Exception as e: Logger.error(f"{type(e)} handling editor command: {e}") @@ -716,7 +754,24 @@ def editor_command_callback(self, item: dict, context): request_uuid=request_uuid, ) - def handle_editor_command(self, action, value, context=None): + def handle_editor_command(self, action, value, context=None, modify_action=None): + # nodelist_modify cannot ride the dispatch table below: every entry is + # invoked as command_dict[action](value, context) — a two-argument + # contract with nowhere to put modify_action. Registering it there sends + # modify_action=None to nodeconf, which answers + # {'OK': False, 'error': 'Invalid modify_action: None'} to EVERY click. + # So bind the message here and dispatch it directly. + if action == "nodelist_modify": + message = { + "action": action, + "value": value, + "modify_action": modify_action, + } + if self.nodelist_modify(message, context): + self.confirm_to_editor(context, type=action, value="OK") + self.set_editor_request("") + return + command_dict = { "project_deploy": partial(self.load_project, deploy_only=True), "project_ready": self.load_project, @@ -725,6 +780,7 @@ def handle_editor_command(self, action, value, context=None): "go_script": self.go_script, "project_status": self.get_project_status, "project_unload": self.unload_project, + "cluster_status": self.get_cluster_status, } if action in command_dict.keys(): result = command_dict[action](value, context) @@ -802,6 +858,157 @@ def nodeconf(self, message: dict, context=None) -> bool: Logger.error(f"{type(e)} sending nodeconf request: {e}") return False + def _nodelist_refuse(self, context, msg: str) -> bool: + """Refuse a nodelist_modify with one clean sentence for the operator. + + Every failure path in nodelist_modify goes through here and returns + False. It must NOT raise: editor_command_callback's except clause also + replies (prefixing "Command : ..."), and the + editor wraps that again as "Engine reports error: ...". Raising after + replying would also put two replies on a single NNG Rep context. + + Clears the pending editor request the same way the generic error path + does (see editor_command_callback): capture the uuid, clear it, then + reply with it explicitly, so no stale uuid is left behind for whatever + reads get_editor_request() next. + """ + Logger.error(f"nodelist_modify refused: {msg}") + request_uuid = self.get_editor_request() + self.set_editor_request("") + self.error_to_editor( + context, value=msg, request_uuid=request_uuid, action="nodelist_modify" + ) + return False + + def nodelist_modify(self, message: dict, context=None) -> bool: + """Adopt or un-adopt a node, then refresh our cached topology. + + Forwards {'action','value','modify_action'} to cuems-nodeconf over + /tmp/nodeconf.ipc, which flips in network_map.xml. On success + we re-read that file, because our own copy was loaded once at startup + (ConfigManager(load_all=True)) and would otherwise keep the GO gate + blind to the change until an engine restart. + + Returns True only when nodeconf confirmed AND our reload succeeded; the + caller confirms "OK" to the editor on True. Every False path has + already sent an explanatory error. + """ + node_uuid = message.get("value") + modify_action = message.get("modify_action") + Logger.info(f"nodelist_modify: {modify_action} node {node_uuid}") + + # Adoption is a setup-time operation. Refuse (never stop anything) when + # a project is running, and also when one is merely loaded: the GO + # gate's _required_nodes snapshot is computed once, in load_project, so + # a REMOVE accepted now would leave it waiting on a node the operator + # just removed until the next load. + if self.get_status("running") == "yes": + return self._nodelist_refuse( + context, + "Cannot modify the node list while a project is running. " + "Stop playback first.", + ) + if self.get_status("load"): + return self._nodelist_refuse( + context, + "Cannot modify the node list while a project is loaded. " + "Unload the project first.", + ) + + if not node_uuid or not isinstance(node_uuid, str): + return self._nodelist_refuse( + context, f"nodelist_modify needs a node uuid, got {node_uuid!r}" + ) + if modify_action not in ("ADD", "REMOVE"): + return self._nodelist_refuse( + context, + f"Invalid modify_action: {modify_action!r}. Must be 'ADD' or 'REMOVE'", + ) + + # Fail fast when the daemon is not there at all. Measured on the rig: + # with the socket absent, request_to_nodeconf does NOT return quickly — + # it burns the full 15 s IPC timeout and then raises. Editor commands + # are serialized by the engine's single listener, so a second click + # queues behind the first and can blow the editor's own 25 s timeout, + # turning a clear refusal into "Engine did not respond". nodeconf ships + # disabled on most of the fleet, so this is the common path, not a rare + # one. + if not os.path.exists(NODECONF_IPC_PATH): + return self._nodelist_refuse( + context, + "The node configuration service (cuems-nodeconf) is not " + "running on this controller, so the node list cannot be " + "changed. Enable it with: systemctl enable --now cuems-nodeconf", + ) + + try: + reply = self.communications_thread.request_to_nodeconf( + message, timeout=NODECONF_TIMEOUT_S + ) + except Exception as e: + return self._nodelist_refuse( + context, f"Could not reach cuems-nodeconf: {type(e).__name__}: {e}" + ) + + # Communicator.send_request swallows every exception and returns None — + # an absent socket (nodeconf disabled, which is the fleet default) and a + # peer that never answers both land here. + if reply is None: + return self._nodelist_refuse( + context, + "The node configuration service (cuems-nodeconf) is not " + "responding. Is it enabled on this controller?", + ) + if not isinstance(reply, dict): + return self._nodelist_refuse( + context, f"Unexpected reply from cuems-nodeconf: {reply!r}" + ) + if not reply.get("OK", False): + return self._nodelist_refuse( + context, reply.get("error", "cuems-nodeconf reported an error") + ) + + Logger.info( + f"nodelist_modify: cuems-nodeconf confirmed {modify_action} " + f"for node {node_uuid}" + ) + + # nodeconf has already rewritten network_map.xml. If we cannot re-read + # it the operation is a PARTIAL success: the map (and so the editor and + # the UI) is correct, only this process is behind. Say exactly that + # instead of confirming OK over a stale topology. + try: + self._reload_network_map() + except Exception as e: + Logger.error(f"{type(e).__name__} reloading network_map: {e}") + return self._nodelist_refuse( + context, + f"Node {modify_action.lower()}ed, but the engine could not " + f"reload the topology ({type(e).__name__}: {e}) — restart " + f"cuems-controller-engine before loading a project.", + ) + return True + + def _reload_network_map(self) -> None: + """Re-read network_map.xml and re-register per-node OSC routes. + + Targeted on purpose: ConfigManager.load_network_map() reassigns + self.cm.network_map to a NEW dict, so a concurrent reader sees either + the old one or the new one, never a torn one. A full + ConfigManager(load_all=True) would also re-read settings and mappings — + that is the "restart both daemons" trap, not what we want here. + + ORDER MATTERS: _register_node_osc_handlers() calls + NetworkMap.get_nodes_by_adoption(), which mutates online/adopted from + str to bool IN PLACE and therefore survives only one pass over a given + dict (same hazard _adopted_uuids_from_network_map documents). It is safe + only on the freshly reloaded dict — never call it as a standalone + "shortcut" refresh. + """ + self.cm.load_network_map() + Logger.info("network_map reloaded after a node list change") + self._register_node_osc_handlers() + ######################### # Status Updates (stub - OSCQuery removed) ######################### @@ -977,6 +1184,44 @@ def _broadcast_status(self, key: str, value) -> None: ): self.communications_thread.broadcast_osc(f"/engine/status/{key}", value) + @staticmethod + def _cluster_warning_payload(diagnosis: dict | None) -> str: + """Serialize a load diagnosis for the OSC status channel. + + OSC cannot carry lists, so this travels as a JSON string — the same + thing the codebase already does elsewhere on this channel. A cleared + diagnosis serializes to empty lists rather than being withheld: a clean + load after a bad one has to actively erase the previous warning, or it + stays on the operator's screen forever. + """ + d = diagnosis or {} + return json.dumps( + { + "load_id": d.get("load_id", 0), + "project": d.get("project", ""), + "missing": d.get("missing", []), + "unreachable": d.get("unreachable", []), + } + ) + + def _broadcast_cluster_warning(self, diagnosis: dict | None) -> None: + """Push the load diagnosis to every connected UI. Always sent.""" + self._broadcast_status( + "cluster_warning", self._cluster_warning_payload(diagnosis) + ) + + def _clear_load_diagnosis(self) -> None: + """Forget the last load's diagnosis and erase it from every UI. + + Called wherever a load ENDS — a new load starting, a load failing + early, an unload — but deliberately NOT from _clear_playback_state(), + which also runs on STOP: a stopped project is still loaded, and its + missing nodes are still missing. + """ + with self._cluster_lock: + self._load_diagnosis = None + self._broadcast_cluster_warning(None) + async def _on_ws_client_connect(self, websocket) -> None: """Send full state dump to a newly connected WebSocket client.""" from .osc.WebSocketOscHandler import build_osc_message @@ -989,6 +1234,18 @@ async def _on_ws_client_connect(self, websocket) -> None: if data: await websocket.send(data) + # Last load's cluster diagnosis. Replayed here because with boot + # auto-load the browser is normally opened well after the load, so the + # push in _resolve_cluster_state has nobody listening. + with self._cluster_lock: + diagnosis = dict(self._load_diagnosis) if self._load_diagnosis else None + data = build_osc_message( + "/engine/status/cluster_warning", + self._cluster_warning_payload(diagnosis), + ) + if data: + await websocket.send(data) + # Per-cue playback status for cid, status in self.cue_status.items(): data = build_osc_message(f"/engine/status/cue/{cid}", status) @@ -1079,6 +1336,23 @@ def load_project(self, project_name, context=None, deploy_only=False): self._bump_run_seq("load") self._clear_playback_state() self.reset_script() + # Before anything can fail. load_project_config() and read_script() + # both return False further down, and without this the previous + # project's warning would stay on screen for a project that, to the + # operator, just failed to load. A successful _resolve_cluster_state + # overwrites it a few lines later. + self._clear_load_diagnosis() + # The same argument, applied to the load status. reset_script() drops + # self.script, but nothing dropped `load`, so a failed load left the + # engine holding no project while still broadcasting the PREVIOUS + # project's name on /engine/status/load. That lie reached the transport + # bar and power-bridge alike — project_loaded() stayed True, so + # /setnextcue and /gocue were accepted against a project the engine no + # longer had. The two signals now die together, which is what lets + # get_project_status() trust either of them. + # Deliberate cost: during a *successful* load the transport bar shows + # "—" until the new name arrives, instead of the stale previous one. + self.set_status("load", "") if deploy_only: Logger.info(f"Deploy only requested for {project_name}") @@ -1134,9 +1408,13 @@ def load_project(self, project_name, context=None, deploy_only=False): f"Cue enabled status initialised for {len(self.cue_enabled_status)} cues" ) - # Update internal status - # TODO: send project UUID instead of name for robustness (would break - # UI contract) + # Update internal status. + # This broadcast carries the project's unix_name, NOT its uuid, and + # that is a decision rather than a leftover: the frontend prints this + # string in the transport bar and matches it against project.unix_name, + # so switching it to a uuid shows the operator a raw uuid and breaks + # that match. Consumers that need the uuid ask project_status, which + # returns both (869dr2k2h, closed the additive way). self.set_status("load", project_name) # Per-cue names, deliberately AFTER the load status (unlike cue_status @@ -1319,9 +1597,46 @@ def _adopted_uuids_from_network_map(self) -> set[str]: Logger.warning(f"Could not read network_map: {e}") return out + def _node_label(self, uuid: str) -> str: + """Human-readable name for a node, for LOG LINES ONLY. + + A bare UUID in a journal is unreadable under pressure, so the log says + `node01 (2b6f…)`. Wire payloads stay UUID-only: role_id/alias are + mutable projections per the node-identity contract, and the browser + already resolves them from initial_mappings. Two sources of truth for a + name is exactly what that contract forbids. + """ + try: + node_list = (self.cm.network_map or {}).get("node_list", []) + for entry in node_list: + if not isinstance(entry, dict): + continue + node = entry.get("node") or {} + if node.get("uuid") != uuid: + continue + for field in ("alias", "role_id", "hostname"): + name = node.get(field) + if name: + return f"{name} ({uuid[:8]}…)" + break + except Exception: + pass + return uuid + def _probe_cluster_liveness(self, timeout: float = 1.5) -> set[str]: """Broadcast a ping to all nodes and collect pong replies. + ⚠ NOT re-entrant, and not safe to run twice concurrently: the pong + bookkeeping below (_pong_responses / _pong_expected / _pong_event) is + shared instance state with no per-call correlation id, so two probes in + flight would eat each other's replies. Nothing enforces that — it holds + only because every caller arrives through the engine's single editor + listener, which awaits one command's handler before accepting the next + (see ControllerCommunications.editor_listener). If editor commands are + ever processed concurrently, this needs a real lock or a correlation id + first. get_cluster_status made this reachable from the UI, so the + invariant now matters beyond project loads. + The set of senders that respond within `timeout` is the authoritative "alive right now" view of the cluster. The controller's own UUID is always included (it never needs to ping itself). @@ -1408,25 +1723,47 @@ def _resolve_cluster_state(self) -> None: if in_alive and in_project: continue # the silent happy path — tracked via armed_ready if in_alive and not in_project: - Logger.info(f"node {uuid} online but unused by this project") + Logger.info( + f"node {self._node_label(uuid)} online but unused by this " + f"project" + ) elif not in_alive and in_project: + # Say what the code below actually does. `required` is + # (adopted & alive & project), so this node is EXCLUDED from + # the gate: GO is enabled without it and its cues stay silent. + # The old wording here claimed "GO blocked", which sent anyone + # debugging a mute show from the journal the wrong way. Logger.error( - f"node {uuid} required by this project but did not " - f"respond to ping; cues for it will not play. GO blocked." + f"node {self._node_label(uuid)} is used by this project " + f"but did not respond to ping; its cues will NOT play. " + f"GO is NOT blocked — the node is excluded from the arm " + f"gate." ) else: Logger.warning( - f"node {uuid} is adopted but did not respond to ping; " - f"not required by this project — investigate why it is" - f"offline" + f"node {self._node_label(uuid)} is adopted but did not " + f"respond to ping; not required by this project — " + f"investigate why it is offline" ) + # The two categories worth telling the operator about. Warn loudly, + # never block: `required` below is untouched by any of this. + # + # The controller is excluded from both. It is added to `required` + # unconditionally, so it can never be a node whose cues silently do not + # play — and it is not always flagged `adopted` in network_map.xml, so + # without this it would be reported as "not in the cluster" on a + # perfectly healthy load. A false alarm here is worse than no alarm: + # it teaches operators to ignore the real one. + missing = sorted(project - adopted - {controller_uuid}) + unreachable = sorted((project & adopted) - alive - {controller_uuid}) + # Project nodes that are NOT adopted at all — script is broken for # this cluster. - for uuid in sorted(project - adopted): + for uuid in missing: Logger.warning( - f"project references node {uuid} which is not in the cluster; " - f"cues for it will not fire" + f"project references node {self._node_label(uuid)} which is " + f"not in the cluster; cues for it will not fire" ) required = (adopted & alive & project) | {controller_uuid} @@ -1436,6 +1773,14 @@ def _resolve_cluster_state(self) -> None: self._required_nodes = required self._armed_nodes.clear() self._finished_nodes.clear() + self._load_id += 1 + self._load_diagnosis = { + "load_id": self._load_id, + "project": str(self.get_status("load") or ""), + "missing": missing, + "unreachable": unreachable, + } + diagnosis = dict(self._load_diagnosis) Logger.info( f"Cluster state resolved: required={sorted(required)} " @@ -1443,6 +1788,10 @@ def _resolve_cluster_state(self) -> None: f"project={sorted(project)}" ) + # Tell the UI, even when both lists are empty — see + # _cluster_warning_payload. + self._broadcast_cluster_warning(diagnosis) + # The probe's `alive` set is a runtime liveness snapshot (sub-second, # used here for GO gating). The field in network_map.xml is # a different concept: nodeconf's startup-discovery flag, persisted @@ -1511,13 +1860,111 @@ def stop_script(self, value): return True def get_project_status(self, value, context=None): - """Return current project playback status.""" + """Return the current project load/playback state. + + Three states, deliberately distinct: + + * `running` — a project is loaded AND playing. + * `loaded` — a project is loaded and NOT playing. This is the state + the reply used to collapse into `none`, which made a loaded-but- + stopped project indistinguishable from no project at all — and left + this reply contradicting /engine/status/load, which kept naming the + project all along. + * `none` — no project. + + `loaded` says nothing about readiness: the GO gate waits on `armed`, + published separately on /engine/status/armed. A project whose nodes + never answer sits in `loaded` for the whole 120 s arm watchdog and + never becomes GO-able. Do not treat this field as a GO gate. + + `project_uuid` is populated in both `running` and `loaded`, and the + invariant is one-way: **a status other than `none` always carries a + non-empty uuid**. That is why `running` with no script — which the + guards in load_project/unload_project should make unreachable — + degrades to `none` here rather than to a `running` a client cannot + act on. + + `project_unix_name` is the same string /engine/status/load broadcasts, + returned here so a consumer that needs the uuid never has to make that + broadcast carry one (869dr2k2h). + + The load state is read from two attributes that another thread can be + mutating mid-load. They are ANDed, so a torn read falls to `none` — + never to a `loaded` naming a project the engine does not hold. + """ running = self.get_status("running") == "yes" + loaded = self.script is not None and bool(self.get_status("load")) + + if running: + status = "running" + elif loaded: + status = "loaded" + else: + status = "none" + + if status == "none" or self.script is None: + return {"status": "none", "project_uuid": "", "project_unix_name": ""} + + # unix_name is not a CuemsScript field — load_project attaches it by + # hand — so any script built by another path simply lacks it. return { - "status": "running" if running else "none", - "project_uuid": (str(self.script.id) if running and self.script else ""), + "status": status, + "project_uuid": str(self.script.id), + "project_unix_name": str(getattr(self.script, "unix_name", "") or ""), } + def get_cluster_status(self, value, context=None) -> dict: + """Return who is adopted and who answers right now. + + Two different signals, deliberately kept apart (the UI must never merge + them): + + * `adopted` — from network_map.xml, i.e. what cuems-nodeconf knows. + * `alive` — this engine's sub-second ping/pong, the only thing the GO + gate trusts. `` in the XML is nodeconf's discovery view and + is NOT this. + + `age_s` is how old the probe behind this answer is, computed here: the + browser cannot interpret our monotonic clock. + + Read-only — it never loads, unloads, arms or stops anything, so it is + safe while a show runs. Repeat calls inside CLUSTER_STATUS_CLAMP_S are + served from cache so a polling panel cannot turn into a ping flood. + Always returns a populated dict: the dispatch path confirms only on a + truthy result, and an empty one would leave the editor waiting out its + 25 s timeout. + """ + now = time.monotonic() + cached = getattr(self, "_cluster_status_cache", None) + if cached is not None and (now - cached[0]) < CLUSTER_STATUS_CLAMP_S: + probed_at, payload = cached + else: + alive = self._probe_cluster_liveness() + adopted = self._adopted_uuids_from_network_map() + payload = { + "alive": sorted(alive), + "adopted": sorted(adopted), + "controller": self._controller_uuid(), + } + probed_at = time.monotonic() + self._cluster_status_cache = (probed_at, payload) + + # Read the diagnosis fresh, OUTSIDE the clamp. Folding it into the + # cached payload would serve the pre-transition diagnosis to any poll + # landing within CLUSTER_STATUS_CLAMP_S of a load, unload or reload — + # a warning outliving its cause, which is the worst outcome this + # surface has. The clamp exists to stop ping floods; a lock-guarded + # dict read is not a ping. Same treatment as age_s. + with self._cluster_lock: + diagnosis = dict(self._load_diagnosis) if self._load_diagnosis else {} + + return dict( + payload, + age_s=round(time.monotonic() - probed_at, 3), + missing=diagnosis.get("missing", []), + unreachable=diagnosis.get("unreachable", []), + ) + def unload_project(self, value, context=None): """Unload the current project. Rejects if playback is running.""" if self.get_status("running") == "yes": @@ -1535,6 +1982,7 @@ def unload_project(self, value, context=None): with self._cluster_lock: self._required_nodes.clear() self._adopted_nodes.clear() + self._clear_load_diagnosis() self._forward_command_to_nodes("/engine/command/stop", value) Logger.info("Project unloaded") return True diff --git a/tests/test_cluster_warning.py b/tests/test_cluster_warning.py new file mode 100644 index 0000000..985b876 --- /dev/null +++ b/tests/test_cluster_warning.py @@ -0,0 +1,416 @@ +# SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileContributor: Ion Reguera + +"""Tests for the load-time cluster diagnosis (the "missing node" alert). + +Un-adopt a node — or let it die — then load a project whose cues use its +outputs: the load answers OK, the GO gate turns green, and those cues never +fire. `_resolve_cluster_state` already knew, but only told the journal. + +Two invariants are pinned here, and the second one is the important one: + +* the diagnosis reaches the UI, at load and on a late join, and clears itself + when its cause goes away; +* `required` is byte-identical to what it was before this feature. The operator + decision was *warn loudly, never block GO*, and a plan that quietly changed + the arm gate could turn a mute cue into a dead show. +""" + +import json +from os import environ +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + + +@pytest.fixture(autouse=True) +def set_config_path(): + """Point CUEMS_CONF_PATH at test XML files.""" + test_conf_path = Path(__file__).parent / ".." / "dev" / "test_xml_files" + environ["CUEMS_CONF_PATH"] = str(test_conf_path) + + +@pytest.fixture +def controller(): + """A minimal ControllerEngine with the heavy dependencies mocked out.""" + with ( + patch("cuemsengine.core.BaseEngine.ConfigManager") as MockCM, + patch( + "cuemsengine.core.BaseEngine.BaseEngine.get_controller_ip", + return_value="localhost", + ), + ): + mock_cm_instance = MockCM.return_value + mock_cm_instance.node_conf = { + "uuid": CONTROLLER, + "mtc_port": "MTC_MIDI_PORT", + } + mock_cm_instance.library_path = str( + Path(__file__).parent / ".." / "dev" / "test_xml_files" + ) + mock_cm_instance.tmp_path = "/tmp" + + from cuemsengine.ControllerEngine import ControllerEngine + + engine = ControllerEngine(with_mtc=False) + + engine.communications_thread = Mock() + engine.communications_thread.broadcast_osc = Mock() + engine.communications_thread.nng_hub = Mock() + + engine.set_status("running", "no") + engine.set_status("load", "") + + yield engine + + engine.stop() + + +CONTROLLER = "test-controller-uuid" +NODE1 = "4b9b5a1e-0000-4000-8000-000000000001" +NODE2 = "4b9b5a1e-0000-4000-8000-000000000002" + + +def resolve(controller, *, adopted, alive, project): + """Run _resolve_cluster_state over a made-up cluster shape.""" + controller.script = Mock() + with ( + patch.object( + controller, "_adopted_uuids_from_network_map", return_value=set(adopted) + ), + patch.object(controller, "_probe_cluster_liveness", return_value=set(alive)), + patch.object(controller, "_collect_project_nodes", return_value=set(project)), + patch.object(controller, "_arm_arm_watchdog"), + ): + controller._resolve_cluster_state() + return controller._load_diagnosis + + +def broadcast_warnings(controller): + """Every cluster_warning payload pushed so far, parsed.""" + return [ + json.loads(call[0][1]) + for call in controller.communications_thread.broadcast_osc.call_args_list + if call[0][0] == "/engine/status/cluster_warning" + ] + + +# ─── what the diagnosis says ───────────────────────────────────────────── + + +class TestDiagnosisShapes: + def test_clean_load_says_nothing_is_wrong(self, controller): + d = resolve( + controller, + adopted={CONTROLLER, NODE1}, + alive={CONTROLLER, NODE1}, + project={CONTROLLER, NODE1}, + ) + assert d["missing"] == [] + assert d["unreachable"] == [] + + def test_project_node_not_in_the_cluster_is_missing(self, controller): + """The operator un-adopted it, or it was never adopted.""" + d = resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + assert d["missing"] == [NODE1] + assert d["unreachable"] == [] + + def test_adopted_but_silent_project_node_is_unreachable(self, controller): + """It is in the map and the project uses it — it just did not answer.""" + d = resolve( + controller, + adopted={CONTROLLER, NODE1}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + assert d["missing"] == [] + assert d["unreachable"] == [NODE1] + + def test_both_at_once_are_reported_separately(self, controller): + """Two different fixes — adopt one, go switch the other on.""" + d = resolve( + controller, + adopted={CONTROLLER, NODE1}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1, NODE2}, + ) + assert d["missing"] == [NODE2] + assert d["unreachable"] == [NODE1] + + def test_adopted_node_the_project_does_not_use_is_not_reported(self, controller): + """A spare node being off is not this project's problem.""" + d = resolve( + controller, + adopted={CONTROLLER, NODE1}, + alive={CONTROLLER}, + project={CONTROLLER}, + ) + assert d["missing"] == [] + assert d["unreachable"] == [] + + def test_the_controller_is_never_missing(self, controller): + """Its own UUID is in project and always in required; if it failed to + answer its own probe there are much larger problems than this alert. + """ + d = resolve(controller, adopted=set(), alive=set(), project={CONTROLLER, NODE1}) + assert CONTROLLER not in d["missing"] + assert CONTROLLER not in d["unreachable"] + + def test_the_diagnosis_names_the_project(self, controller): + controller.set_status("load", "test_18") + d = resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + assert d["project"] == "test_18" + + +# ─── the regression guard: GO is not blocked ───────────────────────────── + + +class TestRequiredIsUnchanged: + """`required` must stay exactly (adopted & alive & project) | {controller} + in every shape. This is the whole safety argument of the feature. + """ + + @pytest.mark.parametrize( + "adopted,alive,project", + [ + ({CONTROLLER, NODE1}, {CONTROLLER, NODE1}, {CONTROLLER, NODE1}), + ({CONTROLLER}, {CONTROLLER}, {CONTROLLER, NODE1}), + ({CONTROLLER, NODE1}, {CONTROLLER}, {CONTROLLER, NODE1}), + ({CONTROLLER, NODE1}, {CONTROLLER}, {CONTROLLER, NODE1, NODE2}), + (set(), set(), set()), + ], + ) + def test_required_matches_the_original_formula( + self, controller, adopted, alive, project + ): + resolve(controller, adopted=adopted, alive=alive, project=project) + expected = (adopted & alive & project) | {CONTROLLER} + assert controller._required_nodes == expected + + def test_an_unreachable_project_node_is_excluded_from_the_gate(self, controller): + """The old log line claimed "GO blocked" here. It was false: the node + is dropped from required and GO is enabled without it. + """ + resolve( + controller, + adopted={CONTROLLER, NODE1}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + assert NODE1 not in controller._required_nodes + + +# ─── how it reaches the UI ─────────────────────────────────────────────── + + +class TestPush: + def test_a_clean_load_still_broadcasts_empty_lists(self, controller): + """A clean load after a bad one must erase the previous warning; a + payload withheld would leave it on screen forever. + """ + resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER}, + ) + sent = broadcast_warnings(controller) + assert len(sent) == 1 + assert sent[0]["missing"] == [] + assert sent[0]["unreachable"] == [] + + def test_the_payload_carries_the_lists(self, controller): + resolve( + controller, + adopted={CONTROLLER, NODE1}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1, NODE2}, + ) + sent = broadcast_warnings(controller)[-1] + assert sent["unreachable"] == [NODE1] + assert sent["missing"] == [NODE2] + + def test_unload_clears_it(self, controller): + resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + assert controller._load_diagnosis["missing"] == [NODE1] + + with ( + patch.object(controller, "_clear_playback_state"), + patch.object(controller, "reset_script"), + patch.object(controller, "_forward_command_to_nodes"), + ): + controller.unload_project(None) + + assert controller._load_diagnosis is None + assert broadcast_warnings(controller)[-1]["missing"] == [] + + def test_load_id_increases_so_a_retry_is_not_a_replay(self, controller): + """Content equality cannot tell a reconnect replay from an operator + retrying a load they have not managed to fix: the diagnosis is + byte-identical. The id is what separates them. + """ + first = dict( + resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + ) + second = resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + assert first["missing"] == second["missing"] + assert second["load_id"] > first["load_id"] + + +class TestLateJoin: + def test_readable_before_any_load(self, controller): + """__init__ must define it: a browser connecting before the first load + of a freshly restarted engine reaches _on_ws_client_connect, and an + AttributeError there would also kill the existing late-join dump. + """ + assert controller._load_diagnosis is None + payload = json.loads(controller._cluster_warning_payload(None)) + assert payload == { + "load_id": 0, + "project": "", + "missing": [], + "unreachable": [], + } + + def test_a_client_connecting_after_the_load_still_gets_it(self, controller): + import asyncio + + resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + + ws = Mock() + sent = [] + + async def fake_send(data): + sent.append(data) + + ws.send = fake_send + asyncio.run(controller._on_ws_client_connect(ws)) + + assert any(b"/engine/status/cluster_warning" in d for d in sent) + assert any(NODE1.encode() in d for d in sent) + + +# ─── the pull surface ──────────────────────────────────────────────────── + + +class TestClusterStatusCarriesIt: + def _probe(self, controller, alive, adopted): + return ( + patch.object(controller, "_probe_cluster_liveness", return_value=alive), + patch.object( + controller, "_adopted_uuids_from_network_map", return_value=adopted + ), + ) + + def test_the_two_lists_are_in_the_reply(self, controller): + resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + p1, p2 = self._probe(controller, {CONTROLLER}, {CONTROLLER}) + with p1, p2: + out = controller.get_cluster_status(None) + + assert out["missing"] == [NODE1] + assert out["unreachable"] == [] + + def test_a_change_inside_the_clamp_window_is_served_immediately(self, controller): + """The clamp caches the *probe*, not the diagnosis. Folding the two + together would serve the pre-transition diagnosis for up to two seconds + after a load or unload — a warning outliving its cause, which is the + worst outcome this feature has. + """ + resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + p1, p2 = self._probe(controller, {CONTROLLER}, {CONTROLLER}) + with p1 as mock_probe, p2: + first = controller.get_cluster_status(None) + with ( + patch.object(controller, "_clear_playback_state"), + patch.object(controller, "reset_script"), + patch.object(controller, "_forward_command_to_nodes"), + ): + controller.unload_project(None) + second = controller.get_cluster_status(None) + + assert first["missing"] == [NODE1] + assert second["missing"] == [] # not the cached one + assert mock_probe.call_count == 1 # and still clamped + + def test_still_never_returns_empty(self, controller): + p1, p2 = self._probe(controller, set(), set()) + with p1, p2: + out = controller.get_cluster_status(None) + + assert out + assert out["missing"] == [] + + +# ─── failure paths ─────────────────────────────────────────────────────── + + +class TestFailedLoadLeavesNothingBehind: + def test_a_load_that_fails_before_the_probe_clears_the_old_diagnosis( + self, controller + ): + """load_project_config() and read_script() both return False further + down. Without an early clear, the previous project's warning would sit + on screen for a project that, to the operator, just failed to load. + """ + resolve( + controller, + adopted={CONTROLLER}, + alive={CONTROLLER}, + project={CONTROLLER, NODE1}, + ) + assert controller._load_diagnosis["missing"] == [NODE1] + + controller.cm.load_project_config.side_effect = Exception("no such project") + with ( + patch.object(controller, "_clear_playback_state"), + patch.object(controller, "reset_script"), + patch.object(controller, "error_to_editor"), + ): + result = controller.load_project("ghost_project", context="ctx") + + assert result is False + assert controller._load_diagnosis is None + assert broadcast_warnings(controller)[-1]["missing"] == [] diff --git a/tests/test_controller_commands.py b/tests/test_controller_commands.py index 070484e..08f4943 100644 --- a/tests/test_controller_commands.py +++ b/tests/test_controller_commands.py @@ -151,37 +151,202 @@ def test_stop_returns_true(self, controller): # ─── get_project_status ────────────────────────────────────────────────── +UUID_A = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" + + +def _loaded_script(uuid=UUID_A, unix_name="my_show"): + """A script mock standing in for a loaded project. + + unix_name is always set explicitly: a bare Mock auto-vivifies the + attribute into a truthy Mock, so str(...) would yield a repr instead of + the empty string the code is supposed to produce when it is absent. + """ + script = Mock() + script.id = uuid + script.unix_name = unix_name + return script + + class TestGetProjectStatus: def test_returns_none_when_not_running(self, controller): controller.set_status("running", "no") result = controller.get_project_status(None) - assert result == {"status": "none", "project_uuid": ""} + assert result == { + "status": "none", + "project_uuid": "", + "project_unix_name": "", + } def test_returns_none_when_no_script(self, controller): controller.set_status("running", "no") controller.script = None result = controller.get_project_status(None) - assert result == {"status": "none", "project_uuid": ""} + assert result == { + "status": "none", + "project_uuid": "", + "project_unix_name": "", + } def test_returns_running_with_uuid(self, controller): controller.set_status("running", "yes") - mock_script = Mock() - mock_script.id = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" - controller.script = mock_script + controller.script = _loaded_script() result = controller.get_project_status(None) assert result == { "status": "running", - "project_uuid": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + "project_uuid": UUID_A, + "project_unix_name": "my_show", + } + + def test_loaded_but_not_playing_returns_loaded(self, controller): + """A loaded, stopped project is 'loaded' — not 'none'. + + This is the whole point of 869dr2kzh: before it, this case was + indistinguishable from having no project at all, while + /engine/status/load went on naming the project. + """ + controller.set_status("running", "no") + controller.set_status("load", "my_show") + controller.script = _loaded_script() + result = controller.get_project_status(None) + assert result == { + "status": "loaded", + "project_uuid": UUID_A, + "project_unix_name": "my_show", + } + + def test_script_without_load_status_is_none(self, controller): + """Window 1: mid-load, the script is gone but `load` may lag.""" + controller.set_status("running", "no") + controller.set_status("load", "") + controller.script = _loaded_script() + result = controller.get_project_status(None) + assert result["status"] == "none" + + def test_load_status_without_script_is_none(self, controller): + """Window 2: a failed load leaves no script; never claim 'loaded'.""" + controller.set_status("running", "no") + controller.set_status("load", "my_show") + controller.script = None + result = controller.get_project_status(None) + assert result == { + "status": "none", + "project_uuid": "", + "project_unix_name": "", + } + + def test_running_wins_over_loaded(self, controller): + controller.set_status("running", "yes") + controller.set_status("load", "my_show") + controller.script = _loaded_script() + assert controller.get_project_status(None)["status"] == "running" + + def test_script_without_unix_name_does_not_break_status(self, controller): + """unix_name is bolted on by load_project, not a CuemsScript field.""" + controller.set_status("running", "no") + controller.set_status("load", "my_show") + script = Mock(spec=["id"]) + script.id = UUID_A + controller.script = script + result = controller.get_project_status(None) + assert result == { + "status": "loaded", + "project_uuid": UUID_A, + "project_unix_name": "", + } + + def test_running_without_script_degrades_to_none(self, controller): + """A status other than 'none' must always carry a uuid. + + running=yes with no script should be unreachable — load and unload + both refuse while running — but if it happens, the reply must not + offer a 'running' the client cannot act on. + """ + controller.set_status("running", "yes") + controller.script = None + result = controller.get_project_status(None) + assert result == { + "status": "none", + "project_uuid": "", + "project_unix_name": "", } - def test_loaded_but_not_playing_returns_none(self, controller): - """A loaded but not playing project should report status 'none'.""" + def test_uuid_is_coerced_to_str(self, controller): + """CuemsScript.id is a Uuid, and the editor serialises this to JSON.""" + + class FakeUuid: + def __str__(self): + return UUID_A + controller.set_status("running", "no") - mock_script = Mock() - mock_script.id = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" - controller.script = mock_script + controller.set_status("load", "my_show") + controller.script = _loaded_script(uuid=FakeUuid()) result = controller.get_project_status(None) - assert result == {"status": "none", "project_uuid": ""} + assert result["project_uuid"] == UUID_A + assert isinstance(result["project_uuid"], str) + + +# ─── load_project clears the load status ──────────────────────────────── + + +class TestLoadProjectClearsLoadStatus: + """`load` must die with self.script, not survive it. + + Until 869dr2kzh, load_project dropped the script before validating the + new project but never touched `load`, so a failed load left the engine + holding no project while /engine/status/load still named the previous + one — a lie the transport bar printed and power-bridge's project_loaded() + believed. + """ + + def test_failed_config_load_clears_load_status(self, controller): + controller.set_status("running", "no") + controller.set_status("load", "previous_show") + controller.script = _loaded_script(unix_name="previous_show") + controller.cm.load_project_config.side_effect = Exception("boom") + + with patch.object(controller, "error_to_editor"): + assert controller.load_project("new_show") is False + + assert controller.get_status("load") == "" + assert controller.get_project_status(None)["status"] == "none" + + def test_failed_script_read_clears_load_status(self, controller): + controller.set_status("running", "no") + controller.set_status("load", "previous_show") + controller.script = _loaded_script(unix_name="previous_show") + + with ( + patch.object(controller, "read_script", side_effect=Exception("boom")), + patch.object(controller, "error_to_editor"), + ): + assert controller.load_project("new_show") is False + + assert controller.get_status("load") == "" + assert controller.get_project_status(None)["status"] == "none" + + def test_deploy_only_clears_load_status(self, controller): + """deploy_only drops the script too, so it must drop the name.""" + controller.set_status("running", "no") + controller.set_status("load", "previous_show") + controller.script = _loaded_script(unix_name="previous_show") + + assert controller.load_project("new_show", deploy_only=True) is True + + assert controller.get_status("load") == "" + assert controller.get_project_status(None)["status"] == "none" + + def test_refused_load_while_running_leaves_load_status_alone(self, controller): + """The running guard returns before anything is cleared. + + Never auto-stop a running project — and never quietly forget which + project is playing either. + """ + controller.set_status("running", "yes") + controller.set_status("load", "playing_show") + + assert controller.load_project("new_show") is False + + assert controller.get_status("load") == "playing_show" # ─── unload_project ───────────────────────────────────────────────────── diff --git a/tests/test_nodelist_modify.py b/tests/test_nodelist_modify.py new file mode 100644 index 0000000..e73564f --- /dev/null +++ b/tests/test_nodelist_modify.py @@ -0,0 +1,490 @@ +# SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileContributor: Ion Reguera + +"""Tests for the editor -> engine -> nodeconf adopt/un-adopt path. + +The engine hop for `nodelist_modify` never existed on any live branch (it was +written once and archived), so the button in the UI reached +`raise ValueError("Command nodelist_modify not recognized")`. These tests pin +the behaviour of the re-landed hop, and in particular the two things that made +the archived version wrong: + +* `modify_action` must actually reach nodeconf — the dispatch table calls every + handler as `(value, context)`, so a table-registered handler receives None and + nodeconf answers "Invalid modify_action: None" to every click. +* `request_to_nodeconf` returns None (not an exception) when nodeconf is absent, + which is the fleet default; `.get()` on that raised AttributeError. +""" + +from os import environ +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + + +@pytest.fixture(autouse=True) +def set_config_path(): + """Point CUEMS_CONF_PATH at test XML files.""" + test_conf_path = Path(__file__).parent / ".." / "dev" / "test_xml_files" + environ["CUEMS_CONF_PATH"] = str(test_conf_path) + + +@pytest.fixture(autouse=True) +def nodeconf_socket_present(): + """Most tests exercise the hop past the socket pre-check. + + Points the constant at a path that exists rather than patching + os.path.exists process-wide, which would also lie to engine startup. + """ + with patch("cuemsengine.ControllerEngine.NODECONF_IPC_PATH", "/tmp"): + yield + + +@pytest.fixture +def controller(): + """A minimal ControllerEngine with the heavy dependencies mocked out.""" + with ( + patch("cuemsengine.core.BaseEngine.ConfigManager") as MockCM, + patch( + "cuemsengine.core.BaseEngine.BaseEngine.get_controller_ip", + return_value="localhost", + ), + ): + mock_cm_instance = MockCM.return_value + mock_cm_instance.node_conf = { + "uuid": "test-controller-uuid", + "mtc_port": "MTC_MIDI_PORT", + } + mock_cm_instance.library_path = str( + Path(__file__).parent / ".." / "dev" / "test_xml_files" + ) + mock_cm_instance.tmp_path = "/tmp" + + from cuemsengine.ControllerEngine import ControllerEngine + + engine = ControllerEngine(with_mtc=False) + + engine.communications_thread = Mock() + engine.communications_thread.broadcast_osc = Mock() + engine.communications_thread.nng_hub = Mock() + + # No project loaded, nothing running: the setup-time state in which + # adoption is allowed. + engine.set_status("running", "no") + engine.set_status("load", "") + + yield engine + + engine.stop() + + +NODE = "4b9b5a1e-0000-4000-8000-0123456789ab" + + +def _ok_reply(): + return {"OK": True} + + +# ─── the message that reaches nodeconf ─────────────────────────────────── + + +class TestModifyActionReachesNodeconf: + def test_modify_action_survives_dispatch(self, controller): + """The whole point: modify_action must not be lost on the way down. + + A table-registered handler would be called as (value, context) and send + modify_action=None, which nodeconf rejects for every single click. + """ + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with ( + patch.object(controller, "confirm_to_editor") as mock_confirm, + patch.object(controller, "set_editor_request"), + patch.object(controller, "_reload_network_map"), + ): + controller.handle_editor_command( + "nodelist_modify", NODE, context="ctx", modify_action="ADD" + ) + + sent = controller.communications_thread.request_to_nodeconf.call_args[0][0] + assert sent == { + "action": "nodelist_modify", + "value": NODE, + "modify_action": "ADD", + } + mock_confirm.assert_called_once() + assert mock_confirm.call_args[1]["value"] == "OK" + assert mock_confirm.call_args[1]["type"] == "nodelist_modify" + + def test_success_clears_the_pending_editor_request(self, controller): + """The generic dispatch path clears it after confirming; this one has + to do it for itself, and the archived version forgot to. + """ + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with ( + patch.object(controller, "confirm_to_editor"), + patch.object(controller, "set_editor_request") as mock_clear, + patch.object(controller, "_reload_network_map"), + ): + controller.handle_editor_command( + "nodelist_modify", NODE, context="ctx", modify_action="ADD" + ) + + mock_clear.assert_called_once_with("") + + def test_refusal_clears_the_pending_editor_request(self, controller): + controller.set_status("running", "yes") + with ( + patch.object(controller, "error_to_editor"), + patch.object(controller, "set_editor_request") as mock_clear, + ): + controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + + mock_clear.assert_called_once_with("") + + def test_remove_is_forwarded_too(self, controller): + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with ( + patch.object(controller, "confirm_to_editor"), + patch.object(controller, "set_editor_request"), + patch.object(controller, "_reload_network_map"), + ): + controller.handle_editor_command( + "nodelist_modify", NODE, context="ctx", modify_action="REMOVE" + ) + sent = controller.communications_thread.request_to_nodeconf.call_args[0][0] + assert sent["modify_action"] == "REMOVE" + + def test_callback_forwards_modify_action_from_the_wire(self, controller): + """editor_command_callback must pick modify_action out of the frame.""" + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with ( + patch.object(controller, "confirm_to_editor"), + patch.object(controller, "set_editor_request"), + patch.object(controller, "_reload_network_map"), + ): + controller.editor_command_callback( + { + "action": "nodelist_modify", + "value": NODE, + "modify_action": "ADD", + "action_uuid": "uuid-1", + }, + "ctx", + ) + sent = controller.communications_thread.request_to_nodeconf.call_args[0][0] + assert sent["modify_action"] == "ADD" + + +# ─── failure paths reach the operator as a sentence ────────────────────── + + +class TestFailuresAreLegible: + def test_absent_socket_refuses_immediately(self, controller): + """Measured on the rig: with /tmp/nodeconf.ipc gone, the IPC call does + NOT fail fast — it burns the engine's full 15 s timeout. Editor + commands are serialized, so a second click queues behind the first and + can blow the editor's own 25 s timeout. nodeconf is disabled on most of + the fleet, so this is the common path. + """ + with ( + patch( + "cuemsengine.ControllerEngine.NODECONF_IPC_PATH", + "/tmp/definitely-not-a-nodeconf-socket", + ), + patch.object(controller, "error_to_editor") as mock_error, + ): + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + + assert result is False + controller.communications_thread.request_to_nodeconf.assert_not_called() + assert "not running" in mock_error.call_args[1]["value"] + + def test_the_ipc_call_carries_an_explicit_timeout(self, controller): + """The default is the engine-wide 15 s; an XML rewrite needs seconds.""" + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with patch.object(controller, "_reload_network_map"): + controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + + kwargs = controller.communications_thread.request_to_nodeconf.call_args[1] + assert kwargs["timeout"] == 5.0 + + def test_none_reply_is_an_error_not_a_traceback(self, controller): + """An absent nodeconf socket returns None, not an exception.""" + controller.communications_thread.request_to_nodeconf.return_value = None + with ( + patch.object(controller, "confirm_to_editor") as mock_confirm, + patch.object(controller, "error_to_editor") as mock_error, + ): + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + assert result is False + mock_confirm.assert_not_called() + mock_error.assert_called_once() + assert "not responding" in mock_error.call_args[1]["value"] + assert mock_error.call_args[1]["action"] == "nodelist_modify" + + def test_nodeconf_error_text_is_forwarded_verbatim(self, controller): + controller.communications_thread.request_to_nodeconf.return_value = { + "OK": False, + "error": "Cannot unadopt master node", + } + with patch.object(controller, "error_to_editor") as mock_error: + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "REMOVE"}, + "ctx", + ) + assert result is False + assert mock_error.call_args[1]["value"] == "Cannot unadopt master node" + + def test_transport_exception_is_caught(self, controller): + controller.communications_thread.request_to_nodeconf.side_effect = TimeoutError( + "boom" + ) + with patch.object(controller, "error_to_editor") as mock_error: + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + assert result is False + assert "Could not reach cuems-nodeconf" in mock_error.call_args[1]["value"] + + def test_missing_uuid_is_refused_before_the_hop(self, controller): + with patch.object(controller, "error_to_editor") as mock_error: + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": "", "modify_action": "ADD"}, + "ctx", + ) + assert result is False + controller.communications_thread.request_to_nodeconf.assert_not_called() + assert "node uuid" in mock_error.call_args[1]["value"] + + def test_bad_modify_action_is_refused_before_the_hop(self, controller): + with patch.object(controller, "error_to_editor") as mock_error: + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "FLIP"}, + "ctx", + ) + assert result is False + controller.communications_thread.request_to_nodeconf.assert_not_called() + assert "Invalid modify_action" in mock_error.call_args[1]["value"] + + def test_refusal_never_raises(self, controller): + """A guard must not raise: the callback's except clause replies too, + and the operator would read + "Engine reports error: Command : ...". + """ + controller.set_status("running", "yes") + with patch.object(controller, "error_to_editor"): + # No pytest.raises here on purpose — raising is the defect. + assert ( + controller.nodelist_modify( + { + "action": "nodelist_modify", + "value": NODE, + "modify_action": "ADD", + }, + "ctx", + ) + is False + ) + + +# ─── the guards ────────────────────────────────────────────────────────── + + +class TestProjectStateGuards: + def test_refused_while_running(self, controller): + controller.set_status("running", "yes") + with patch.object(controller, "error_to_editor") as mock_error: + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + assert result is False + controller.communications_thread.request_to_nodeconf.assert_not_called() + assert "while a project is running" in mock_error.call_args[1]["value"] + + def test_refused_while_merely_loaded(self, controller): + """_required_nodes is computed once, at load_project. A REMOVE accepted + now would leave the GO gate waiting on a node just removed. + """ + controller.set_status("running", "no") + controller.set_status("load", "some-project") + with patch.object(controller, "error_to_editor") as mock_error: + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "REMOVE"}, + "ctx", + ) + assert result is False + controller.communications_thread.request_to_nodeconf.assert_not_called() + assert "while a project is loaded" in mock_error.call_args[1]["value"] + + def test_allowed_when_idle(self, controller): + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with ( + patch.object(controller, "error_to_editor") as mock_error, + patch.object(controller, "_reload_network_map"), + ): + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + assert result is True + mock_error.assert_not_called() + + +# ─── the cached topology stops lying ───────────────────────────────────── + + +class TestTopologyReload: + def test_success_reloads_the_map_and_reregisters_routes(self, controller): + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with ( + patch.object(controller, "_register_node_osc_handlers") as mock_reg, + patch.object(controller, "error_to_editor"), + ): + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + assert result is True + controller.cm.load_network_map.assert_called_once() + mock_reg.assert_called_once() + + def test_no_reload_when_nodeconf_refused(self, controller): + controller.communications_thread.request_to_nodeconf.return_value = { + "OK": False, + "error": "node is offline", + } + with patch.object(controller, "error_to_editor"): + controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + controller.cm.load_network_map.assert_not_called() + + def test_partial_success_is_reported_as_an_error(self, controller): + """XML written by nodeconf, reload failed: the UI must not get a green + toast over a stale engine. + """ + controller.communications_thread.request_to_nodeconf.return_value = _ok_reply() + with ( + patch.object( + controller, "_reload_network_map", side_effect=OSError("permission") + ), + patch.object(controller, "confirm_to_editor") as mock_confirm, + patch.object(controller, "error_to_editor") as mock_error, + ): + result = controller.nodelist_modify( + {"action": "nodelist_modify", "value": NODE, "modify_action": "ADD"}, + "ctx", + ) + assert result is False + mock_confirm.assert_not_called() + msg = mock_error.call_args[1]["value"] + assert "could not reload the topology" in msg + assert "restart cuems-controller-engine" in msg + + +# ─── cluster_status (liveness for the UI) ──────────────────────────────── + + +class TestClusterStatus: + def _probe(self, controller, alive, adopted): + return ( + patch.object(controller, "_probe_cluster_liveness", return_value=alive), + patch.object( + controller, "_adopted_uuids_from_network_map", return_value=adopted + ), + ) + + def test_returns_the_expected_keys(self, controller): + """`missing`/`unreachable` were added later by the load-diagnosis work + (see test_cluster_warning.py); they ride on this same reply. + """ + p1, p2 = self._probe(controller, {"c", "n1"}, {"c", "n1", "n2"}) + with p1, p2: + out = controller.get_cluster_status(None) + + assert set(out) == { + "alive", + "adopted", + "controller", + "age_s", + "missing", + "unreachable", + } + assert out["alive"] == ["c", "n1"] + assert out["adopted"] == ["c", "n1", "n2"] + assert isinstance(out["age_s"], float) + + def test_never_returns_empty(self, controller): + """The dispatch path confirms only on a truthy result; an empty dict + would leave the editor waiting out its 25 s timeout. + """ + p1, p2 = self._probe(controller, set(), set()) + with p1, p2: + out = controller.get_cluster_status(None) + + assert out # truthy even with nothing adopted and nothing alive + assert out["alive"] == [] + assert out["adopted"] == [] + + def test_repeat_calls_are_clamped(self, controller): + """A polling settings panel must not become a ping flood.""" + p1, p2 = self._probe(controller, {"c"}, {"c"}) + with p1 as mock_probe, p2: + controller.get_cluster_status(None) + controller.get_cluster_status(None) + controller.get_cluster_status(None) + + assert mock_probe.call_count == 1 + + def test_age_grows_between_clamped_calls(self, controller): + p1, p2 = self._probe(controller, {"c"}, {"c"}) + with p1, p2: + first = controller.get_cluster_status(None) + second = controller.get_cluster_status(None) + + assert second["age_s"] >= first["age_s"] + + def test_reachable_through_the_dispatch_table(self, controller): + p1, p2 = self._probe(controller, {"c"}, {"c"}) + with ( + p1, + p2, + patch.object(controller, "confirm_to_editor") as mock_confirm, + patch.object(controller, "set_editor_request"), + ): + controller.handle_editor_command("cluster_status", None, context="ctx") + + mock_confirm.assert_called_once() + assert mock_confirm.call_args[1]["type"] == "cluster_status" + assert isinstance(mock_confirm.call_args[1]["value"], dict) + + def test_single_box_takes_the_early_return(self, controller): + """formitgo shape: one adopted node, itself. _probe_cluster_liveness + must answer {controller} without pinging anyone. + """ + controller.communications_thread.nng_hub = Mock() + with patch.object( + controller, + "_adopted_uuids_from_network_map", + return_value={controller._controller_uuid()}, + ): + alive = controller._probe_cluster_liveness(timeout=0.1) + + assert alive == {controller._controller_uuid()}