status-indicators/tests/test_apply.py

"""Saving a change, offline: the shared draft, what Apply will do, the pipeline against a temp data
dir with a fake launchctl and a fake Companion, rollback when the watchdog does not come back, the
recording guard with its Emergency override, history and Go back."""
import json
import os
import sys
import tempfile
import threading
import time
import unittest
from unittest import mock

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import status_panel as sp  # noqa: E402
from settings import apply as ap  # noqa: E402
from settings import server  # noqa: E402
from test_status_panel import CONFIG  # noqa: E402


class FakeLaunchctl:
    """Records calls; a kickstart or bootstrap "starts" the watchdog by writing started.json
    (with the sha of the config on disk) unless told to fail."""

    def __init__(self, data_dir, fail=False):
        self.data_dir, self.fail, self.calls, self.pid = data_dir, fail, [], 100

    def __call__(self, argv, **kw):
        self.calls.append(argv)

        class R:
            returncode, stdout, stderr = 0, "", ""
        r = R()
        if argv[1] == "print":
            r.stdout = "state = running\n\tpid = %d\n" % self.pid
        elif argv[1] in ("kickstart", "bootstrap"):
            if not self.fail:
                self.pid += 1
                cfg_sha = ap.sha(open(os.path.join(self.data_dir, "config.json"), "rb").read())
                sp.write_json(os.path.join(self.data_dir, "started.json"), {"pid": self.pid, "started_at": time.time(), "config_sha": cfg_sha, "version": "test"})
        return r


class TestDraftAndPlan(unittest.TestCase):
    def setUp(self):
        self.data = tempfile.mkdtemp(prefix="sp-apply-")
        self.cfg = json.loads(json.dumps(CONFIG))
        self.cfg["companion"] = {"host": "127.0.0.1", "port": 1}
        with open(os.path.join(self.data, "config.json"), "wb") as f:
            f.write(ap.config_text(self.cfg))

    def test_draft_is_shared_and_never_overwritten_silently(self):
        d = ap.Draft(self.data)
        a, _ = d.update({"x": 1}, "sam", "sha1")
        self.assertEqual(oct(os.stat(d.path).st_mode)[-3:], "600")
        b, conflict = d.update({"x": 2}, "chris", "sha1", expect=a["version"])
        self.assertIsNone(conflict)
        _, conflict = d.update({"x": 3}, "sam", "sha1", expect=a["version"])      # sam's tab is behind
        self.assertEqual(conflict["by"], "chris")
        self.assertEqual(d.read()["config"], {"x": 2})

    def test_what_apply_does(self):
        hashes = ap.build_hashes(self.cfg)
        same = json.loads(json.dumps(self.cfg))
        self.assertEqual(ap.what_apply_does(self.cfg, same, hashes)["changes"], [])
        notes = json.loads(json.dumps(self.cfg))
        notes["gear"] = {"UHDJr": {"location": "lobby rack"}}
        plan = ap.what_apply_does(self.cfg, notes, hashes)
        self.assertFalse(plan["restart"])                                        # page-only data
        self.assertEqual(plan["pages"], [])
        timing = json.loads(json.dumps(self.cfg))
        timing["debounce_seconds"] = 4
        self.assertTrue(ap.what_apply_does(self.cfg, timing, hashes)["restart"])
        self.assertEqual(ap.what_apply_does(self.cfg, timing, hashes)["pages"], [])
        moved = json.loads(json.dumps(self.cfg))
        next(t for t in moved["tiles"] if t["id"] == "led")["pos"] = [2, 8]
        plan = ap.what_apply_does(self.cfg, moved, hashes)
        self.assertEqual(plan["pages"], ["status"])
        self.assertTrue(plan["nav"])
        label = json.loads(json.dumps(self.cfg))
        label["fixes"]["vista"]["label"] = "start Vista please"                  # only a key note changes
        self.assertEqual(ap.what_apply_does(self.cfg, label, hashes)["pages"], [])

    def test_words(self):
        after = json.loads(json.dumps(self.cfg))
        hd2 = next(t for t in after["tiles"] if t["id"] == "hd2")
        hd2["checks"][0]["optional"] = False
        after["site"]["editing_code"] = "new words"
        after["hosts"]["lighting"]["notes"] = "keep it Public"
        words = [ap.words_for(p, a, b, self.cfg, after) for p, a, b in ap.changed_paths(self.cfg, after)]
        texts = [w["text"] for w in words]
        self.assertIn("key HD2 › check 1 › optional: yes → no", texts)
        self.assertIn("computer Lighting PC › notes changed", texts)                    # long values fold open
        note = next(w for w in words if w["text"] == "computer Lighting PC › notes changed")
        self.assertEqual((note["before"][:8], note["after"]), ("Keep the", "keep it Public"))
        self.assertEqual(ap.plain_value([{"days": ["sun"], "start": "06:30", "end": "13:30"}, {"days": ["wed"], "start": "16:00", "end": "21:00"}]), "Sun 6:30 am – 1:30 pm; Wed 4 pm – 9 pm")
        self.assertTrue(any(t == "site › editing code changed" for t in texts), texts)   # a secret: never its value
        more = json.loads(json.dumps(self.cfg))
        midi = next(t for t in more["tiles"] if t["id"] == "midi")
        midi["checks"].append({"type": "midi_peer", "session": "NA Tracks", "peer": "Studio B", "tag": "STUDIO B"})
        texts = [ap.words_for(p, a, b, self.cfg, more)["text"] for p, a, b in ap.changed_paths(self.cfg, more)]
        self.assertEqual(len(texts), 1)
        self.assertTrue(texts[0].startswith("key MIDI › check %d added: " % (len(midi["checks"]))), texts)


