christianandClaude Opus 5.5
d5ede0c91e
- readiness.py looks every 2 minutes (never while a HyperDeck records) at this weekend's PCO plan and what is in place for it: the Ableton set (songs, mix, routing, lighting cues), ProPresenter (every PCO item, MC announcements, baptism names, timelines, backgrounds), the LTC note of every song against the timecode the tracks play, the Band Lineup TV, Resolume's LED cross, and the tracks and ProP bundles for the locations. - READY key on the Status page (row 2, col 7; SPOTIFY now shows inside SHOW) opens page 24. Amber until done, blinking red from Saturday noon; not part of ALL. Tap keys for BOH mix, mics and stage reset every Monday. - Holds fix what can be fixed: re-sync the lineup, put the cross back up, write LTC into PCO, copy tracks to locations, look again. Hold READY fixes everything. - Settings page: a READY page tab (live keys, drag to move, why each key is amber), and Apply re-imports page 24. - Auto-fix: "while_muted" fixes keep running with ALARMS muted (Spotify playlist). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
303 lines
17 KiB
Python
303 lines
17 KiB
Python
"""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", "readiness.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", "rc": "ready"}
|
|
ATTRS = {"self.cfg": "top", "engine.cfg": "top", "self.net": "network", "self.auto": "auto", "self.site": "site",
|
|
"self.atem": "atem", "self.ready_cfg": "ready",
|
|
"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",
|
|
"ready": "ready", "ready_page": "ready_page"}
|
|
SUB = {("auto", "windows"): "windows", ("site", "live_services"): "windows", ("tile", "checks"): "checks",
|
|
("restart_page", "keys"): "restart_keys", ("tile", "members"): "names", ("ready", "due"): "windows",
|
|
("ready_page", "keys"): "tiles"}
|
|
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 isinstance(node.func, ast.Name) and node.func.id in ("all_tiles", "ready_keys"):
|
|
return "tiles" # the Status page's tiles and the READY page's keys
|
|
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[].", "ready": "ready.", "ready_page": "ready_page."}.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", "gear.*.kind", "gear.*.address", "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",
|
|
"ready_page", "ready_page.keys", "ready"}
|
|
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, "ready.due[].%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"),
|
|
(make_page.build_ready, "ready-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()
|