christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Hold a failing button to fix what it is failing on.

Browse Source 
Fixes come from config.json only and run over SSH with NA Ingest's
status-panel key (ProPresenter restart, Ableton / Church Translate start,
reopen DVS), locally (Resi bridge, captions, ProdDash) or as a HyperDeck
reboot. A green button does nothing, each fix waits 60 s before it can run
again, and a recording deck is never rebooted. New remote checks ask the Macs
every 10 s whether their apps are running. Group buttons and INGEST now have a
hold too; tests never run a real fix or open an SSH connection.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-08 11:34:44 -04:00
co-authored by Claude Opus 5.5
parent e873d1aa40
commit 51f09900ab
6 changed files with 310 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_status_panel.py
+102 -7
View File
@@ -428,6 +428,88 @@ class TestBehaviour(unittest.TestCase):
self.assertEqual(engine.restart_targets("ingest"), [])
class TestFixes(unittest.TestCase):
"""Hold = fix only what that button is failing on. Nothing here runs a real fix."""
def setUp(self):
self.snap = healthy_snapshot(SUNDAY)
self.engine = sp.Engine(CONFIG)
def plan(self, tile):
run_steps(self.engine, self.snap, SUNDAY, 4)
return [(kind, what) for _, kind, what, _ in self.engine.fix_plan(tile, self.snap)]
def test_green_buttons_do_nothing(self):
for tile in ("apps", "dante", "rec", "hd1", "x32", "ingest", "show"):
self.assertEqual(self.plan(tile), [], tile)
def test_propresenter_is_restarted_once(self):
self.snap.conns["Propresenter-API"]["category"] = "error"
self.snap.remote["prop\0pgrep -xq ProPresenter"] = False
self.assertEqual(self.plan("apps"), [("fix", "propresenter")])
self.assertEqual(self.plan("show"), [("fix", "propresenter")]) # the group does the same
def test_apps_on_other_macs(self):
self.snap.remote["tracks\0pgrep -xq Live"] = False
self.snap.remote["translation\0pgrep -xq 'Church Translate'"] = False
self.assertEqual(self.plan("apps"), [("fix", "ableton"), ("fix", "translate")])
def test_unreachable_mac_is_not_a_verdict(self):
self.snap.remote["tracks\0pgrep -xq Live"] = None
self.assertEqual(self.plan("apps"), [])
def test_dante_reopens_dvs_on_the_right_mac(self):
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
devices.remove("NA-ProPresenter-Mac")
devices.remove("BOH-WING") # no fix for a console
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
self.assertEqual(self.plan("dante"), [("fix", "dvs_prop")])
def test_rec_reboots_only_hung_decks_not_recording_ones(self):
for i in (2, 3, 5):
self.snap.conns["HyperDeck_%d" % i]["category"] = "error"
self.snap.vars[("HyperDeck_3", "status")] = "Record" # last known state: recording
self.assertEqual(self.plan("rec"), [("fix", "hd_reboot"), ("fix", "hd_reboot")])
decks = [chk["conn"] for _, _, _, chk in self.engine.fix_plan("rec", self.snap)]
self.assertEqual(decks, ["HyperDeck_2", "HyperDeck_5"])
self.assertEqual(self.plan("hd3"), [])
def test_not_recording_is_not_fixed_by_a_reboot(self):
for i in (1, 3):
self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
self.assertEqual(self.plan("hd5"), []) # NOT REC: a person decides
def test_connection_without_a_fix_is_restarted(self):
self.snap.conns["x32"]["category"] = "error"
self.assertEqual(self.plan("x32"), [("restart", "x32")])
def test_ingest_services(self):
del self.snap.launchd["org.waterschurch.resi-propresenter"]
self.snap.procs["/Applications/ProdDash.app/Contents/MacOS/ProdDash"] = False
self.assertEqual(self.plan("ingest"), [("fix", "bridge"), ("fix", "proddash")])
def test_cooldown_and_executor(self):
ran = []
self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) # record, never run
self.snap.remote["tracks\0pgrep -xq Live"] = False
run_steps(self.engine, self.snap, SUNDAY, 4)
self.engine.run_fixes("apps", self.snap, None, SUNDAY + 10)
self.engine.run_fixes("apps", self.snap, None, SUNDAY + 20) # held again 10 s later: skipped
self.engine.run_fixes("apps", self.snap, None, SUNDAY + 80)
time.sleep(0.2)
self.assertEqual(ran, ["ableton", "ableton"])
def test_remote_script_round_trip(self):
script = sp.remote_script(["pgrep -xq Live", "pgrep -xq 'Church Translate'"])
self.assertIn("(pgrep -xq 'Church Translate') >/dev/null 2>&1; echo SP1=$?", script)
self.assertEqual(sp.parse_remote("SP0=0\nSP1=1\n", 2), [True, False])
self.assertEqual(sp.parse_remote("", 2), [None, None])
def test_every_fix_is_valid(self):
for f in CONFIG["fixes"].values():
self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck_reboot"))
class FakeCompanion(BaseHTTPRequestHandler):
"""Just enough of Companion 5's HTTP API."""
store = {}
@@ -509,12 +591,21 @@ class TestAgainstFakeCompanion(unittest.TestCase):
def test_commands(self):
comp = sp.Companion("127.0.0.1", self.port)
snap = sp.Snapshot()
sp.poll_connections(comp, snap)
store = FakeCompanion.store
snap = healthy_snapshot(SUNDAY)
engine = sp.Engine(CONFIG)
engine.executor = lambda cfg, fix_id, chk: self.fail("no real fix may run in a test")
now = SUNDAY
sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now)
self.assertEqual(FakeCompanion.store["restarts"][-3:], ["id-HyperDeck_3", "id-VX1000", "id-UHDJr"])
store["conns"]["VX1000"]["category"] = "error"
try:
sp.poll_connections(comp, snap)
run_steps(engine, snap, now, 4)
before = len(store["restarts"])
# hd3 is green, so its hold does nothing; LED is failing on VX1000, so that restarts.
sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now + 5)
self.assertEqual(store["restarts"][before:], ["id-VX1000"])
finally:
store["conns"]["VX1000"]["category"] = "good"
self.assertFalse(engine.service.on(now))
sp.handle_commands("service toggle", engine, comp, snap, now)
self.assertTrue(engine.service.on(now))
@@ -533,10 +624,11 @@ class TestAgainstFakeCompanion(unittest.TestCase):
store["conns"]["ATEM"]["category"] = "error"
store["custom"]["sp_cmd"] = ";ack atem"
data_dir = tempfile.mkdtemp()
originals = sp.poll_mdns, sp.poll_system, sp.poll_midi
originals = sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote
sp.poll_mdns = lambda snap, seconds=3: None
sp.poll_system = lambda snap, cfg: None
sp.poll_midi = lambda snap, cfg: None # CoreMIDI has its own test
sp.poll_remote = lambda snap, cfg: None # never SSH anywhere from a test
stop = threading.Event()
loop = threading.Thread(target=sp.run, args=(cfg, data_dir), kwargs={"stop": stop}, daemon=True)
try:
@@ -552,7 +644,7 @@ class TestAgainstFakeCompanion(unittest.TestCase):
finally:
stop.set()
loop.join(timeout=15)
sp.poll_mdns, sp.poll_system, sp.poll_midi = originals
sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote = originals
store["conns"]["ATEM"]["category"] = "good"
self.assertFalse(loop.is_alive()) # it shuts down cleanly
self.assertTrue(store["custom"]["sp_hb"].isdigit())
@@ -614,7 +706,10 @@ class TestPage(unittest.TestCase):
self.assertIn("ackall", master["action_sets"]["up"][0]["options"]["value"]["value"])
self.assertNotIn("2000", master["action_sets"])
ingest = self.controls["3"]["5"]["steps"]["0"]
self.assertNotIn("2000", ingest["action_sets"]) # nothing to restart there
self.assertIn("2000", ingest["action_sets"]) # restarts its services
self.assertIn("2000", self.controls["0"]["2"]["steps"]["0"]["action_sets"]) # REC group
net = self.controls["3"]["4"]["steps"]["0"]
self.assertNotIn("2000", net["action_sets"]) # nothing to fix there
ids = [a["id"] for r in self.controls.values() for c in r.values()
for s in c.get("steps", {}).values() for al in s["action_sets"].values() for a in al]
self.assertEqual(len(ids), len(set(ids)))