status-indicators/tests/test_schema.py

"""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"), ("check", "repoint")}   # set by the watchdog on a fix's copy of a check


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", "discover", "discover.auto"}   # discover: settings/campus.py
        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()