Step 0 of the settings page: registry, safe restart, remembered state, test verb, page markers
- schema.py: every setting with type, default, help, owner screen and the rules a config must pass; tests/test_schema.py keeps it in step with the code (every key the code reads is declared, every default matches, check types and fix kinds agree, the glossary covers every word a key can show, today's config still builds today's pages apart from the key notes) - status_panel.py: VERSION + started.json; SIGTERM lets a running fix finish and starts nothing new (RESTARTING SOON); fixes-running.json and live-now.json for the settings page; state.json keeps the ALARMS override, acknowledgements, auto-fix tries and fix times across restarts, keyed by what a check is about rather than its position; 'test' and 'resurvey' commands; queued commands dropped at start; overnight windows; a HyperDeck switched off is always switched back on; the spotify fix kind (playlist set once, in config 'spotify'); Windows remote checks batched for cmd.exe with conditional echoes; every fix's kind and host validated; never_auto; wording and numbers from config (site block, timing, page_hold_ms, auto-fix keep_months); hyperdeck_probe removed; --check and --exit-timeout - make_page.py: numbers from config, markers on every key, nav templates per Companion version, --out; push_page.py: a page is overwritten only when every key carries a marker (foreign keys named by position), secret-free fetch, host from the site; nav_keys.py: host from the site; netgear.py: switches and the shared-port limit from config - install.sh: validates with the new code first, keeps prev/ for --rollback, --keep-config, derived ExitTimeOut, wider deploy list, waits for the start stamp - config.json: site, timing, spotify sections; Windows hosts marked; deployed to NA Ingest and both pages re-imported once (--force) so every key carries its marker Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5.1
@@ -0,0 +1,296 @@
|
||||
"""schema.py must describe exactly what the code reads: every config key the watchdog and the page
|
||||
tools read is declared, every default in the code equals the schema's, every check type and fix
|
||||
kind is listed, every word a key can show is in the glossary, and today's config still builds
|
||||
today's pages."""
|
||||
import ast
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, ROOT)
|
||||
import make_page # noqa: E402
|
||||
import schema # noqa: E402
|
||||
import status_panel as sp # noqa: E402
|
||||
|
||||
WALKED = ["status_panel.py", "make_page.py", "push_page.py", "nav_keys.py", "netgear.py"]
|
||||
|
||||
# What a variable holds, by name, when nothing in the function says otherwise (parameters mostly).
|
||||
NAMES = {"cfg": "top", "config": "top", "chk": "check", "c": "check", "fix": "fix", "f": "fix", "t": "tile",
|
||||
"tile": "tile", "k": "restart_key", "key": "restart_key", "rp": "restart_page", "w": "window",
|
||||
"names": "variables", "host": "host", "h": "host", "windows": "windows", "site": "site"}
|
||||
ATTRS = {"self.cfg": "top", "engine.cfg": "top", "self.net": "network", "self.auto": "auto", "self.site": "site",
|
||||
"self.atem": "atem",
|
||||
"self.tiles": "tiles", "engine.tiles": "tiles", "self.restart_keys": "restart_keys",
|
||||
"engine.restart_keys": "restart_keys"}
|
||||
# A key of a top-level dict that holds a record of another kind.
|
||||
TOP = {"companion": "companion", "network": "network", "variables": "variables", "auto_fix": "auto", "site": "site",
|
||||
"atem": "atem", "spotify": "spotify", "timing": "timing", "flap": "flap", "hosts": "hosts", "fixes": "fixes",
|
||||
"tiles": "tiles", "restart_page": "restart_page", "gear": "gears", "service_windows": "windows"}
|
||||
SUB = {("auto", "windows"): "windows", ("site", "live_services"): "windows", ("tile", "checks"): "checks",
|
||||
("restart_page", "keys"): "restart_keys", ("tile", "members"): "names"}
|
||||
ELEMENT = {"tiles": "tile", "checks": "check", "restart_keys": "restart_key", "windows": "window", "fixes": "fix",
|
||||
"hosts": "host", "gears": "gear"}
|
||||
# Keys the code reads that are the watchdog's own working data, not settings.
|
||||
NOT_SETTINGS = {("check", "ip")}
|
||||
|
||||
|
||||
def dotted(node):
|
||||
if isinstance(node, ast.Name):
|
||||
return node.id
|
||||
if isinstance(node, ast.Attribute):
|
||||
base = dotted(node.value)
|
||||
return base + "." + node.attr if base else None
|
||||
return None
|
||||
|
||||
|
||||
def collect(path):
|
||||
"""[(kind, key, default node or None, line)] for every config read in one file."""
|
||||
tree = ast.parse(open(path, encoding="utf-8").read())
|
||||
reads = []
|
||||
|
||||
def kind_of(node, local):
|
||||
if isinstance(node, ast.Name):
|
||||
return local.get(node.id, NAMES.get(node.id))
|
||||
if isinstance(node, ast.Attribute):
|
||||
return ATTRS.get(dotted(node))
|
||||
if isinstance(node, ast.Subscript):
|
||||
base = kind_of(node.value, local)
|
||||
key = node.slice.value if isinstance(node.slice, ast.Constant) else None
|
||||
return step(base, key)
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == "get" and node.args:
|
||||
base = kind_of(node.func.value, local)
|
||||
key = node.args[0].value if isinstance(node.args[0], ast.Constant) else None
|
||||
return step(base, key)
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr in ("items", "values"):
|
||||
return kind_of(node.func.value, local)
|
||||
if isinstance(node, ast.Call) and dotted(node.func) == "json.load":
|
||||
return "jsonfile" # a config file when it lands in cfg / config
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in ("enumerate", "dict", "sorted", "list") and node.args:
|
||||
return kind_of(node.args[0], local)
|
||||
if isinstance(node, ast.BoolOp):
|
||||
return kind_of(node.values[0], local)
|
||||
if isinstance(node, ast.IfExp):
|
||||
return kind_of(node.body, local)
|
||||
return None
|
||||
|
||||
def step(base, key):
|
||||
if base is None:
|
||||
return None
|
||||
if key is None: # dict[variable]: an element
|
||||
return ELEMENT.get(base)
|
||||
if base == "top":
|
||||
return TOP.get(key)
|
||||
return SUB.get((base, key))
|
||||
|
||||
def scan(body, local):
|
||||
for node in ast.walk(ast.Module(body=body, type_ignores=[])):
|
||||
if isinstance(node, ast.FunctionDef):
|
||||
continue
|
||||
if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
|
||||
kind = kind_of(node.value, local)
|
||||
if kind == "jsonfile":
|
||||
kind = "top" if node.targets[0].id in ("cfg", "config") else None
|
||||
local[node.targets[0].id] = kind
|
||||
elif isinstance(node, (ast.For, ast.comprehension)):
|
||||
elem = ELEMENT.get(kind_of(node.iter, local))
|
||||
if isinstance(node.target, ast.Name):
|
||||
local[node.target.id] = elem
|
||||
elif isinstance(node.target, ast.Tuple) and len(node.target.elts) == 2 \
|
||||
and isinstance(node.target.elts[1], ast.Name):
|
||||
local[node.target.elts[1].id] = elem
|
||||
for node in ast.walk(ast.Module(body=body, type_ignores=[])):
|
||||
if isinstance(node, ast.FunctionDef):
|
||||
continue
|
||||
if isinstance(node, ast.Subscript) and isinstance(node.slice, ast.Constant) and isinstance(node.slice.value, str):
|
||||
kind = kind_of(node.value, local)
|
||||
if kind:
|
||||
reads.append((kind, node.slice.value, None, node.lineno))
|
||||
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == "get" \
|
||||
and node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str):
|
||||
kind = kind_of(node.func.value, local)
|
||||
if kind:
|
||||
reads.append((kind, node.args[0].value, node.args[1] if len(node.args) > 1 else None, node.lineno))
|
||||
|
||||
def functions(node):
|
||||
for child in ast.iter_child_nodes(node):
|
||||
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
local = {a.arg: NAMES.get(a.arg) for a in child.args.args + child.args.kwonlyargs}
|
||||
# a parameter shadows the name map only when the map knows it; "page", "kind" etc. stay unknown
|
||||
scan(child.body, {k: v for k, v in local.items() if v})
|
||||
functions(child)
|
||||
elif isinstance(child, ast.ClassDef):
|
||||
functions(child)
|
||||
scan([n for n in tree.body if not isinstance(n, (ast.FunctionDef, ast.ClassDef))], {})
|
||||
functions(tree)
|
||||
return reads
|
||||
|
||||
|
||||
def all_reads():
|
||||
out = []
|
||||
for name in WALKED:
|
||||
out += [(name,) + r for r in collect(os.path.join(ROOT, name))]
|
||||
return out
|
||||
|
||||
|
||||
def path_of(kind, key):
|
||||
prefix = {"top": "", "check": "tiles[].checks[].", "tile": "tiles[].", "fix": "fixes.*.", "host": "hosts.*.",
|
||||
"restart_key": "restart_page.keys[].", "restart_page": "restart_page.", "auto": "auto_fix.",
|
||||
"network": "network.", "variables": "variables.", "site": "site.", "atem": "atem.", "spotify": "spotify.",
|
||||
"companion": "companion.", "timing": "timing.", "flap": "flap.", "gear": "gear.*.",
|
||||
"window": "auto_fix.windows[]."}.get(kind)
|
||||
return None if prefix is None else prefix + key
|
||||
|
||||
|
||||
class TestSchemaMatchesCode(unittest.TestCase):
|
||||
def test_every_config_read_is_declared(self):
|
||||
record = schema.key_record()
|
||||
missing = sorted({(kind, key) for _, kind, key, _, _ in all_reads()
|
||||
if kind in record and key not in record[kind] and (kind, key) not in NOT_SETTINGS})
|
||||
self.assertEqual(missing, [], "config keys the code reads that schema.py does not declare")
|
||||
kinds = {kind for _, kind, _, _, _ in all_reads()}
|
||||
self.assertNotIn(None, kinds)
|
||||
unknown = sorted(k for k in kinds if k not in record and k not in ("names", "windows", "checks", "tiles",
|
||||
"fixes", "hosts", "restart_keys", "gears"))
|
||||
self.assertEqual(unknown, [])
|
||||
|
||||
def test_defaults_in_code_match_the_schema(self):
|
||||
defaults = schema.defaults()
|
||||
wrong = []
|
||||
for name, kind, key, default, line in all_reads():
|
||||
if default is None or not isinstance(default, ast.Constant):
|
||||
continue
|
||||
path = path_of(kind, key)
|
||||
if path in defaults and defaults[path] != default.value:
|
||||
wrong.append("%s:%d %s is %r in the code, %r in the schema" % (name, line, path, default.value, defaults[path]))
|
||||
self.assertEqual(wrong, [])
|
||||
|
||||
def test_every_declared_key_is_read_somewhere(self):
|
||||
"""Dead settings are caught; keys only the settings page uses are listed here."""
|
||||
ui_only = {"site.id", "site.name", "site.ingest_hostname", "site.ingest_ips", "site.contacts", "site.editing_code",
|
||||
"site.settings_port", "site.dvs_helper", "site.solo_dir", "hosts.*.name", "hosts.*.role",
|
||||
"hosts.*.notes", "hosts.*.pingable", "hosts.*.runs", "hosts.*.networks", "hosts.*.manual_done",
|
||||
"gear", "gear.*.location", "gear.*.notes", "atem.source_folder", "fixes.*.template", "fixes.*.params",
|
||||
"tiles[].checks[].template", "tiles[].checks[].params", "spotify.name", "site", "timing",
|
||||
"fixes", "hosts", "tiles", "flap", "auto_fix", "network", "variables", "companion", "atem", "spotify",
|
||||
"restart_page", "service_windows", "restart_page.keys", "tiles[].checks", "auto_fix.windows",
|
||||
"site.live_services", "tiles[].checks[].type", "fixes.*.playlist", "tiles[].checks[].playlist"}
|
||||
read = {path_of(kind, key) for _, kind, key, _, _ in all_reads()}
|
||||
read |= {"service_windows[].days", "service_windows[].start", "service_windows[].end",
|
||||
"site.live_services[].days", "site.live_services[].start", "site.live_services[].end"}
|
||||
for w in ("days", "start", "end"):
|
||||
if "auto_fix.windows[]." + w in read:
|
||||
read |= {"service_windows[].%s" % w, "site.live_services[].%s" % w}
|
||||
dead = sorted(p for p in schema.F if p not in read and p not in ui_only)
|
||||
self.assertEqual(dead, [], "declared settings nothing reads")
|
||||
|
||||
def test_check_types_fix_kinds_and_subs(self):
|
||||
self.assertEqual(set(schema.CHECKS), set(sp.CHECKS))
|
||||
self.assertEqual(tuple(schema.FIX_KINDS), sp.FIX_KINDS)
|
||||
for t, spec in schema.CHECKS.items():
|
||||
for f in spec["fields"] + spec["identity"]:
|
||||
self.assertIn("tiles[].checks[]." + f, schema.F, "%s field %s" % (t, f))
|
||||
for k in sp.FIX_KINDS:
|
||||
self.assertIn(k, schema.FIX_KIND_FIELDS)
|
||||
|
||||
def test_glossary_covers_every_word_a_key_can_show(self):
|
||||
words = set()
|
||||
for group in schema.GLOSSARY.values():
|
||||
for phrase in group:
|
||||
words |= set(phrase.split())
|
||||
tree = ast.parse(open(os.path.join(ROOT, "status_panel.py"), encoding="utf-8").read())
|
||||
texts = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Call):
|
||||
name = node.func.id if isinstance(node.func, ast.Name) else (node.func.attr if isinstance(node.func, ast.Attribute) else "")
|
||||
if name == "Result" and len(node.args) > 1 and isinstance(node.args[1], ast.Constant):
|
||||
texts.append(node.args[1].value)
|
||||
elif name == "note" and len(node.args) > 1 and isinstance(node.args[1], ast.Constant):
|
||||
texts.append(node.args[1].value)
|
||||
elif name == "note" and len(node.args) > 1 and isinstance(node.args[1], ast.BinOp):
|
||||
for part in ast.walk(node.args[1]):
|
||||
if isinstance(part, ast.Constant) and isinstance(part.value, str):
|
||||
texts.append(part.value)
|
||||
unknown = set()
|
||||
for text in texts:
|
||||
for word in re.findall(r"[A-Z][A-Z-]+", text):
|
||||
if word not in words:
|
||||
unknown.add(word)
|
||||
self.assertEqual(sorted(unknown), [], "words a key can show that the glossary does not explain")
|
||||
|
||||
def test_todays_config_builds_todays_pages(self):
|
||||
cfg = sp.load_config(os.path.join(ROOT, "config.json"))
|
||||
|
||||
def strip(export):
|
||||
for row in export["page"]["controls"].values():
|
||||
for ctl in row.values():
|
||||
if ctl and "options" in ctl:
|
||||
ctl["options"].pop("notes", None)
|
||||
return export
|
||||
for build, name in ((make_page.build, "status-page.companionconfig"), (make_page.build_restart, "restart-page.companionconfig")):
|
||||
on_disk = json.loads(gzip.open(os.path.join(ROOT, name)).read())
|
||||
self.assertEqual(strip(build(cfg)), strip(on_disk), name)
|
||||
self.assertNotIn("restart_page", {k for k in cfg if k == "nothing"})
|
||||
bare = dict(cfg)
|
||||
bare.pop("restart_page")
|
||||
export = make_page.build(bare)
|
||||
notes = [((ctl.get("options") or {}).get("notes") or "") for row in export["page"]["controls"].values()
|
||||
for ctl in row.values() if ctl]
|
||||
self.assertFalse(any("STATUS jump key" in n for n in notes)) # no Restart page: no jump key
|
||||
|
||||
def test_todays_config_passes_the_rules(self):
|
||||
cfg = sp.load_config(os.path.join(ROOT, "config.json"))
|
||||
errors, warnings = schema.validate(cfg)
|
||||
self.assertEqual(errors, [])
|
||||
self.assertTrue(all("only during a service" in w for w in warnings), warnings)
|
||||
|
||||
def test_rules_catch_the_classic_mistakes(self):
|
||||
cfg = sp.load_config(os.path.join(ROOT, "config.json"))
|
||||
bad = json.loads(json.dumps(cfg))
|
||||
bad["tiles"][7]["pos"] = bad["tiles"][8]["pos"] # two keys on one cell
|
||||
bad["tiles"].append({"id": "all", "label": "X", "pos": [3, 8], "kind": "master"})
|
||||
bad["fixes"]["vista"]["host"] = "nowhere"
|
||||
bad["restart_page"]["keys"][0]["fix"] = "no-such-fix"
|
||||
bad["hosts"]["bibletv"]["mac"] = "not-a-mac"
|
||||
bad["auto_fix"]["fixes"].append("propresenter") # closes a program
|
||||
bad["fixes"]["ableton"]["never_auto"] = True # ...and ableton is in the auto list
|
||||
errors, _ = schema.validate(bad)
|
||||
text = "\n".join(errors)
|
||||
for expect in ("share row", "two keys with the same id", "exactly one ALL key", "unknown computer nowhere",
|
||||
"unknown fix no-such-fix", "not a hardware", "closes a running program", "never to run by itself"):
|
||||
self.assertIn(expect, text)
|
||||
home = json.loads(json.dumps(cfg))
|
||||
home["hosts"]["ingest"] = {"self": True, "name": "NA Ingest"}
|
||||
home["fixes"]["vista"]["host"] = "ingest"
|
||||
self.assertIn("is this Mac", "\n".join(schema.validate(home)[0]))
|
||||
|
||||
def test_live_warnings(self):
|
||||
cfg = sp.load_config(os.path.join(ROOT, "config.json"))
|
||||
live = {"connections": {"HyperDeck_2": "generic-tcp-udp"}}
|
||||
_, warnings = schema.validate(cfg, live)
|
||||
self.assertTrue(any("HyperDeck_2 uses module generic-tcp-udp" in w for w in warnings), warnings)
|
||||
self.assertTrue(any("no connection called HyperDeck_1" in w for w in warnings))
|
||||
|
||||
def test_secrets_are_masked(self):
|
||||
masked = schema.mask_secrets({"alert_webhook": "https://hooks/abc", "site": {"editing_code": "apple bee", "name": "NA"},
|
||||
"fixes": {"x": {"run": "ls"}}})
|
||||
self.assertEqual(masked["alert_webhook"], "set (17 characters)")
|
||||
self.assertEqual(masked["site"]["editing_code"], "set (9 characters)")
|
||||
self.assertEqual(masked["site"]["name"], "NA")
|
||||
self.assertEqual(masked["fixes"]["x"]["run"], "ls")
|
||||
|
||||
def test_phrases(self):
|
||||
cfg = sp.load_config(os.path.join(ROOT, "config.json"))
|
||||
p = schema.phrases(cfg)
|
||||
self.assertEqual(p["auto_fix"], "may run by itself on Sundays 06:30-13:30")
|
||||
self.assertEqual(p["live"], "changes wait while a HyperDeck is recording")
|
||||
self.assertEqual(p["hold"], "HOLD 2s")
|
||||
self.assertEqual(sp.windows_phrase([{"days": ["sat", "sun"], "start": "18:00", "end": "00:30"}]),
|
||||
"on Saturdays and Sundays 18:00-00:30")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+148
-90
@@ -130,7 +130,8 @@ class TestChecks(unittest.TestCase):
|
||||
s.spotify_recent.append("spotify:track:q")
|
||||
self.assertEqual(msg().msg, "WRONG PLAYLIST") # full list: one unknown is enough
|
||||
fix = CONFIG["fixes"][chk["fix"]]
|
||||
self.assertIn(chk["playlist"], fix["run"])
|
||||
self.assertEqual(fix["kind"], "spotify")
|
||||
self.assertEqual(sp.spotify_playlist(CONFIG, chk), CONFIG["spotify"]["playlist"])
|
||||
self.assertIn(chk["fix"], CONFIG["auto_fix"]["fixes"])
|
||||
|
||||
def test_connection_states(self):
|
||||
@@ -1352,93 +1353,145 @@ class TestFixes(unittest.TestCase):
|
||||
self.assertEqual(sp.parse_remote("SP0=0\nSP1=1\n", 2), [True, False])
|
||||
self.assertEqual(sp.parse_remote("", 2), [None, None])
|
||||
|
||||
def test_windows_remote_checks_use_conditional_echoes(self):
|
||||
line = sp.remote_script(["tasklist | find /I \"Vista.exe\"", "sc query rtpMIDIService | find \"RUNNING\""], "cmd")
|
||||
self.assertNotIn("%ERRORLEVEL%", line)
|
||||
self.assertIn("&& echo SP0=0 || echo SP0=1", line)
|
||||
self.assertIn("&& echo SP1=0 || echo SP1=1", line)
|
||||
self.assertEqual(sp.parse_remote("SP0=0\r\nSP1=1\r\n", 2), [True, False])
|
||||
self.assertEqual(sp.host_shell(CONFIG, "lighting"), "cmd")
|
||||
self.assertEqual(sp.host_shell(CONFIG, "tracks"), "sh")
|
||||
|
||||
def test_overnight_windows(self):
|
||||
sat_2350 = datetime(2026, 10, 10, 23, 50).timestamp()
|
||||
sun_0020 = datetime(2026, 10, 11, 0, 20).timestamp()
|
||||
sun_0100 = datetime(2026, 10, 11, 1, 0).timestamp()
|
||||
w = [{"days": ["sat"], "start": "18:00", "end": "00:30"}]
|
||||
self.assertTrue(sp.in_windows(w, sat_2350))
|
||||
self.assertTrue(sp.in_windows(w, sun_0020)) # past midnight, still Saturday's service
|
||||
self.assertFalse(sp.in_windows(w, sun_0100))
|
||||
self.assertFalse(sp.in_windows([{"days": ["sun"], "start": "18:00", "end": "00:30"}], sat_2350))
|
||||
|
||||
def test_check_identity_is_about_the_thing_not_the_position(self):
|
||||
midi = [c for c in CONFIG["tiles"][9]["checks"] if c["type"] == "midi_peer"]
|
||||
self.assertEqual(len({sp.check_identity(c) for c in midi}), len(midi)) # six peers, six identities
|
||||
self.assertEqual(sp.check_identity({"type": "midi_peer", "host": "tracks", "session": "NA Tracks", "peer": "Lighting PC"}),
|
||||
"tracks/NA Tracks/Lighting PC")
|
||||
self.assertEqual(sp.check_identity({"type": "ping", "conn": "Axient_1", "host": "10.1.1.101"}), "Axient_1")
|
||||
self.assertEqual(sp.check_identity({"type": "conn", "conn": "ATEM"}), "ATEM")
|
||||
|
||||
def test_acknowledgements_survive_a_restart(self):
|
||||
state = os.path.join(tempfile.mkdtemp(prefix="sp-state-"), "state.json")
|
||||
engine = sp.Engine(CONFIG, state)
|
||||
engine.executor = engine.poe_executor = lambda *a: True
|
||||
snap = healthy_snapshot(SUNDAY)
|
||||
snap.conns["ATEM"]["category"] = "error"
|
||||
run_steps(engine, snap, SUNDAY, 5)
|
||||
engine.ack("atem")
|
||||
self.assertEqual(level(run_steps(engine, snap, SUNDAY + 6, 1)[0]["atem"]), "ack")
|
||||
saved = json.load(open(state))
|
||||
self.assertIn("atem|ATEM|conn", saved["acks"])
|
||||
again = sp.Engine(CONFIG, state) # the watchdog restarts
|
||||
again.executor = again.poe_executor = lambda *a: True
|
||||
tiles, _ = run_steps(again, snap, SUNDAY + 10, 5)
|
||||
self.assertEqual(level(tiles["atem"]), "ack") # still acknowledged, no new flashing
|
||||
snap.conns["ATEM"]["category"] = "good"
|
||||
run_steps(again, snap, SUNDAY + 20, 2)
|
||||
self.assertEqual(json.load(open(state))["acks"], []) # cleared: forgotten
|
||||
|
||||
def test_run_this_fix_now_goes_through_the_hold_path(self):
|
||||
ran = []
|
||||
self.engine.executor = lambda cfg, fix_id, chk: ran.append((fix_id, chk.get("conn"), chk.get("ip"), chk.get("peer"))) or True
|
||||
snap = self.snap
|
||||
run_steps(self.engine, snap, SUNDAY, 2)
|
||||
comp = mock_comp()
|
||||
now = SUNDAY + 3
|
||||
self.engine.run_test("hd2", "hyperdeck", "0", snap, comp, now) # the check's number...
|
||||
self.engine.run_test("midi", "midi_reconnect", "tracks/NA Tracks/Studio B Mac Studio", snap, comp, now) # ...or its identity
|
||||
self.engine.run_test("midi", "midi_reconnect", "9", snap, comp, now) # no such check
|
||||
self.engine.run_test("atem", "atem_audio", "1", snap, comp, now) # never from the page
|
||||
self.engine.run_test("hd2", "propresenter", "0", snap, comp, now) # not attached to that key
|
||||
for t in self.engine.fix_threads:
|
||||
t.join(2)
|
||||
self.assertEqual(ran, [("hyperdeck", "HyperDeck_2", "10.1.10.12", None),
|
||||
("midi_reconnect", None, None, "Studio B Mac Studio")])
|
||||
self.assertEqual(self.engine.notes["hd2"]["text"], "DONE")
|
||||
|
||||
def test_a_restart_lets_the_running_fix_finish_and_starts_nothing_new(self):
|
||||
gate = threading.Event()
|
||||
self.engine.executor = lambda cfg, fix_id, chk: gate.wait(3) or True
|
||||
run_steps(self.engine, self.snap, SUNDAY, 2)
|
||||
self.engine.run_force("rs_bridge", self.snap, mock_comp(), SUNDAY + 3)
|
||||
self.assertEqual(len(self.engine.running), 1) # published for the settings page
|
||||
self.engine.stopping = True # SIGTERM arrived
|
||||
self.engine.run_force("rs_captions", self.snap, mock_comp(), SUNDAY + 4)
|
||||
self.assertEqual(self.engine.notes["rs_captions"]["text"], "RESTARTING\nSOON")
|
||||
self.assertEqual(self.engine.run_fixes("ingest", self.snap, mock_comp(), SUNDAY + 4), [])
|
||||
gate.set()
|
||||
for t in self.engine.fix_threads:
|
||||
t.join(2)
|
||||
self.assertEqual(self.engine.running, {})
|
||||
self.assertEqual(sp.exit_timeout(CONFIG), 90) # the longest fix + its verify + 10
|
||||
|
||||
def test_a_hyperdeck_switched_off_is_always_switched_back_on(self):
|
||||
calls = []
|
||||
|
||||
class Comp:
|
||||
def connections(self):
|
||||
return [{"label": "HyperDeck_2", "id": "x", "enabled": True, "status": {"category": "good"}}]
|
||||
|
||||
def disable_connection(self, cid):
|
||||
calls.append("off")
|
||||
return True
|
||||
|
||||
def enable_connection(self, cid):
|
||||
calls.append("on")
|
||||
return True
|
||||
import unittest.mock as mock
|
||||
|
||||
def boom(seconds):
|
||||
raise RuntimeError("killed mid-way")
|
||||
with mock.patch.object(sp, "Companion", lambda *a, **k: Comp()):
|
||||
with self.assertRaises(RuntimeError):
|
||||
sp.reconnect_and_verify(CONFIG, CONFIG["fixes"]["hyperdeck"], "x", sleep=boom, conn="HyperDeck_2", toggle=True)
|
||||
self.assertEqual(calls, ["off", "on"])
|
||||
|
||||
def test_spotify_fix_opens_and_plays(self):
|
||||
calls = []
|
||||
|
||||
class R:
|
||||
def __init__(self, code, out=""):
|
||||
self.returncode, self.stdout, self.stderr = code, out, ""
|
||||
|
||||
def run(argv, **kw):
|
||||
calls.append(argv)
|
||||
if argv[1] == "-xq":
|
||||
return R(1) # Spotify is closed
|
||||
if "player state" in argv[-1]:
|
||||
return R(0, "playing\n")
|
||||
return R(0)
|
||||
import unittest.mock as mock
|
||||
with mock.patch.object(sp.time, "sleep", lambda s: None):
|
||||
ok = sp.play_spotify(CONFIG, CONFIG["fixes"]["spotify_play"], {}, "spotify", run=run)
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(calls[1][:3], ["/usr/bin/open", "-g", "-a"])
|
||||
self.assertIn(CONFIG["spotify"]["playlist"], calls[2][-1])
|
||||
|
||||
def test_never_auto_fixes_stay_a_hold(self):
|
||||
cfg = json.loads(json.dumps(CONFIG))
|
||||
cfg["fixes"]["ableton"]["never_auto"] = True
|
||||
cfg["auto_fix"]["fixes"].remove("ableton")
|
||||
engine = sp.Engine(cfg, None)
|
||||
self.assertNotIn("ableton", {f for f in engine.auto.get("fixes", []) if not engine.cfg["fixes"][f].get("never_auto")})
|
||||
cfg["auto_fix"]["fixes"].append("ableton")
|
||||
path = os.path.join(tempfile.mkdtemp(prefix="sp-cfg-"), "config.json")
|
||||
json.dump(cfg, open(path, "w"))
|
||||
with self.assertRaises(ValueError):
|
||||
sp.load_config(path)
|
||||
|
||||
def test_every_fix_is_valid(self):
|
||||
for f in CONFIG["fixes"].values():
|
||||
self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck", "midi_connect", "reconnect", "atem"))
|
||||
|
||||
|
||||
class FakeDeck:
|
||||
"""A pretend HyperDeck on a local port: answers like a real one, stays silent, or greets only."""
|
||||
|
||||
def __init__(self, mode):
|
||||
import socket as _s
|
||||
self.mode, self.received = mode, b""
|
||||
self.srv = _s.socket()
|
||||
self.srv.bind(("127.0.0.1", 0))
|
||||
self.srv.listen(5)
|
||||
self.port = self.srv.getsockname()[1]
|
||||
threading.Thread(target=self.serve, daemon=True).start()
|
||||
|
||||
def serve(self):
|
||||
while True:
|
||||
try:
|
||||
conn, _ = self.srv.accept()
|
||||
except OSError:
|
||||
return
|
||||
with conn:
|
||||
conn.settimeout(4)
|
||||
if self.mode != "silent":
|
||||
conn.sendall(b"500 connection info:\r\nprotocol version: 1.11\r\nmodel: HyperDeck Studio 4K Pro\r\n\r\n")
|
||||
try:
|
||||
data = conn.recv(1024)
|
||||
self.received += data
|
||||
if self.mode in ("record", "preview") and b"transport info" in data:
|
||||
conn.sendall(b"208 transport info:\r\nstatus: %s\r\nspeed: 100\r\n\r\n" % self.mode.encode())
|
||||
more = conn.recv(1024)
|
||||
self.received += more
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def close(self):
|
||||
self.srv.close()
|
||||
|
||||
|
||||
class TestHyperDeckProbe(unittest.TestCase):
|
||||
def test_probe_answers(self):
|
||||
for mode, expected in (("record", "record"), ("preview", "preview"), ("greet", "answered"), ("silent", "hung")):
|
||||
deck = FakeDeck(mode)
|
||||
try:
|
||||
self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=deck.port), expected, mode)
|
||||
finally:
|
||||
deck.close()
|
||||
deck = FakeDeck("preview")
|
||||
port = deck.port
|
||||
deck.close()
|
||||
self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=port), "unreachable")
|
||||
|
||||
def test_the_hold_switches_the_connection_off_and_on_and_never_reboots(self):
|
||||
from unittest import mock
|
||||
deck = FakeDeck("silent") # even a deck that would once have been rebooted
|
||||
try:
|
||||
comp = mock.Mock()
|
||||
comp.connections.side_effect = [
|
||||
[{"label": "HyperDeck_1", "id": "h1", "enabled": True, "status": {"category": "error"}}],
|
||||
[{"label": "HyperDeck_1", "id": "h1", "enabled": True, "status": {"category": "good"}}]]
|
||||
comp.disable_connection.return_value = comp.enable_connection.return_value = True
|
||||
chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port}
|
||||
with mock.patch.object(sp, "Companion", return_value=comp), mock.patch.object(sp.time, "sleep"):
|
||||
self.assertTrue(sp.execute_fix(CONFIG, "hyperdeck", chk))
|
||||
self.assertEqual([c[0] for c in comp.method_calls if c[0].endswith("_connection")],
|
||||
["disable_connection", "enable_connection"])
|
||||
comp.restart_connection.assert_not_called()
|
||||
time.sleep(0.3)
|
||||
self.assertNotIn(b"reboot", deck.received)
|
||||
finally:
|
||||
deck.close()
|
||||
|
||||
def test_a_deck_that_stays_offline_says_so(self):
|
||||
from unittest import mock
|
||||
fix = CONFIG["fixes"]["hyperdeck"]
|
||||
for pings, expect in ((True, "STILL OFFLINE\nHOLD AGAIN"), (False, "NO NETWORK\nCHECK POWER")):
|
||||
comp = mock.Mock()
|
||||
comp.connections.side_effect = [
|
||||
[{"label": "HyperDeck_2", "id": "h2", "enabled": True, "status": {"category": "error"}}],
|
||||
[{"label": "HyperDeck_2", "id": "h2", "enabled": True, "status": {"category": "error"}}]]
|
||||
with mock.patch.object(sp, "Companion", return_value=comp), \
|
||||
mock.patch.object(sp, "ping_once", return_value=pings) as ping:
|
||||
self.assertEqual(sp.reconnect_and_verify(CONFIG, fix, "x", sleep=lambda s: None,
|
||||
conn="HyperDeck_2", host="10.1.10.12", toggle=True), expect)
|
||||
ping.assert_called_once_with("10.1.10.12") # the deck's own address, from Companion
|
||||
self.assertIn(f.get("kind", "ssh"), sp.FIX_KINDS)
|
||||
|
||||
|
||||
class TestNetgearParsing(unittest.TestCase):
|
||||
@@ -1679,10 +1732,15 @@ class TestPage(unittest.TestCase):
|
||||
nav = {"0": {"0": {"type": "pageup"}}, "1": {"0": {"type": "button-layered",
|
||||
"options": {"notes": "STATUS nav key (pagenum): x"}}}}
|
||||
self.assertTrue(push_page.safe_target({"page": {"name": "PAGE", "controls": nav}}, "23\\nRestart")[0])
|
||||
busy = {"page": {"name": "PAGE", "controls": dict(nav, **{"2": {"3": {"type": "button"}}})}}
|
||||
self.assertFalse(push_page.safe_target(busy, "23\\nRestart")[0]) # pages 22 and 97
|
||||
busy["page"]["name"] = "23\\nRestart"
|
||||
self.assertTrue(push_page.safe_target(busy, "23\\nRestart")[0]) # our own page again
|
||||
ours = {"page": {"name": "whatever", "controls": dict(nav, **{"3": {"7": {"type": "button-layered",
|
||||
"options": {"notes": "STATUS tile 'led' - drawn by status_panel.py on NA Ingest."}},
|
||||
"6": {"type": "button-layered", "options": {"notes": "STATUS jump key - Tap: open the Restart page."}}}})}}
|
||||
self.assertTrue(push_page.safe_target(ours, "23\\nRestart")[0]) # our keys, whatever the page is called
|
||||
busy = {"page": {"name": "23\\nRestart", "controls": dict(nav, **{"2": {"3": {"type": "button"}}})}}
|
||||
ok, why = push_page.safe_target(busy, "23\\nRestart")
|
||||
self.assertFalse(ok) # a key someone made by hand: refused
|
||||
self.assertIn("row 2 col 3", why) # ...and named by its place
|
||||
self.assertIn("move or delete them first", why)
|
||||
|
||||
def test_each_button_reads_its_own_tile(self):
|
||||
for t in CONFIG["tiles"]:
|
||||
|
||||
Reference in New Issue
Block a user