status-indicators/tests/test_settings.py

"""The settings page's server and probes, offline: routes against the fake Companion, the LAN
binding, the editing-code gate, the probe parsers on captured output, and the teaching pages."""
import json
import os
import socket
import sys
import tempfile
import threading
import time
import unittest
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer

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 keyguide, probes, server  # noqa: E402
from test_status_panel import CONFIG, FakeCompanion, healthy_snapshot  # noqa: E402


def get(port, path, json_body=None, origin=None, method=None, code=None, who=None):
    data = json.dumps(json_body).encode() if json_body is not None else None
    headers = {"Content-Type": "application/json", **({"Origin": origin} if origin else {}),
               **({"X-Edit-Key": code} if code else {}), **({"X-Editor": who} if who else {})}
    req = urllib.request.Request("http://127.0.0.1:%d%s" % (port, path), data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            body = r.read()
            return r.status, body
    except urllib.error.HTTPError as e:
        return e.code, e.read()


class TestServer(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        snap = healthy_snapshot(time.time())
        FakeCompanion.store = {"conns": snap.conns, "vars": dict(snap.vars),
                               "custom": {"sp_state": json.dumps({"all": {"x": "ALL OK", "b": 1, "f": 2, "b2": 1, "f2": 2, "s": 20}}),
                                          "sp_hb": str(int(time.time())), "sp_cmd": "", "flash_tick": "0"}, "restarts": []}
        cls.fake = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion)
        threading.Thread(target=cls.fake.serve_forever, daemon=True).start()
        cls.data = tempfile.mkdtemp(prefix="sp-settings-")
        cfg = json.loads(json.dumps(CONFIG))
        cfg["companion"] = {"host": "127.0.0.1", "port": cls.fake.server_address[1]}
        cfg["site"]["editing_code"] = "apple river stone"
        json.dump(cfg, open(os.path.join(cls.data, "config.json"), "w"))
        now = time.time()
        json.dump({"pid": 1, "started_at": now - 100, "config_sha": "x", "version": "test"}, open(os.path.join(cls.data, "started.json"), "w"))
        json.dump({"live": True, "reason": "a HyperDeck is recording", "time": now}, open(os.path.join(cls.data, "live-now.json"), "w"))
        json.dump({"time": now, "conns": {}, "faults": {"atem": [{"check": "ATEM", "type": "conn", "sub": "conn", "level": "err",
                                                                   "msg": "NO ANSWER", "since": now - 60, "acked": False}],
                                                        "midi": [{"check": "tracks/NA Tracks/Lighting PC", "type": "midi_peer", "sub": "peer",
                                                                  "level": "err", "msg": "FX LIGHTS MIDI LOST", "since": now - 60, "acked": False}]},
                   "midi": {"NA Ingest": {"peers": [{"name": "Lighting PC", "state": "running"}]}}, "remote_midi": {},
                   "http": {"http://10.6.7.13:8080/": True}, "ping": {"10.1.3.227": 0}, "service_on": True, "dante_map": {}},
                  open(os.path.join(cls.data, "snapshot.json"), "w"))
        sock = socket.socket(); sock.bind(("127.0.0.1", 0)); cls.port = sock.getsockname()[1]; sock.close()
        cls.site, cls.servers = server.serve(cls.data, cls.port, ["127.0.0.1"])

    @classmethod
    def tearDownClass(cls):
        for s in cls.servers:
            s.shutdown()
        cls.fake.shutdown()

    def test_reads_are_open_and_secrets_are_masked(self):
        status, body = get(self.port, "/api/config")
        self.assertEqual(status, 200)
        cfg = json.loads(body)["config"]
        self.assertEqual(cfg["site"]["editing_code"], "set (17 characters)")
        self.assertNotIn("apple river stone", body.decode())

    def test_overview_shows_the_deck_the_problems_and_the_live_state(self):
        status, body = get(self.port, "/api/overview")
        o = json.loads(body)
        self.assertEqual(o["deck"]["all"]["x"], "ALL OK")
        self.assertTrue(o["companion_ok"])
        self.assertEqual([p["text"] for p in o["problems"]], ["ATEM: NO ANSWER", "MIDI: FX LIGHTS MIDI LOST"])
        words = {c["alias"]: c["word"] for c in o["computers"]}
        self.assertEqual((words["tracks"], words["lighting"], words["prop"], words["ingest"]), ("Problem", "Problem", "Not looked yet", "OK"))
        self.assertIn("hold the ATEM key", o["problems"][0]["action"])
        self.assertTrue(o["live"]["live"])
        self.assertEqual(len(o["computers"]), len(CONFIG["hosts"]))
        self.assertEqual(o["watchdog"]["version"], "test")

    def test_companion_view_lists_connections_with_their_watchers(self):
        o = json.loads(get(self.port, "/api/companion")[1])
        by = {c["label"]: c for c in o["connections"]}
        self.assertIn("hd2", by["HyperDeck_2"]["watched_by"])
        self.assertTrue(o["lights"]["variable sp_state"])
        self.assertEqual(o["missing"], [])

    def test_machines_use_what_the_watchdog_sees(self):
        o = json.loads(get(self.port, "/api/machines")[1])
        ingest = next(c for c in o["computers"] if c["self"])
        self.assertEqual(ingest["word"], "OK")
        bible = next(c for c in o["computers"] if c["alias"] == "bibletv")
        web = next(i for i in bible["items"] if i["kind"] == "web")
        self.assertEqual(web["light"], "ok")                                  # from snapshot.json, not a probe
        self.assertIn("mac", o["setup_steps"])

    def test_gear_and_keys_and_pages(self):
        gear = json.loads(get(self.port, "/api/gear")[1])
        self.assertTrue(any(r["name"] == "HyperDeck_2" and "hd2" in r["watched_by"] for r in gear))
        keys = json.loads(get(self.port, "/api/keys")[1])
        hd2 = next(t for t in keys["tiles"] if t["id"] == "hd2")
        self.assertIn("NO DISK", hd2["says"])
        self.assertIn("off and on", hd2["holding"])
        self.assertEqual(keys["hold"], "HOLD 2s")
        for page in ("/key-guide.html", "/cheat.html", "/where.html", "/", "/app.js", "/style.css"):
            status, body = get(self.port, page)
            self.assertEqual(status, 200, page)
        guide = get(self.port, "/key-guide.html")[1].decode()
        for t in CONFIG["tiles"]:
            self.assertIn(t["label"], guide)
        self.assertNotIn("?", keyguide.cheat_sheet(CONFIG, "http://x/").split("<h2>Where to look")[0].split("</h1>")[1][:200])

    def test_logs_are_whitelisted(self):
        self.assertEqual(get(self.port, "/api/logs/nope")[0], 404)
        self.assertEqual(get(self.port, "/api/logs/watchdog")[0], 200)
        self.assertEqual(json.loads(get(self.port, "/api/logs/autofix")[1])["file"], None)

    def test_editing_code_gate_and_lockout(self):
        self.site.attempts.clear()
        self.assertFalse(json.loads(get(self.port, "/api/editing/check", {"code": "nope"})[1])["ok"])
        self.assertTrue(json.loads(get(self.port, "/api/editing/check", {"code": "Apple River Stone"})[1])["ok"])
        for _ in range(5):
            get(self.port, "/api/editing/check", {"code": "wrong"})
        r = json.loads(get(self.port, "/api/editing/check", {"code": "apple river stone"})[1])
        self.assertFalse(r["ok"])
        self.assertIn("locked_for", r)
        self.site.attempts.clear()
        self.assertEqual(get(self.port, "/api/editing/check", {"code": "x"}, origin="http://evil.example")[0], 403)

    # ---- step 2: the write routes ----
    CODE = "apple river stone"

    def put(self, path, body, method="PUT", who="sam", code=None):
        st, raw = get(self.port, path, body, method=method, code=code or self.CODE, who=who)
        return st, json.loads(raw)

    def test_writes_need_the_code_and_a_name(self):
        self.site.attempts.clear()
        self.assertEqual(get(self.port, "/api/draft", {"config": {}}, method="PUT")[0], 401)
        self.assertEqual(get(self.port, "/api/draft", {"config": {}}, method="PUT", code="wrong", who="sam")[0], 401)
        self.assertEqual(get(self.port, "/api/draft", {"config": {}}, method="PUT", code=self.CODE)[0], 400)      # no name
        self.site.attempts.clear()

    def test_draft_round_trip_conflict_and_discard(self):
        self.site.attempts.clear()
        live = self.site.load()
        mine = json.loads(json.dumps(live)); mine["debounce_seconds"] = 4
        st, r = self.put("/api/draft", {"config": mine})
        self.assertEqual(st, 200)
        v1 = r["draft"]["version"]
        self.assertEqual(r["draft"]["config"]["debounce_seconds"], 4)
        self.assertTrue(r["draft"]["config"]["site"]["editing_code"].startswith("set ("))             # masked on the way out
        self.assertEqual(json.loads(get(self.port, "/api/draft")[1])["draft"]["by"], "sam")
        theirs = json.loads(json.dumps(live)); theirs["debounce_seconds"] = 5
        st, r = self.put("/api/draft", {"config": theirs, "expect": v1}, who="chris")
        self.assertEqual(st, 200)
        st, r = self.put("/api/draft", {"config": mine, "expect": v1})                           # sam's tab is behind
        self.assertEqual(st, 409)
        self.assertIn("chris", r["error"])
        self.assertEqual(r["draft"]["config"]["debounce_seconds"], 5)
        self.assertEqual(self.site.draft.read()["config"]["site"]["editing_code"], self.CODE)   # the secret survived the masked round trip
        st, r = self.put("/api/validate", {})
        self.assertEqual(st, 200)
        self.assertTrue(any("debounce seconds: %d → 5" % live["debounce_seconds"] in c["text"] for c in r["changes"]), r["changes"])
        self.assertIn("purple", r["feel"].lower())
        self.assertTrue(r["live"]["live"])
        self.assertEqual(self.put("/api/draft", {}, method="DELETE")[0], 200)
        self.assertIsNone(json.loads(get(self.port, "/api/draft")[1])["draft"])

    def test_apply_waits_for_live_then_runs_restores_and_undoes(self):
        from test_apply import FakeLaunchctl
        self.site.attempts.clear()
        fake = FakeLaunchctl(self.data)
        self.site.applier.run, self.site.applier.sleep = fake, lambda s: None
        live = self.site.load()
        mine = json.loads(json.dumps(live)); mine["debounce_seconds"] = 6
        self.put("/api/draft", {"config": mine})
        st, r = self.put("/api/apply", {"note": "slower"}, method="POST")
        self.assertEqual(st, 409)
        self.assertTrue(r.get("live"))
        self.assertIsNotNone(self.site.draft.read())                                            # the draft is kept
        st, r = self.put("/api/apply", {"note": "slower", "emergency": {"word": "LIVE", "reason": "deck wrong"}}, method="POST")
        self.assertEqual(st, 200, r)
        self.site.applier.thread.join(30)
        self.assertIsNone(self.site.draft.read())
        status = json.loads(get(self.port, "/api/apply/status")[1])
        self.assertEqual(status["state"], "done", status)
        self.assertEqual(self.site.load()["debounce_seconds"], 6)
        hist = json.loads(get(self.port, "/api/history")[1])["entries"]
        self.assertEqual((hist[0]["by"], hist[0]["note"]), ("sam", "slower"))
        detail = json.loads(get(self.port, "/api/history/" + hist[0]["id"])[1])
        self.assertTrue(any("debounce seconds: %d → 6" % live["debounce_seconds"] in c["text"] for c in detail["changes_to_now"]), detail)
        st, r = self.put("/api/restore/" + hist[0]["id"], {}, method="POST")
        self.assertEqual(r["draft"]["config"]["debounce_seconds"], live["debounce_seconds"])
        self.put("/api/draft", {}, method="DELETE")
        st, r = self.put("/api/undo", {}, method="POST")
        self.assertEqual((st, r["draft"]["config"]["debounce_seconds"], r["note"]), (200, live["debounce_seconds"], "slower"))
        self.put("/api/draft", {}, method="DELETE")
        # watchdog control waits for live too; a command goes to the watchdog through Companion
        self.assertEqual(self.put("/api/watchdog/restart", {}, method="POST")[0], 409)
        st, r = self.put("/api/command", {"text": "service off"}, method="POST")
        self.assertEqual((st, r["sent"]), (200, "service off"))
        self.assertIn("service off", FakeCompanion.store["custom"]["sp_cmd"])
        self.assertEqual(self.put("/api/command", {"text": "rm -rf"}, method="POST")[0], 400)
        self.assertEqual(self.put("/api/command", {"text": "test hd2 vista 0"}, method="POST")[0], 409)   # no test while live

    def test_setup_bundle_import_and_install_routes(self):
        import io, zipfile
        from test_apply import FakeLaunchctl  # noqa: F401
        self.site.attempts.clear()
        st, raw = get(self.port, "/api/setup")
        comps = json.loads(raw)["computers"]
        self.assertEqual(st, 200)
        self.assertEqual({c["alias"] for c in comps}, set(self.site.load()["hosts"]))
        self.assertTrue(any(s["name"].startswith("Set-up script") for c in comps for s in c["scripts"]))
        st, raw = get(self.port, "/api/site/bundle")
        self.assertEqual(st, 200)
        z = zipfile.ZipFile(io.BytesIO(raw))
        self.assertIn("config.json", z.namelist())
        bundle_cfg = json.loads(z.read("config.json"))
        self.assertEqual(bundle_cfg["site"]["editing_code"], self.CODE)                     # the bundle is the real file
        # a new campus from that bundle: this campus's own site block stays, computers are flagged
        other = json.loads(json.dumps(bundle_cfg)); other["site"]["name"] = "Woonsocket"; other["site"]["editing_code"] = "theirs"
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w") as z2:
            z2.writestr("config.json", json.dumps(other))
        req = urllib.request.Request("http://127.0.0.1:%d/api/site/import" % self.port, data=buf.getvalue(),
                                     headers={"Content-Type": "application/octet-stream", "X-Edit-Key": self.CODE, "X-Editor": "sam"}, method="POST")
        with urllib.request.urlopen(req, timeout=10) as r:
            body = json.loads(r.read())
        self.assertEqual(body["from"], "Woonsocket")
        d = self.site.draft.read()
        self.assertEqual(d["config"]["site"]["name"], self.site.load()["site"]["name"])
        self.assertEqual(d["config"]["site"]["editing_code"], self.CODE)
        self.assertTrue(all(h["notes"].startswith("copied from Woonsocket - check it") for h in d["config"]["hosts"].values()))
        self.assertEqual(body["errors"], [])
        self.put("/api/draft", {}, method="DELETE")
        self.assertEqual(self.put("/api/site/import", {"nope": 1}, method="POST")[0], 400)
        # the program: staged folder info is readable; an install is refused while live
        st, raw = get(self.port, "/api/program/staged")
        self.assertEqual(st, 200)
        self.assertIn("running_version", json.loads(raw))
        self.assertEqual(self.put("/api/program/install", {}, method="POST")[0], 409)

    def test_names_check_lists_what_this_companion_lacks(self):
        self.site.attempts.clear()
        n = json.loads(get(self.port, "/api/site/names")[1])
        self.assertTrue(n["companion_ok"])
        live = self.site.load()
        mine = json.loads(json.dumps(live))
        for t in mine["tiles"]:
            for c in t.get("checks", []):
                if c.get("conn") == "HyperDeck_2":
                    c["conn"] = "HyperDeck_Z"
        self.put("/api/draft", {"config": mine})
        n = json.loads(get(self.port, "/api/site/names")[1])
        gone = [c for c in n["connections"] if not c["live"]]
        self.assertEqual([(c["name"], c["keys"]) for c in gone], [("HyperDeck_Z", ["HD2"])])
        self.assertIn("HyperDeck_2", gone[0]["candidates"])                    # same module, unused
        self.assertNotIn("HyperDeck_1", gone[0]["candidates"])                 # used by HD1
        self.put("/api/draft", {}, method="DELETE")

    def test_a_draft_is_rebased_when_the_live_settings_move_on(self):
        from test_apply import FakeLaunchctl
        self.site.attempts.clear()
        fake = FakeLaunchctl(self.data); self.site.applier.run, self.site.applier.sleep = fake, lambda s: None
        live = self.site.load()
        mine = json.loads(json.dumps(live)); mine["hosts"]["lighting"]["notes"] = "sam's note"
        self.put("/api/draft", {"config": mine}, who="sam")                                   # sam's draft
        d = self.site.draft.read()
        self.site.draft.clear()
        other = json.loads(json.dumps(live)); other["debounce_seconds"] = (live["debounce_seconds"] or 3) + 1   # chris applies something else
        self.put("/api/draft", {"config": other}, who="chris")
        self.put("/api/apply", {"note": "chris", "emergency": {"word": "LIVE", "reason": "test"}}, method="POST", who="chris")
        self.site.applier.thread.join(30)
        sp.write_json(self.site.draft.path, d)                                                  # sam's draft is back, based on the old settings
        st, raw = get(self.port, "/api/draft")
        r = json.loads(raw)
        self.assertFalse(r["stale"])
        self.assertEqual(r["draft"]["config"]["hosts"]["lighting"]["notes"], "sam's note")      # sam's own change
        self.assertEqual(r["draft"]["config"]["debounce_seconds"], other["debounce_seconds"])   # chris's change kept
        st, v = self.put("/api/validate", {})
        self.assertEqual(len(v["changes"]), 1); self.assertTrue(v["changes"][0]["text"].startswith("computer Lighting PC › notes"))
        self.put("/api/draft", {}, method="DELETE")

    def test_probe_is_refused_while_a_fix_runs(self):
        json.dump({"fix-x": {"key": "hd2"}}, open(os.path.join(self.data, "fixes-running.json"), "w"))
        try:
            self.assertEqual(get(self.port, "/api/machines/probe", {})[0], 409)
        finally:
            os.remove(os.path.join(self.data, "fixes-running.json"))


class TestBindingAndProbes(unittest.TestCase):
    def test_bind_addresses(self):
        ifaces = [("en0", "10.1.1.1", True), ("en1", "10.6.3.222", True), ("en17", "169.254.1.2", True), ("en20", "10.66.1.2", True),
                  ("en22", "10.104.40.6", True)]
        self.assertEqual(server.bind_addresses(CONFIG, ifaces), ["127.0.0.1", "10.1.1.1", "10.6.3.222"])   # the LAN ports only
        cfg = json.loads(json.dumps(CONFIG))
        cfg["hosts"]["ingest"]["networks"] = []
        self.assertEqual(server.bind_addresses(cfg, ifaces), ["127.0.0.1", "10.1.1.1", "10.6.3.222", "169.254.1.2", "10.66.1.2", "10.104.40.6"])
        self.assertTrue(server.allowed_client("192.168.4.7"))
        self.assertFalse(server.allowed_client("8.8.8.8"))

    def test_mac_probe_parsing(self):
        out = ("SP_NAME=NA-Tracks.localdomain\nSP_OSV=26.5.1\nSP_PY=Python 3.9.6\nSP_KEY=1\nSP_AGENTS\nPID\tStatus\tLabel\n"
               "780\t0\tapplication.com.audinate.DanteVirtualSoundcard.1.2\n-\t0\tcom.google.keystone.agent\nSP_PROCS\n"
               "/Applications/Ableton Live 11 Suite.app/Contents/MacOS/Live\n/usr/sbin/sshd\n/Library/Audio/dvsd\nSP_APPS\nAbleton Live 11 Suite.app\n"
               "SP_HELPER\n-rwxr-xr-x  1 root  wheel  375 Oct  8 12:05 /usr/local/sbin/status-panel-dvs-start\n/usr/local/sbin/status-panel-dvs-start\n"
               "SP_SOLO\n\nSP_NOW\n1791900000\nSP_END\n")
        parsed = probes.sections(out, ("SP_AGENTS", "SP_PROCS", "SP_APPS", "SP_HELPER", "SP_SOLO", "SP_NOW", "SP_END"))
        items, found = probes.mac_items(CONFIG["hosts"]["tracks"], parsed, CONFIG)
        by = {i["name"]: i for i in items}
        self.assertEqual(by["Ableton Live 11 Suite"]["light"], "ok")
        self.assertEqual(by["the DVS helper"]["light"], "ok")
        self.assertEqual(by["Dante Virtual Soundcard"]["light"], "ok")
        self.assertEqual(probes.values(out)["KEY"], "1")
        stopped = out.replace("/Contents/MacOS/Live\n", "")
        items, _ = probes.mac_items(CONFIG["hosts"]["tracks"], probes.sections(stopped, ("SP_AGENTS", "SP_PROCS", "SP_APPS", "SP_HELPER", "SP_SOLO", "SP_NOW", "SP_END")), CONFIG)
        self.assertEqual({i["name"]: i for i in items}["Ableton Live 11 Suite"]["reason"], "not running")

    def test_windows_probe_parsing(self):
        out = ('SP_NAME=LIGHTING-PC\r\nSP_TASKS\r\n"\\StatusPanel\\Start Arena","N/A","Running"\r\n"\\StatusPanel\\Start rtpMIDI","N/A","Ready"\r\n'
               '"\\StatusPanel\\Start Vista","N/A","Running"\r\nSP_PROCS\r\n"Vista.exe","1234","Console","1","100 K"\r\n"Arena.exe","1","Console","1","1 K"\r\n'
               'SP_SVC\r\nSERVICE_NAME: sshd\r\n        STATE              : 4  RUNNING\r\nSERVICE_NAME: rtpMIDIService\r\n        STATE              : 4  RUNNING\r\n'
               'SP_PS\r\nPublic\r\nEnabled\r\n10.1.1.1,10.6.3.222\r\nSP_END\r\n')
        parsed = probes.sections(out, ("SP_TASKS", "SP_PROCS", "SP_SVC", "SP_PS", "SP_END"))
        items, found, setup = probes.windows_items(CONFIG["hosts"]["lighting"], parsed)
        by = {i["name"]: i for i in items}
        self.assertEqual(by["Start Vista"]["light"], "ok")
        self.assertEqual(by["Start rtpMIDI"]["reason"], "rtpMIDI.exe is not running")
        self.assertEqual(by["rtpMIDIService"]["light"], "ok")
        self.assertEqual(by["sshd"]["light"], "ok")
        self.assertEqual((setup["profile"], setup["wake_on_lan"], setup["firewall_allow"]), ("Public", "Enabled", "10.1.1.1,10.6.3.222"))
        self.assertIn("\\StatusPanel\\Start Arena", found["tasks"])

    def test_probe_commands_never_carry_names_from_the_settings(self):
        cfg = json.loads(json.dumps(CONFIG))
        cfg["hosts"]["tracks"]["runs"][1]["process"] = "x; rm -rf ~"        # a hostile name in the settings
        cfg["site"]["dvs_helper"] = "/usr/local/sbin/x; rm -rf ~"            # and a hostile path
        line = probes.MAC_PROBE.format(**probes.quoted_paths(cfg))
        self.assertNotIn("rm -rf ~", line.replace("'/usr/local/sbin/x; rm -rf ~'", ""))   # only inside quotes
        self.assertNotIn("x; rm -rf ~\n", line)
        self.assertNotIn("Live", line)                                        # names are matched here, never sent
        self.assertIn('"$HOME"\'/Library/Application Support/status-panel\'', probes.quoted_paths(CONFIG)["solo"])


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