christianandClaude Opus 5.5
5adab6c518
- ON/OFF on every key, check, fix, Restart and READY key, computer, MIDI link, HyperDeck, Dante device, watched Companion connection, switcher and Spotify check, outlet fix, and one main switch each for auto-fix, the auto switch and the READY page. Switched off = kept in the settings, grey OFF on the deck, never checked or run (effective_config in the watchdog). - Settings › Auto switch: a plain summary, what counts as down (each trigger editable in place: address, program and computer, Dante device, the button it presses, timings), the hours folded away, the buttons; + Button, + What counts as down, Remove. - Pick a Companion button by clicking it on its page, drawn like Companion (read-only API: /api/companion/pages and /api/companion/page). - STREAM STB / STREAM FOH Restart keys removed (the Studio B page has its own); the alert names the switch-back button instead. - Dante screen: shows the DANTE key's device list again (it picked STB AUDIO's one device). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
307 lines
17 KiB
Python
307 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.failover_cfg": "failover",
|
|
"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", "failover": "failover"}
|
|
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", ("failover", "windows"): "windows"}
|
|
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 == "effective_config":
|
|
return "top" # the config with switched-off keys and checks taken out
|
|
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.",
|
|
"failover": "failover."}.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,
|
|
"failover.windows[].%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()
|