class TestRebase(unittest.TestCase):
    def test_set_path_and_rebase(self):
        base = json.loads(json.dumps(CONFIG))
        live = json.loads(json.dumps(base)); live["debounce_seconds"] = 7                            # moved on since the draft started
        live["fixes"]["vista"]["label"] = "start Vista (new words)"
        draft = {"config": json.loads(json.dumps(base))}
        draft["config"]["auto_fix"]["windows"].append({"days": ["wed"], "start": "16:00", "end": "21:00"})   # the draft's own change
        next(t for t in draft["config"]["tiles"] if t["id"] == "hd2")["checks"][0]["optional"] = False
        draft["config"]["hosts"]["lighting"]["notes"] = "mine"
        merged, changes = ap.rebase(draft, live, base)
        self.assertEqual(merged["debounce_seconds"], 7)                                          # the live change survives
        self.assertEqual(merged["fixes"]["vista"]["label"], "start Vista (new words)")
        self.assertEqual(len(merged["auto_fix"]["windows"]), 2)                                  # and the draft's own
        self.assertFalse(next(t for t in merged["tiles"] if t["id"] == "hd2")["checks"][0]["optional"])
        self.assertEqual(merged["hosts"]["lighting"]["notes"], "mine")
        self.assertEqual(len(changes), 3)
        cfg = json.loads(json.dumps(base))
        ap.set_path(cfg, "tiles[hd2]", None); self.assertFalse(any(t["id"] == "hd2" for t in cfg["tiles"]))
        ap.set_path(cfg, "tiles[hd9]", {"id": "hd9", "label": "HD9"}); self.assertTrue(any(t["id"] == "hd9" for t in cfg["tiles"]))
        ap.set_path(cfg, "hosts.newpc.ssh", "a@b"); self.assertEqual(cfg["hosts"]["newpc"]["ssh"], "a@b")


