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

Step 2: the settings page can change things - shared draft, Review & Apply, rollback, history, undo

Browse Source 
- settings/apply.py: the Apply pipeline (check, keep history, write, restart the watchdog, re-import
  pages by the marker-only rule, recolour page keys), rollback when the watchdog does not come back,
  the recording guard with the Emergency (LIVE) override, history entries and Go back / Undo.
- settings/server.py: the write routes behind the editing code (draft PUT/DELETE with 409 on a
  conflict, validate, apply, restore, undo, watchdog restart/stop/start, commands to the watchdog,
  the switcher file upload / pick newest / discard, gear notes).
- settings/ui/app.js: every screen grows inputs with Editing on; keys (checks, fixes, hold-always,
  members), computers, fixes, site setup (schedules, Dante devices, Spotify, ATEM), Review & Apply
  with live progress, History with Go back and Undo, Run this fix now.
- schema.unmask, push_page.push, nav_keys.colour; tests for all of it (163 pass).
- Verified live on NA Ingest: a page-only Apply, a restart Apply, Undo, history.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-14 14:06:42 -04:00
co-authored by Claude Fable 5.1
parent ae3e85f152
commit 02d0c77cd9
10 changed files with 1686 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_apply.py
+228
View File
@@ -0,0 +1,228 @@
"""Saving a change, offline: the shared draft, what Apply will do, the pipeline against a temp data
dir with a fake launchctl and a fake Companion, rollback when the watchdog does not come back, the
recording guard with its Emergency override, history and Go back."""
import json
import os
import sys
import tempfile
import threading
import time
import unittest
from unittest import mock
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import status_panel as sp # noqa: E402
from settings import apply as ap # noqa: E402
from settings import server # noqa: E402
from test_status_panel import CONFIG # noqa: E402
class FakeLaunchctl:
"""Records calls; a kickstart or bootstrap "starts" the watchdog by writing started.json
(with the sha of the config on disk) unless told to fail."""
def __init__(self, data_dir, fail=False):
self.data_dir, self.fail, self.calls, self.pid = data_dir, fail, [], 100
def __call__(self, argv, **kw):
self.calls.append(argv)
class R:
returncode, stdout, stderr = 0, "", ""
r = R()
if argv[1] == "print":
r.stdout = "state = running\n\tpid = %d\n" % self.pid
elif argv[1] in ("kickstart", "bootstrap"):
if not self.fail:
self.pid += 1
cfg_sha = ap.sha(open(os.path.join(self.data_dir, "config.json"), "rb").read())
sp.write_json(os.path.join(self.data_dir, "started.json"), {"pid": self.pid, "started_at": time.time(), "config_sha": cfg_sha, "version": "test"})
return r
class TestDraftAndPlan(unittest.TestCase):
def setUp(self):
self.data = tempfile.mkdtemp(prefix="sp-apply-")
self.cfg = json.loads(json.dumps(CONFIG))
self.cfg["companion"] = {"host": "127.0.0.1", "port": 1}
with open(os.path.join(self.data, "config.json"), "wb") as f:
f.write(ap.config_text(self.cfg))
def test_draft_is_shared_and_never_overwritten_silently(self):
d = ap.Draft(self.data)
a, _ = d.update({"x": 1}, "sam", "sha1")
self.assertEqual(oct(os.stat(d.path).st_mode)[-3:], "600")
b, conflict = d.update({"x": 2}, "chris", "sha1", expect=a["version"])
self.assertIsNone(conflict)
_, conflict = d.update({"x": 3}, "sam", "sha1", expect=a["version"]) # sam's tab is behind
self.assertEqual(conflict["by"], "chris")
self.assertEqual(d.read()["config"], {"x": 2})
def test_what_apply_does(self):
hashes = ap.build_hashes(self.cfg)
same = json.loads(json.dumps(self.cfg))
self.assertEqual(ap.what_apply_does(self.cfg, same, hashes)["changes"], [])
notes = json.loads(json.dumps(self.cfg))
notes["gear"] = {"UHDJr": {"location": "lobby rack"}}
plan = ap.what_apply_does(self.cfg, notes, hashes)
self.assertFalse(plan["restart"]) # page-only data
self.assertEqual(plan["pages"], [])
timing = json.loads(json.dumps(self.cfg))
timing["debounce_seconds"] = 4
self.assertTrue(ap.what_apply_does(self.cfg, timing, hashes)["restart"])
self.assertEqual(ap.what_apply_does(self.cfg, timing, hashes)["pages"], [])
moved = json.loads(json.dumps(self.cfg))
next(t for t in moved["tiles"] if t["id"] == "led")["pos"] = [2, 8]
plan = ap.what_apply_does(self.cfg, moved, hashes)
self.assertEqual(plan["pages"], ["status"])
self.assertTrue(plan["nav"])
label = json.loads(json.dumps(self.cfg))
label["fixes"]["vista"]["label"] = "start Vista please" # only a key note changes
self.assertEqual(ap.what_apply_does(self.cfg, label, hashes)["pages"], [])
def test_words(self):
after = json.loads(json.dumps(self.cfg))
hd2 = next(t for t in after["tiles"] if t["id"] == "hd2")
hd2["checks"][0]["optional"] = False
after["site"]["editing_code"] = "new words"
after["hosts"]["lighting"]["notes"] = "keep it Public"
texts = [ap.words_for(p, a, b, self.cfg, after)["text"] for p, a, b in ap.changed_paths(self.cfg, after)]
self.assertIn("key HD2 › check 1 › optional: yes → no", texts)
self.assertIn("computer Lighting PC › notes: Keep the network profile on Public: rtpMIDI's Bonjour rules only allow Public. → keep it Public", texts)
self.assertTrue(any(t == "site › editing code changed" for t in texts), texts) # a secret: never its value
class TestPipeline(unittest.TestCase):
def setUp(self):
self.data = tempfile.mkdtemp(prefix="sp-pipe-")
self.cfg = json.loads(json.dumps(CONFIG))
self.cfg["companion"] = {"host": "127.0.0.1", "port": 1}
with open(os.path.join(self.data, "config.json"), "wb") as f:
f.write(ap.config_text(self.cfg))
open(os.path.join(self.data, "atem-reference.xml"), "w").write("<x/>")
self.site = server.Site(self.data, companion=mock.Mock())
self.launchctl = FakeLaunchctl(self.data)
self.applier = ap.Applier(self.site, run=self.launchctl, sleep=lambda s: None)
def run_apply(self, new_cfg, note="t", emergency=None):
r = self.applier.start(new_cfg, note, "sam", "10.1.1.50", emergency)
if r.get("ok"):
self.applier.thread.join(30)
return r
def test_a_timing_change_restarts_and_is_kept_in_history(self):
new = json.loads(json.dumps(self.cfg))
new["debounce_seconds"] = 4
r = self.run_apply(new, "slower")
self.assertTrue(r["ok"])
st = self.applier.status
self.assertEqual(st["state"], "done", st)
names = {s["name"]: s["state"] for s in st["steps"]}
self.assertEqual((names["check"], names["save"], names["restart"], names["pages"]), ("ok", "ok", "ok", "skipped"))
self.assertEqual(json.load(open(os.path.join(self.data, "config.json")))["debounce_seconds"], 4)
self.assertTrue(any(a[1] == "kickstart" for a in self.launchctl.calls))
entries = ap.history_entries(self.data)
self.assertEqual((entries[0]["by"], entries[0]["note"], entries[0]["actions"]), ("sam", "slower", ["restart"]))
self.assertEqual(ap.history_config(self.data, entries[0]["id"])["debounce_seconds"], 3) # the version before
self.assertEqual(oct(os.stat(os.path.join(self.data, "history", entries[0]["id"] + ".json")).st_mode)[-3:], "600")
def test_page_only_data_needs_no_restart(self):
new = json.loads(json.dumps(self.cfg))
new["gear"] = {"UHDJr": {"location": "lobby"}}
self.run_apply(new)
names = {s["name"]: s["state"] for s in self.applier.status["steps"]}
self.assertEqual(names["restart"], "skipped")
self.assertFalse(any(a[1] == "kickstart" for a in self.launchctl.calls))
def test_a_bad_config_is_refused_before_anything_happens(self):
new = json.loads(json.dumps(self.cfg))
new["fixes"]["vista"]["host"] = "nowhere"
self.run_apply(new)
self.assertEqual(self.applier.status["state"], "failed")
self.assertIn("unknown computer nowhere", self.applier.status["outcome"])
self.assertEqual(json.load(open(os.path.join(self.data, "config.json")))["fixes"]["vista"]["host"], "lighting")
self.assertEqual(ap.history_entries(self.data), [])
def test_rollback_when_the_watchdog_does_not_come_back(self):
self.launchctl.fail = True
new = json.loads(json.dumps(self.cfg))
new["debounce_seconds"] = 5
self.run_apply(new, "will fail")
st = self.applier.status
self.assertEqual(st["state"], "failed")
self.assertIn("previous settings", st["outcome"].lower())
self.assertEqual(json.load(open(os.path.join(self.data, "config.json")))["debounce_seconds"], 3) # put back
self.assertTrue(os.path.exists(os.path.join(self.data, "last-apply-failed.json")))
meta = ap.history_entries(self.data)[0]
self.assertIn("rolled back", meta["outcome"])
def test_layout_change_imports_the_page_and_recolours_the_nav_keys(self):
new = json.loads(json.dumps(self.cfg))
next(t for t in new["tiles"] if t["id"] == "led")["pos"] = [2, 8]
pushed, coloured, calls = [], [], {"n": 0}
def fetch(h, p, page): # before the push: only page keys; after: as built
calls["n"] += 1
return {"page": {"name": "19\\nStatus", "controls": {}}} if calls["n"] % 2 == 1 else ap.make_page.build(new)
with mock.patch.object(ap.push_page, "fetch_page", fetch), \
mock.patch.object(ap.push_page, "push", lambda *a: pushed.append(a[2])), \
mock.patch.object(ap.nav_keys, "colour", lambda *a: coloured.append(a) or 96):
self.run_apply(new, "move LED")
st = self.applier.status
self.assertEqual(st["state"], "done", st)
self.assertEqual(pushed, [19])
self.assertEqual(len(coloured), 1)
self.assertEqual(json.load(open(os.path.join(self.data, "build-hashes.json")))["status"], ap.build_hashes(new)["status"])
def test_a_foreign_key_on_the_page_stops_the_import_and_rolls_back(self):
new = json.loads(json.dumps(self.cfg))
next(t for t in new["tiles"] if t["id"] == "led")["pos"] = [2, 8]
busy = {"page": {"name": "19\\nStatus", "controls": {"2": {"3": {"type": "button-layered", "options": {"notes": "someone's key"}}}}}}
with mock.patch.object(ap.push_page, "fetch_page", lambda h, p, page: busy), mock.patch.object(ap.push_page, "push", lambda *a: self.fail("must not push")):
self.run_apply(new, "move LED")
st = self.applier.status
self.assertEqual(st["state"], "failed")
self.assertIn("row 2 col 3", st["outcome"])
self.assertEqual(next(t for t in json.load(open(os.path.join(self.data, "config.json")))["tiles"] if t["id"] == "led")["pos"], [3, 6])
self.assertTrue(os.path.exists(os.path.join(self.data, "pages-out-of-sync.json")))
def test_live_refuses_unless_emergency(self):
sp.write_json(os.path.join(self.data, "live-now.json"), {"live": True, "reason": "a HyperDeck is recording", "time": time.time()})
new = json.loads(json.dumps(self.cfg))
new["debounce_seconds"] = 4
r = self.run_apply(new)
self.assertFalse(r["ok"])
self.assertTrue(r.get("live"))
r = self.run_apply(new, "urgent", {"word": "LIVE", "reason": "the deck is wrong"})
self.assertTrue(r["ok"])
self.assertEqual(self.applier.status["state"], "done")
self.assertIn("EMERGENCY apply by sam", open(os.path.join(self.data, "apply.log")).read())
sp.write_json(os.path.join(self.data, "live-now.json"), {"live": True, "reason": "stale", "time": time.time() - 120})
self.assertEqual(self.applier.live(), (False, "")) # stale: the schedule (none) decides
def test_a_new_switcher_file_goes_with_the_config_and_is_kept(self):
pending = os.path.join(self.data, "draft-atem.xml")
open(pending, "w").write("<y/>")
new = json.loads(json.dumps(self.cfg))
r = self.applier.start(new, "new switcher file", "sam", "10.1.1.50", None, pending)
self.applier.thread.join(30)
self.assertEqual(open(os.path.join(self.data, "atem-reference.xml")).read(), "<y/>")
self.assertTrue(os.listdir(os.path.join(self.data, "history", "atem"))) # the old one, dated
self.assertFalse(os.path.exists(pending))
self.assertIn("switcher file", ap.history_entries(self.data)[0]["actions"])
def test_one_apply_at_a_time(self):
gate = threading.Event()
slow = ap.Applier(self.site, run=self.launchctl, sleep=lambda s: gate.wait(2))
new = json.loads(json.dumps(self.cfg))
new["debounce_seconds"] = 4
self.assertTrue(slow.start(new, "a", "sam", "x")["ok"])
self.assertIn("already running", slow.start(new, "b", "sam", "x")["error"])
gate.set()
slow.thread.join(30)
if __name__ == "__main__":
unittest.main()