"""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("

Where to look")[0].split("

")[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_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()