class TestPipeline(unittest.TestCase):
    def setUp(self):
        self.data = tempfile.mkdtemp(prefix="sp-pipe-")
        self.cfg = json.loads(json.dumps(CONFIG))
        self.cfg["companion"] = {"host": "127.0.0.1", "port": 1}
        with open(os.path.join(self.data, "config.json"), "wb") as f:
            f.write(ap.config_text(self.cfg))
        open(os.path.join(self.data, "atem-reference.xml"), "w").write("<x/>")
        self.site = server.Site(self.data, companion=mock.Mock())
        self.launchctl = FakeLaunchctl(self.data)
        self.applier = ap.Applier(self.site, run=self.launchctl, sleep=lambda s: None)

    def run_apply(self, new_cfg, note="t", emergency=None):
        r = self.applier.start(new_cfg, note, "sam", "10.1.1.50", emergency)
        if r.get("ok"):
            self.applier.thread.join(30)
        return r

    def test_a_timing_change_restarts_and_is_kept_in_history(self):
        new = json.loads(json.dumps(self.cfg))
        new["debounce_seconds"] = 4
        r = self.run_apply(new, "slower")
        self.assertTrue(r["ok"])
        st = self.applier.status
        self.assertEqual(st["state"], "done", st)
        names = {s["name"]: s["state"] for s in st["steps"]}
        self.assertEqual((names["check"], names["save"], names["restart"], names["pages"]), ("ok", "ok", "ok", "skipped"))
        self.assertEqual(json.load(open(os.path.join(self.data, "config.json")))["debounce_seconds"], 4)
        self.assertTrue(any(a[1] == "kickstart" for a in self.launchctl.calls))
        entries = ap.history_entries(self.data)
        self.assertEqual((entries[0]["by"], entries[0]["note"], entries[0]["actions"]), ("sam", "slower", ["restart"]))
        self.assertEqual(ap.history_config(self.data, entries[0]["id"])["debounce_seconds"], 3)   # the version before
        self.assertEqual(oct(os.stat(os.path.join(self.data, "history", entries[0]["id"] + ".json")).st_mode)[-3:], "600")

    def test_page_only_data_needs_no_restart(self):
        new = json.loads(json.dumps(self.cfg))
        new["gear"] = {"UHDJr": {"location": "lobby"}}
        self.run_apply(new)
        names = {s["name"]: s["state"] for s in self.applier.status["steps"]}
        self.assertEqual(names["restart"], "skipped")
        self.assertFalse(any(a[1] == "kickstart" for a in self.launchctl.calls))

    def test_a_bad_config_is_refused_before_anything_happens(self):
        new = json.loads(json.dumps(self.cfg))
        new["fixes"]["vista"]["host"] = "nowhere"
        self.run_apply(new)
        self.assertEqual(self.applier.status["state"], "failed")
        self.assertIn("unknown computer nowhere", self.applier.status["outcome"])
        self.assertEqual(json.load(open(os.path.join(self.data, "config.json")))["fixes"]["vista"]["host"], "lighting")
        self.assertEqual(ap.history_entries(self.data), [])

    def test_rollback_when_the_watchdog_does_not_come_back(self):
        self.launchctl.fail = True
        new = json.loads(json.dumps(self.cfg))
        new["debounce_seconds"] = 5
        self.run_apply(new, "will fail")
        st = self.applier.status
        self.assertEqual(st["state"], "failed")
        self.assertIn("previous settings", st["outcome"].lower())
        self.assertEqual(json.load(open(os.path.join(self.data, "config.json")))["debounce_seconds"], 3)      # put back
        self.assertTrue(os.path.exists(os.path.join(self.data, "last-apply-failed.json")))
        meta = ap.history_entries(self.data)[0]
        self.assertIn("rolled back", meta["outcome"])

    def test_layout_change_imports_the_page_and_recolours_the_nav_keys(self):
        new = json.loads(json.dumps(self.cfg))
        next(t for t in new["tiles"] if t["id"] == "led")["pos"] = [2, 8]
        pushed, coloured, calls = [], [], {"n": 0}

        def fetch(h, p, page):                      # before the push: only page keys; after: as built
            calls["n"] += 1
            return {"page": {"name": "19\\nStatus", "controls": {}}} if calls["n"] % 2 == 1 else ap.make_page.build(new)
        with mock.patch.object(ap.push_page, "fetch_page", fetch), \
                mock.patch.object(ap.push_page, "push", lambda *a: pushed.append(a[2])), \
                mock.patch.object(ap.nav_keys, "colour", lambda *a: coloured.append(a) or 96):
            self.run_apply(new, "move LED")
        st = self.applier.status
        self.assertEqual(st["state"], "done", st)
        self.assertEqual(pushed, [19])
        self.assertEqual(len(coloured), 1)
        self.assertEqual(json.load(open(os.path.join(self.data, "build-hashes.json")))["status"], ap.build_hashes(new)["status"])

    def test_a_foreign_key_on_the_page_stops_the_import_and_rolls_back(self):
        new = json.loads(json.dumps(self.cfg))
        next(t for t in new["tiles"] if t["id"] == "led")["pos"] = [2, 8]
        busy = {"page": {"name": "19\\nStatus", "controls": {"2": {"3": {"type": "button-layered", "options": {"notes": "someone's key"}}}}}}
        with mock.patch.object(ap.push_page, "fetch_page", lambda h, p, page: busy), mock.patch.object(ap.push_page, "push", lambda *a: self.fail("must not push")):
            self.run_apply(new, "move LED")
        st = self.applier.status
        self.assertEqual(st["state"], "failed")
        self.assertIn("row 2 col 3", st["outcome"])
        self.assertEqual(next(t for t in json.load(open(os.path.join(self.data, "config.json")))["tiles"] if t["id"] == "led")["pos"], [3, 6])
        self.assertTrue(os.path.exists(os.path.join(self.data, "pages-out-of-sync.json")))

    def test_live_refuses_unless_emergency(self):
        sp.write_json(os.path.join(self.data, "live-now.json"), {"live": True, "reason": "a HyperDeck is recording", "time": time.time()})
        new = json.loads(json.dumps(self.cfg))
        new["debounce_seconds"] = 4
        r = self.run_apply(new)
        self.assertFalse(r["ok"])
        self.assertTrue(r.get("live"))
        r = self.run_apply(new, "urgent", {"word": "LIVE", "reason": "the deck is wrong"})
        self.assertTrue(r["ok"])
        self.assertEqual(self.applier.status["state"], "done")
        self.assertIn("EMERGENCY apply by sam", open(os.path.join(self.data, "apply.log")).read())
        sp.write_json(os.path.join(self.data, "live-now.json"), {"live": True, "reason": "stale", "time": time.time() - 120})
        self.assertEqual(self.applier.live(), (False, ""))                           # stale: the schedule (none) decides

    def test_a_new_switcher_file_goes_with_the_config_and_is_kept(self):
        pending = os.path.join(self.data, "draft-atem.xml")
        open(pending, "w").write("<y/>")
        new = json.loads(json.dumps(self.cfg))
        r = self.applier.start(new, "new switcher file", "sam", "10.1.1.50", None, pending)
        self.applier.thread.join(30)
        self.assertEqual(open(os.path.join(self.data, "atem-reference.xml")).read(), "<y/>")
        self.assertTrue(os.listdir(os.path.join(self.data, "history", "atem")))      # the old one, dated
        self.assertFalse(os.path.exists(pending))
        self.assertIn("switcher file", ap.history_entries(self.data)[0]["actions"])

    def test_one_apply_at_a_time(self):
        gate = threading.Event()
        slow = ap.Applier(self.site, run=self.launchctl, sleep=lambda s: gate.wait(2))
        new = json.loads(json.dumps(self.cfg))
        new["debounce_seconds"] = 4
        self.assertTrue(slow.start(new, "a", "sam", "x")["ok"])
        self.assertIn("already running", slow.start(new, "b", "sam", "x")["error"])
        gate.set()
        slow.thread.join(30)


if __name__ == "__main__":
    unittest.main()