RESI LIVE NOT REC: every deck with a disk must record while Resi plays; a hold starts the idle decks (never touches a recording one)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_status_panel.py

@@ -630,6 +630,50 @@ class TestChecks(unittest.TestCase):
         self.assertEqual(tiles["ingest"]["x"], "INGEST\nBRIDGE NOT LOADED\n+2")
 
 
+class TestHyperDeckRecord(unittest.TestCase):
+    """hyperdeck_record() against a fake deck speaking the HyperDeck protocol (port 9993)."""
+    def deck(self, recording=False):
+        import socketserver
+        state = {"status": "record" if recording else "preview", "got": []}
+
+        class H(socketserver.BaseRequestHandler):
+            def handle(self):
+                self.request.sendall(b"500 connection info:\r\nprotocol version: 1.19\r\n\r\n")
+                buf = b""
+                while True:
+                    data = self.request.recv(1024)
+                    if not data:
+                        return
+                    buf += data
+                    while b"\r\n" in buf:
+                        line, buf = buf.split(b"\r\n", 1)
+                        cmd = line.decode()
+                        state["got"].append(cmd)
+                        if cmd == "transport info":
+                            self.request.sendall(("208 transport info:\r\nstatus: %s\r\n\r\n" % state["status"]).encode())
+                        elif cmd.startswith("record"):
+                            state["status"] = "record"
+                            self.request.sendall(b"200 ok\r\n")
+        srv = socketserver.ThreadingTCPServer(("127.0.0.1", 0), H)
+        srv.daemon_threads = True
+        threading.Thread(target=srv.serve_forever, daemon=True).start()
+        self.addCleanup(srv.shutdown)
+        return srv.server_address[1], state
+
+    def test_an_idle_deck_is_started_with_its_name(self):
+        port, state = self.deck()
+        self.assertTrue(sp.hyperdeck_record("127.0.0.1", "CAM3-20261027_090000-", port=port))
+        self.assertIn("record: name: CAM3-20261027_090000-", state["got"])
+
+    def test_a_recording_deck_is_sent_no_record(self):
+        port, state = self.deck(recording=True)
+        self.assertTrue(sp.hyperdeck_record("127.0.0.1", "CAM3-x-", port=port))
+        self.assertFalse([c for c in state["got"] if c.startswith("record")])
+
+    def test_no_deck_is_false(self):
+        self.assertFalse(sp.hyperdeck_record("127.0.0.1", "x", port=1, timeout=1))
+
+
 class TestTextSize(unittest.TestCase):
     def test_no_word_is_split(self):
         for text in ("TASCAM\nOK", "SERVICE\nON\nmanual", "OBS\nOFFLINE", "INGEST\n97GB free",
@@ -1473,6 +1517,44 @@ class TestFixes(unittest.TestCase):
         self.assertEqual(self.plan("rec"), [("enable", "HyperDeck_1")])
         self.assertEqual(self.plan("x32"), [("enable", "x32")])
 
+    def test_resi_live_and_no_deck_recording_is_red_and_a_hold_starts_them(self):
+        # 2026-10-27: "notify me if we are not recording on the hyperdecks, but resi is playing"
+        self.snap.vars[("resi-studio-cues", "playback_state")] = "PLAYING"
+        self.snap.vars[("HyperDeck_2", "slotId")] = "-"                 # optional deck, no disk: not in use
+        self.snap.vars[("HyperDeck_2", "recordingTime")] = "--:--:--"
+        tiles, _ = run_steps(self.engine, self.snap, SUNDAY, 30)
+        self.assertEqual(level(tiles["hd1"]), "ok")                       # not yet: Resi just started
+        tiles, _ = run_steps(self.engine, self.snap, SUNDAY + 30, 40)
+        self.assertEqual(tiles["hd1"]["x"], "HD1\nRESI LIVE NOT REC")
+        self.assertEqual(level(tiles["rec"]), "err")
+        self.assertEqual(level(tiles["hd2"]), "off")
+        self.snap.vars[("HyperDeck_3", "status")] = "Record"             # once one records, the others are NOT REC
+        tiles, _ = run_steps(self.engine, self.snap, SUNDAY + 70, 4)
+        self.assertEqual(tiles["hd1"]["x"], "HD1\nNOT REC")
+        started = []
+        self.engine.record_executor = lambda ip, name: started.append((ip, name)) or True
+        self.engine.run_fixes("rec", self.snap, None, SUNDAY + 80)
+        time.sleep(0.2)
+        self.assertEqual([ip for ip, _ in started], ["10.1.10.11", "10.1.10.14", "10.1.10.15", "10.1.10.16"])
+        self.assertTrue(started[0][1].startswith("CAM3-") and started[0][1].endswith("-"))
+        self.assertTrue(started[3][1].startswith("PGM4-"))
+        self.assertEqual(self.engine.notes["rec"]["text"], "STARTING\nREC")
+
+    def test_resi_paused_or_its_module_down_is_not_live(self):
+        self.snap.vars[("resi-studio-cues", "playback_state")] = "PAUSED"
+        tiles, _ = run_steps(self.engine, self.snap, SUNDAY, 70)
+        self.assertEqual(level(tiles["hd1"]), "ok")
+        self.snap.vars[("resi-studio-cues", "playback_state")] = "PLAYING"
+        self.snap.conns["resi-studio-cues"]["category"] = "error"
+        run_steps(self.engine, self.snap, SUNDAY + 70, 70)
+        self.assertFalse(self.engine.resi_live)
+        ran = []
+        self.engine.executor = lambda cfg, fix_id, chk: ran.append(chk.get("conn")) or True
+        self.engine.record_executor = lambda ip, name: self.fail("must not record when Resi is not live")
+        self.engine.run_fixes("hd1", self.snap, None, SUNDAY + 150)      # the usual hold: reconnect
+        time.sleep(0.2)
+        self.assertEqual(ran, ["HyperDeck_1"])
+
     def test_not_recording_is_not_fixed_by_a_reboot(self):
         for i in (1, 3):
             self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"