"""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", "ready.ltc_campuses[].name", "ready.ltc_campuses[].short", "ready.ltc_campuses[].service_type", "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 " + sp.windows_phrase(cfg["auto_fix"]["windows"])) # the campus's own hours self.assertEqual(schema.phrases(dict(cfg, auto_fix={"windows": [{"days": ["sun"], "start": "06:30", "end": "13:30"}]}))["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()