"""Offline tests: no Companion, no gear. Run: python3 -m unittest discover -s tests""" import gzip import json import logging import os import re import sys import tempfile import threading import time import unittest import urllib.parse from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) import make_page # noqa: E402 import status_panel as sp # noqa: E402 sp.log.addHandler(logging.NullHandler()) sp.log.propagate = False CONFIG = sp.load_config(os.path.join(ROOT, "config.json")) # The tests pin their own schedule so they do not depend on the site's choice. CONFIG["service_windows"] = [{"days": ["sun"], "start": "07:00", "end": "12:30"}] CONFIG["tiles"] = [dict(t, label="SERVICE") if t["id"] == "service" else t for t in CONFIG["tiles"]] # A Sunday 08:00 (inside the service window) and a Wednesday 15:00 (outside it). SUNDAY = datetime(2026, 10, 11, 8, 0).timestamp() WEDNESDAY = datetime(2026, 10, 7, 15, 0).timestamp() def healthy_snapshot(now): """Everything in config.json present and fine.""" snap = sp.Snapshot() labels = {c["conn"] for t in CONFIG["tiles"] for c in t.get("checks", []) if c.get("conn")} snap.conns = {l: {"id": "id-" + l, "enabled": True, "category": "good", "message": None} for l in labels} snap.conns_time = now for i in range(1, 7): snap.vars[("HyperDeck_%d" % i, "ip")] = "10.1.10.1%d" % i snap.vars[("HyperDeck_%d" % i, "status")] = "Preview" snap.vars[("HyperDeck_%d" % i, "slotId")] = "1" snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "28:00:00" devices = set() for t in CONFIG["tiles"]: for c in t.get("checks", []): if c["type"] == "dante": devices |= set(c["devices"]) snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(sorted(devices)) snap.vars[("tascam", "error")] = "No Error" snap.vars[("tascam", "caution")] = "No Caution" for label, n in (("Axient_1", 4), ("Axient_2", 4), ("ULXD-4", 1)): for ch in range(1, n + 1): snap.vars[(label, "ch_%d_name" % ch)] = "VOX %d" % ch snap.vars[(label, "ch_%d_battery_bars" % ch)] = "5" snap.vars[(label, "ch_%d_interference_status" % ch)] = "NONE" for label in ("PSM_1", "PSM_2", "PSM_3"): for ch in (1, 2): snap.vars[(label, "ch_%d_name" % ch)] = "PACK" snap.vars[(label, "ch_%d_rf_mute" % ch)] = "UNMUTED" snap.midi = {"NA Ingest": {"peers": [{"name": "Lighting PC", "address": "", "state": "running"}, {"name": "SuperRack-LiveBox", "address": "", "state": "running"}]}} snap.midi_time = now snap.mdns, snap.mdns_time = {"NA Tracks", "Lighting PC", "NA Ingest"}, now snap.ping = {"10.1.3.227": 0, "10.6.7.13": 0, "10.1.0.1": 0} snap.link = True snap.launchd = {"org.waterschurch.resi-propresenter": 884, "org.waterschurch.lobby-captions": 900} snap.launchd_time = now snap.procs = {"/Applications/ProdDash.app/Contents/MacOS/ProdDash": True} snap.http = {"https://central.resi.io/": True, "http://10.6.7.13:8080/": True} snap.disk = {"/": 97} return snap def run_steps(engine, snap, start, seconds, step=1.0): tiles, t = None, start while t <= start + seconds: snap.conns_time = t tiles = engine.step(snap, t) t += step return tiles, t def level(tile): for name, ((b, f), _) in sp.COLORS.items(): if (tile["b"], tile["f"]) == (b, f): return name return "?" class TestChecks(unittest.TestCase): def setUp(self): self.now = SUNDAY self.snap = healthy_snapshot(self.now) self.ctx = {"now": self.now, "service": True, "any_recording": False} def test_all_green_when_healthy(self): tiles = sp.Engine(CONFIG).step(self.snap, self.now) bad = {k: v["x"] for k, v in tiles.items() if level(v) not in ("ok", "service", "off")} self.assertEqual(bad, {}) self.assertEqual(tiles["all"]["x"], "ALL OK") self.assertEqual(tiles["hd1"]["x"], "HD1\n28h") self.assertEqual(tiles["dante"]["x"], "DANTE\n19/19") def test_connection_states(self): c = self.snap.conns["ATEM"] c["category"], c["message"] = "error", "Network error: connect ETIMEDOUT 10.1.10.2:9910" self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "NO ANSWER") c["message"] = "connect EHOSTUNREACH" self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "UNREACHABLE") c["enabled"] = False self.assertEqual((sp.conn_result(self.snap, "ATEM", self.now).level, sp.conn_result(self.snap, "ATEM", self.now).msg), (sp.WARN, "DISABLED")) self.assertEqual(sp.conn_result(self.snap, "nope", self.now).msg, "NOT IN COMPANION") self.assertEqual(sp.conn_result(self.snap, "x32", self.now + 60).msg, "NO DATA") def test_hyperdeck_disk_and_record_sync(self): s = self.snap s.vars[("HyperDeck_2", "slotId")] = "-" s.vars[("HyperDeck_2", "recordingTime")] = "--:--:--" out = dict(sp.check_hyperdeck({"conn": "HyperDeck_2"}, s, self.ctx)) self.assertEqual(out["disk"].msg, "NO DISK") self.assertTrue(out["disk"].service_only) s.vars[("HyperDeck_3", "recordingTime")] = "0:20:00" self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].msg, "LOW 20m") s.vars[("HyperDeck_3", "recordingTime")] = "0:04:00" self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].level, sp.ERR) ctx = dict(self.ctx, any_recording=True) self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))["rec"].msg, "NOT REC") s.vars[("HyperDeck_1", "status")] = "Record" out = dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx)) self.assertEqual(out["rec"].level, sp.OK) self.assertEqual(out["conn"].info, "REC 28h") def test_optional_deck(self): for i in (2, 4): self.snap.vars[("HyperDeck_%d" % i, "slotId")] = "-" self.snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "--:--:--" for i in (1, 3, 5, 6): self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record" tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual((level(tiles["hd2"]), tiles["hd2"]["x"]), ("off", "HD2\nNO DISK")) self.assertEqual(level(tiles["rec"]), "ok") self.assertEqual(tiles["all"]["x"], "ALL OK") self.snap.vars[("HyperDeck_2", "slotId")] = "1" # disk goes in: now it is watched self.snap.vars[("HyperDeck_2", "recordingTime")] = "20:00:00" tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["hd2"]["x"], "HD2\nNOT REC") self.assertEqual(level(tiles["hd2"]), "err") def test_one_deck_not_recording_blinks(self): engine = sp.Engine(CONFIG) for i in (1, 2, 3, 5, 6): self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record" tiles, _ = run_steps(engine, self.snap, self.now, 4) self.assertEqual(tiles["hd4"]["x"], "HD4\nNOT REC") self.assertEqual(level(tiles["hd4"]), "err") self.assertEqual(tiles["all"]["x"], "1 DOWN\nHD4 NOT REC") self.assertEqual(tiles["rec"]["x"], "REC\nHD4 NOT REC") def test_dante(self): self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["NA-Tracks"]) out = dict(sp.check_dante({"conn": "audinate-dantecontroller", "devices": {"NA-Tracks": "DANTE", "BOH-WING": "WING"}}, self.snap, self.ctx)) self.assertEqual(out["BOH-WING"].msg, "WING GONE") self.assertEqual(out["NA-Tracks"].level, sp.OK) self.snap.vars[("audinate-dantecontroller", "devices")] = None out = dict(sp.check_dante({"conn": "audinate-dantecontroller", "devices": ["BOH-WING"]}, self.snap, self.ctx)) self.assertEqual(out["conn"].msg, "NO DATA") def test_dante_button_names_the_missing_device(self): devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) devices.remove("NA-Tracks") self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["dante"]["x"], "DANTE\nTRACKS GONE") self.assertEqual(level(tiles["dante"]), "err") self.assertEqual(tiles["all"]["x"], "1 DOWN\nDANTE TRACKS GONE") def test_shure_psm_tascam(self): s = self.snap s.vars[("Axient_1", "ch_1_battery_bars")] = "1" s.vars[("Axient_1", "ch_2_battery_bars")] = "2" s.vars[("Axient_1", "ch_3_battery_bars")] = "Unknown" s.vars[("Axient_1", "ch_4_interference_status")] = "DETECTED" s.vars[("Axient_1", "ch_3_interference_status")] = "DETECTED" # transmitter off: ignored out = dict(sp.check_shure({"conn": "Axient_1", "channels": 4}, s, self.ctx)) self.assertEqual((out["ch1_batt"].level, out["ch1_batt"].msg), (sp.ERR, "VOX 1 BATT")) self.assertEqual(out["ch2_batt"].level, sp.WARN) self.assertEqual(out["ch3_batt"].level, sp.OK) self.assertEqual(out["ch3_rf"].level, sp.OK) self.assertEqual(out["ch4_rf"].msg, "VOX 4 INTERF") s.vars[("PSM_2", "ch_1_rf_mute")] = "MUTED" self.assertEqual(dict(sp.check_psm({"conn": "PSM_2", "channels": 2}, s, self.ctx))["ch1_mute"].msg, "PACK RF MUTE") s.vars[("tascam", "error")] = "Media Error" self.assertEqual(dict(sp.check_tascam({"conn": "tascam"}, s, self.ctx))["error"].msg, "MEDIA ERROR") def test_rf_group_names_the_mic(self): self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "1" tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["rf"]["x"], "RF\nVOX 3 BATT") self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "5" self.snap.conns["PSM_1"]["category"] = "error" tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertIn("PSM 1 OFFLINE", tiles["rf"]["x"]) def test_bonjour_parsing_morning_and_afternoon(self): out = ("Browsing for _apple-midi._udp.local.\n" " 9:20:51.321 Add 3 13 local. _apple-midi._udp. NA Tracks\n" "19:20:51.321 Add 3 27 local. _apple-midi._udp. Lighting PC\n" "19:21:02.001 Add 3 27 local. _apple-midi._udp. Gone Soon\n" "19:21:09.001 Rmv 1 27 local. _apple-midi._udp. Gone Soon\n") self.assertEqual(sp.parse_dns_sd(out), {"NA Tracks", "Lighting PC"}) def test_vista_heartbeat(self): self.snap.artnet_since = self.now - 60 self.snap.artnet_last = {"10.1.10.206": self.now - 1} chk = {"source": "10.1.10.206", "max_silence_seconds": 5, "msg": "VISTA SILENT"} self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].level, sp.OK) self.snap.artnet_last = {"10.1.10.206": self.now - 8} self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].msg, "VISTA SILENT") self.snap.artnet_since = self.now - 2 # just started listening: no verdict self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].level, sp.OK) self.snap.artnet_since = None # could not listen: no verdict self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].level, sp.OK) def test_artnet_listener_hears_a_poll(self): import socket as _s snap, stop = sp.Snapshot(), threading.Event() probe = _s.socket(_s.AF_INET, _s.SOCK_DGRAM) probe.bind(("127.0.0.1", 0)) port = probe.getsockname()[1] probe.close() try: self.assertIsNotNone(sp.start_artnet_listener(snap, stop, port=port)) time.sleep(0.2) tx = _s.socket(_s.AF_INET, _s.SOCK_DGRAM) tx.sendto(b"Art-Net\x00\x00\x20\x00\x0e\x00\x00", ("127.0.0.1", port)) tx.sendto(b"not art-net", ("127.0.0.1", port)) time.sleep(0.6) self.assertIn("127.0.0.1", snap.artnet_last) finally: stop.set() def test_midi_peer_joining_is_not_an_error(self): for p in self.snap.midi["NA Ingest"]["peers"]: p["state"] = "connected" # joined, clock sync not finished yet tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 8) self.assertEqual(tiles["midi"]["x"], "MIDI\nOK") def test_midi_peer_and_bonjour(self): self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"] if p["name"] != "Lighting PC"] self.snap.mdns.discard("NA Tracks") tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["midi"]["x"], "MIDI\nLIGHT PC MIDI LOST\n+1") self.assertEqual(tiles["show"]["x"], "SHOW\nLIGHT PC MIDI LOST\n+1") # named, no "MIDI" prefix self.assertNotIn("MIDI TRACKS", tiles["show"]["x"]) self.snap.midi["NA Ingest"]["peers"].append({"name": "Lighting PC", "address": "", "state": "running"}) tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["midi"]["x"], "MIDI\nTRACKS MIDI GONE") def test_bible_tv_app_down(self): self.snap.http["http://10.6.7.13:8080/"] = False tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["apps"]["x"], "APPS\nBIBLE APP DOWN") def test_apps_names_the_app(self): self.snap.conns["Propresenter-API"]["category"] = "error" tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["apps"]["x"], "APPS\nPROPRES OFFLINE") self.assertEqual(tiles["all"]["x"], "1 DOWN\nPROPRES OFFLINE") def test_http_intervals(self): checks = dict(sp.http_checks(CONFIG)) self.assertEqual(checks["http://10.6.7.13:8080/"], 5) self.assertEqual(checks["https://central.resi.io/"], 30) def test_device_off_while_module_says_connected(self): # 2026-10-08: UHD Jr powered off, NovaStar module still "Connected". self.snap.ping["10.1.14.115"] = 2 tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["led"]["x"], "LED\nUHDJR NO PING") self.assertEqual(tiles["video"]["x"], "VIDEO\nUHDJR NO PING") self.snap.conns["UHDJr"]["category"] = "error" # and Companion's connection fails too tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["led"]["x"], "LED\nUHDJR OFFLINE\n+1") self.snap.conns["UHDJr"]["category"], self.snap.conns["UHDJr"]["enabled"] = "good", False self.snap.ping["10.1.14.115"] = 0 tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual((tiles["led"]["x"], level(tiles["led"])), ("LED\nUHDJR DISABLED", "warn")) def test_ingest_services(self): del self.snap.launchd["org.waterschurch.resi-propresenter"] self.snap.launchd["org.waterschurch.lobby-captions"] = None self.snap.disk["/"] = 6 tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) self.assertEqual(tiles["ingest"]["x"], "INGEST\nBRIDGE NOT LOADED\n+2") class TestTextSize(unittest.TestCase): def test_no_word_is_split(self): for text in ("TASCAM\nOK", "SERVICE\nON\nmanual", "OBS\nOFFLINE", "INGEST\n97GB free", "4 DOWN\nHD2 NO DISK", "REC\nHD4 NO DISK\n+1", "TRACKS\nMIDI SESS GONE"): size = sp.fit_size(text) longest = max(len(w) for w in text.split()) self.assertLessEqual(longest, int(sp.LINE_CAPACITY / size), text) def test_short_text_stays_big(self): self.assertGreaterEqual(sp.fit_size("HD1\n28h"), 29) self.assertGreater(sp.fit_size("ATEM\nOK"), sp.fit_size("TASCAM\nOK")) def test_every_tile_has_a_size(self): tiles = sp.Engine(CONFIG).step(healthy_snapshot(SUNDAY), SUNDAY) for t in tiles.values(): self.assertTrue(12 <= t["s"] <= 30, t) class TestBehaviour(unittest.TestCase): def setUp(self): self.snap = healthy_snapshot(SUNDAY) def test_debounce(self): engine = sp.Engine(CONFIG) self.snap.conns["x32"]["category"] = "error" tiles = engine.step(self.snap, SUNDAY) self.assertEqual(level(tiles["x32"]), "ok") # one bad reading is not enough tiles, t = run_steps(engine, self.snap, SUNDAY + 1, 3) self.assertEqual(level(tiles["x32"]), "err") self.snap.conns["x32"]["category"] = "good" self.snap.conns_time = t self.assertEqual(level(engine.step(self.snap, t)["x32"]), "ok") # recovery is immediate def test_blip_shorter_than_debounce_is_ignored(self): engine = sp.Engine(CONFIG) self.snap.conns["x32"]["category"] = "error" run_steps(engine, self.snap, SUNDAY, 1) self.snap.conns["x32"]["category"] = "good" tiles, _ = run_steps(engine, self.snap, SUNDAY + 2, 5) self.assertEqual(level(tiles["x32"]), "ok") def test_flapping_shows_amber(self): engine = sp.Engine(CONFIG) t = SUNDAY for _ in range(3): self.snap.conns["x32"]["category"] = "error" _, t = run_steps(engine, self.snap, t, 4) self.snap.conns["x32"]["category"] = "good" tiles, t = run_steps(engine, self.snap, t, 2) self.assertEqual(level(tiles["x32"]), "warn") self.assertEqual(tiles["x32"]["x"], "X32\nOFFLINE x3") tiles, _ = run_steps(engine, self.snap, t + 700, 1) # quiet for 10 minutes -> green again self.assertEqual(level(tiles["x32"]), "ok") def test_ack_then_recover_then_fail_again(self): engine = sp.Engine(CONFIG) self.snap.conns["ATEM"]["category"] = "error" tiles, t = run_steps(engine, self.snap, SUNDAY, 4) self.assertEqual(level(tiles["atem"]), "err") self.assertEqual(engine.ack("atem"), 1) tiles, t = run_steps(engine, self.snap, t, 1) self.assertEqual(level(tiles["atem"]), "ack") self.assertEqual(level(tiles["all"]), "ack") self.snap.conns["ATEM"]["category"] = "good" tiles, t = run_steps(engine, self.snap, t, 1) self.assertEqual(level(tiles["atem"]), "ok") self.snap.conns["ATEM"]["category"] = "error" tiles, t = run_steps(engine, self.snap, t, 4) self.assertEqual(level(tiles["atem"]), "err") # a new failure blinks again def test_group_and_master_ack(self): engine = sp.Engine(CONFIG) self.snap.conns["VX1000"]["category"] = "error" self.snap.conns["x32"]["category"] = "error" tiles, t = run_steps(engine, self.snap, SUNDAY, 4) self.assertEqual(tiles["all"]["x"].split("\n")[0], "2 DOWN") engine.ack("video") tiles, t = run_steps(engine, self.snap, t, 1) self.assertEqual(level(tiles["video"]), "ack") self.assertEqual(level(tiles["x32"]), "err") engine.ack("all") tiles, t = run_steps(engine, self.snap, t, 1) self.assertEqual(level(tiles["all"]), "ack") def test_outside_service_nothing_blinks(self): snap = healthy_snapshot(WEDNESDAY) snap.conns["OBS-Youtube"]["category"] = "error" # OBS PC is off midweek snap.conns["x32"]["category"] = "error" snap.vars[("HyperDeck_3", "slotId")] = "-" # a deck that is always used tiles, _ = run_steps(sp.Engine(CONFIG), snap, WEDNESDAY, 4) self.assertEqual(level(tiles["obs"]), "idle") self.assertEqual(level(tiles["hd3"]), "idle") self.assertEqual(tiles["hd3"]["x"], "HD3\nNO DISK") self.assertEqual(level(tiles["x32"]), "quiet") self.assertEqual(tiles["all"]["x"], "1 DOWN\nX32 OFFLINE") self.assertEqual(level(tiles["all"]), "quiet") self.assertEqual(tiles["service"]["x"], "SERVICE\nOFF") for t in tiles.values(): self.assertEqual((t["b"], t["f"]), (t["b2"], t["f2"]), t["x"]) # no blinking anywhere def test_service_schedule_and_override(self): path = os.path.join(tempfile.mkdtemp(), "service.json") mode = sp.ServiceMode(CONFIG["service_windows"], path) self.assertTrue(mode.on(SUNDAY)) self.assertFalse(mode.on(WEDNESDAY)) mode.set(True, WEDNESDAY) # button: on for a Wednesday event self.assertTrue(mode.on(WEDNESDAY + 3600)) self.assertTrue(sp.ServiceMode(CONFIG["service_windows"], path).on(WEDNESDAY + 3600)) # survives restart sunday_start = datetime(2026, 10, 11, 7, 0).timestamp() self.assertTrue(mode.on(sunday_start)) # window edge clears the override self.assertIsNone(mode.override) mode.set(False, sunday_start + 60) # turned off early on Sunday self.assertFalse(mode.on(sunday_start + 120)) self.assertFalse(mode.on(datetime(2026, 10, 11, 12, 31).timestamp())) self.assertIsNone(mode.override) # end of window clears it too def test_network_burst(self): engine = sp.Engine(CONFIG) t = SUNDAY run_steps(engine, self.snap, t, 2) for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"): # the Oct 5 10:21 drop self.snap.conns[label]["category"] = "error" tiles, t = run_steps(engine, self.snap, t + 3, 4) self.assertEqual(tiles["net"]["x"], "NET\n5 DROPPED") self.assertEqual(level(tiles["net"]), "err") for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"): self.snap.conns[label]["category"] = "good" tiles, t = run_steps(engine, self.snap, t, 1) self.assertTrue(tiles["net"]["x"].startswith("NET\nBLIP ")) self.assertEqual(level(tiles["net"]), "warn") tiles, _ = run_steps(engine, self.snap, t + 200, 1) self.assertEqual(tiles["net"]["x"], "NET\nOK") def test_racks_powering_off_is_not_a_network_event(self): engine = sp.Engine(CONFIG) run_steps(engine, self.snap, SUNDAY, 2) for label in ("Axient_1", "Axient_2", "ULXD-4", "PSM_1", "PSM_2", "PSM_3"): self.snap.conns[label]["category"] = "error" tiles, _ = run_steps(engine, self.snap, SUNDAY + 3, 4) self.assertEqual(tiles["net"]["x"], "NET\nOK") self.assertEqual(tiles["rf"]["x"].split("\n")[-1], "+5") def test_gateway_and_link(self): self.snap.ping["10.1.0.1"] = 2 tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1) self.assertEqual(tiles["net"]["x"], "NET\nNO GATEWAY") self.snap.link = False tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1) self.assertEqual(tiles["net"]["x"], "NET\nLINK DOWN\n+1") def test_restart_targets(self): engine = sp.Engine(CONFIG) self.assertEqual(engine.restart_targets("hd3"), ["HyperDeck_3"]) self.assertEqual(engine.restart_targets("led"), ["VX1000", "UHDJr"]) self.assertEqual(engine.restart_targets("apps"), ["Propresenter-API", "ableton", "BibleTV"]) self.assertEqual(engine.restart_targets("dante"), []) # shared connection: never self.assertEqual(engine.restart_targets("ingest"), []) class TestFixes(unittest.TestCase): """Hold = fix only what that button is failing on. Nothing here runs a real fix.""" def setUp(self): self.snap = healthy_snapshot(SUNDAY) self.engine = sp.Engine(CONFIG) # Safety net: a hold in these tests must never reach a real machine. self.engine.executor = self.engine.poe_executor = lambda *a: True def plan(self, tile): run_steps(self.engine, self.snap, SUNDAY, 4) return [(kind, what) for _, kind, what, _ in self.engine.fix_plan(tile, self.snap)] def test_green_buttons_do_nothing(self): for tile in ("apps", "dante", "rec", "hd1", "x32", "ingest", "show"): self.assertEqual(self.plan(tile), [], tile) def test_propresenter_is_restarted_once(self): self.snap.conns["Propresenter-API"]["category"] = "error" self.snap.remote["prop\0pgrep -xq ProPresenter"] = False self.assertEqual(self.plan("apps"), [("fix", "propresenter")]) self.assertEqual(self.plan("show"), [("fix", "propresenter")]) # the group does the same def test_apps_on_other_macs(self): self.snap.remote["tracks\0pgrep -xq Live"] = False self.snap.remote["translation\0pgrep -xq 'Church Translate'"] = False self.assertEqual(self.plan("apps"), [("fix", "ableton"), ("fix", "translate")]) def test_bible_app_is_restarted_on_the_windows_pc(self): self.snap.http["http://10.6.7.13:8080/"] = False self.assertEqual(self.plan("apps"), [("fix", "bibleapp")]) fix = CONFIG["fixes"]["bibleapp"] self.assertEqual(CONFIG["hosts"][fix["host"]]["ssh"], "media@10.6.7.13") self.assertIn('schtasks /Run /TN "\\StatusPanel\\Start BibleApp"', fix["run"]) # cmd.exe, not sh def test_lighting_and_livestream_pcs(self): self.snap.conns["OBS-Youtube"]["category"] = "error" self.snap.conns["arena"]["category"] = "error" self.snap.artnet_since, self.snap.artnet_last = SUNDAY - 100, {"10.1.10.206": SUNDAY - 100} self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"] if p["name"] != "Lighting PC"] self.snap.mdns = self.snap.mdns - {"Lighting PC"} self.assertEqual(self.plan("obs"), [("fix", "obs")]) self.assertEqual(self.plan("arena"), [("fix", "arena")]) self.assertEqual(self.plan("apps"), [("fix", "vista")]) # Restarting rtpMIDI drops every session and it does not rejoin them (2026-10-08): no auto fix. self.assertEqual(self.plan("midi"), []) for fix_id in ("obs", "arena", "vista"): # never closes a running app self.assertTrue(CONFIG["fixes"][fix_id]["run"].startswith("tasklist "), fix_id) def test_an_off_pc_is_woken(self): from unittest import mock woke = [] done = mock.Mock(returncode=255, stdout="", stderr="ssh: connect to host 10.1.9.126 port 22: timed out") with mock.patch.object(sp.subprocess, "run", return_value=done), \ mock.patch.object(sp, "wake_on_lan", side_effect=lambda mac, bc=None: woke.append((mac, bc))): self.assertEqual(sp.execute_fix(CONFIG, "obs", {"conn": "OBS-Youtube"}), sp.WAKING) self.assertEqual(woke, [("6c:24:08:d7:fb:57", "10.1.255.255")]) self.assertFalse(sp.execute_fix(CONFIG, "arena", {"conn": "arena"})) # no MAC: just fails self.assertEqual(len(woke), 1) # the button says so self.snap.conns["OBS-Youtube"]["category"] = "error" self.engine.executor = lambda cfg, fix_id, chk: sp.WAKING _, t = run_steps(self.engine, self.snap, SUNDAY, 4) self.engine.run_fixes("obs", self.snap, None, t) time.sleep(0.2) self.assertEqual(self.engine.step(self.snap, time.time())["obs"]["x"], "OBS\nWAKING PC\nOBS") def test_magic_packet(self): from unittest import mock with mock.patch.object(sp.socket, "socket") as sock: packet = sp.wake_on_lan("6c:24:08:d7:fb:57", "10.1.255.255") self.assertEqual(packet, b"\xff" * 6 + bytes.fromhex("6c2408d7fb57") * 16) sent = [c.args[1] for c in sock.return_value.__enter__.return_value.sendto.call_args_list] self.assertEqual(sent, [("10.1.255.255", 9), ("255.255.255.255", 9)]) def test_companion_reconnects_after_the_app_starts(self): from unittest import mock comp = mock.Mock() comp.connections.return_value = [{"label": "arena", "id": "abc"}] ok = mock.Mock(returncode=0, stdout="", stderr="") with mock.patch.object(sp.subprocess, "run", return_value=ok), \ mock.patch.object(sp, "Companion", return_value=comp): self.assertTrue(sp.execute_fix(CONFIG, "arena", {"conn": "arena"})) comp.restart_connection.assert_called_once_with("abc") comp.reset_mock() self.assertTrue(sp.execute_fix(CONFIG, "vista", {"type": "artnet"})) # no "reconnect" comp.restart_connection.assert_not_called() def test_unreachable_mac_is_not_a_verdict(self): self.snap.remote["tracks\0pgrep -xq Live"] = None self.assertEqual(self.plan("apps"), []) def test_dante_reopens_dvs_on_the_right_mac(self): devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) devices.remove("NA-ProPresenter-Mac") devices.remove("BOH-WING") # hardware: re-check via Companion's Dante module self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller"), ("fix", "dvs_prop")]) def test_rec_reboots_only_hung_decks_not_recording_ones(self): for i in (2, 3, 5): self.snap.conns["HyperDeck_%d" % i]["category"] = "error" self.snap.vars[("HyperDeck_3", "status")] = "Record" # last known state: recording self.assertEqual(self.plan("rec"), [("fix", "hyperdeck"), ("fix", "hyperdeck")]) decks = [(chk["conn"], chk["ip"]) for _, _, _, chk in self.engine.fix_plan("rec", self.snap)] self.assertEqual(decks, [("HyperDeck_2", "10.1.10.12"), ("HyperDeck_5", "10.1.10.15")]) # the address comes from Companion and is remembered if the connection loses it self.snap.vars[("HyperDeck_2", "ip")] = None decks = [(chk["conn"], chk["ip"]) for _, _, _, chk in self.engine.fix_plan("rec", self.snap)] self.assertEqual(decks[0], ("HyperDeck_2", "10.1.10.12")) self.assertEqual(self.plan("hd3"), []) def test_disabled_connection_is_switched_on_never_rebooted(self): # 2026-10-08: HyperDeck_1 disabled in Companion, a hold sent the deck a reboot. self.snap.conns["HyperDeck_1"]["enabled"] = False self.snap.conns["x32"]["enabled"] = False self.assertEqual(self.plan("hd1"), [("enable", "HyperDeck_1")]) self.assertEqual(self.plan("rec"), [("enable", "HyperDeck_1")]) self.assertEqual(self.plan("x32"), [("enable", "x32")]) def test_not_recording_is_not_fixed_by_a_reboot(self): for i in (1, 3): self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record" self.assertEqual(self.plan("hd5"), []) # NOT REC: a person decides def test_connection_without_a_fix_is_restarted(self): self.snap.conns["x32"]["category"] = "error" self.assertEqual(self.plan("x32"), [("restart", "x32")]) def test_ingest_services(self): del self.snap.launchd["org.waterschurch.resi-propresenter"] self.snap.procs["/Applications/ProdDash.app/Contents/MacOS/ProdDash"] = False self.assertEqual(self.plan("ingest"), [("fix", "bridge"), ("fix", "proddash")]) def test_cooldown_and_executor(self): ran = [] self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) # record, never run self.snap.remote["tracks\0pgrep -xq Live"] = False run_steps(self.engine, self.snap, SUNDAY, 4) self.engine.run_fixes("apps", self.snap, None, SUNDAY + 10) self.engine.run_fixes("apps", self.snap, None, SUNDAY + 20) # held again 10 s later: skipped self.engine.run_fixes("apps", self.snap, None, SUNDAY + 80) time.sleep(0.2) self.assertEqual(ran, ["ableton", "ableton"]) def test_hold_feedback_on_the_button(self): devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) devices.remove("NA-ProPresenter-Mac") self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) gate = threading.Event() outcome = {"ok": True} def slow_fix(cfg, fix_id, chk): gate.wait(2) return outcome["ok"] self.engine.executor = slow_fix _, t = run_steps(self.engine, self.snap, SUNDAY, 4) self.engine.run_fixes("dante", self.snap, None, t) tiles = self.engine.step(self.snap, t) self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nFIXING\nDVS", "fixing")) gate.set() time.sleep(0.2) tiles = self.engine.step(self.snap, time.time()) self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nCHECKING\nDVS", "fixing")) tiles = self.engine.step(self.snap, time.time() + 25) # still gone: back to the alarm self.assertEqual(tiles["dante"]["x"], "DANTE\nPROP MAC GONE") # held again within a minute self.engine.run_fixes("dante", self.snap, None, t + 30) self.assertEqual(self.engine.notes["dante"]["text"], "JUST TRIED\nWAIT 1 MIN") # a fix whose command fails says so outcome["ok"] = False self.engine.run_fixes("dante", self.snap, None, t + 100) time.sleep(0.2) tiles = self.engine.step(self.snap, time.time()) self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nFIX FAILED\nDVS", "failed")) def test_dante_hold_power_cycles_any_poe_device_from_the_map(self): devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) devices.remove("STUDIO-B-MON") # PoE box (per the map) devices.remove("BOH-WING") # console: not PoE self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) self.snap.dante_map = { "STUDIO-B-MON": {"mac": "00:1d:c1:51:18:84", "switch": "10.66.1.102", "port": "0/15", "shared": 1, "poe": True}, "BOH-WING": {"mac": "00:1d:c1:03:ec:3a", "switch": "10.66.1.102", "port": "0/3", "shared": 1, "poe": False}, } self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller"), ("poe", "STUDIO-B-MON")]) self.snap.dante_map = {} # not mapped yet: just re-check via the module self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller")]) def test_poe_hold_hands_the_mapped_entry_to_the_switch_code(self): calls = [] self.engine.poe_executor = lambda cfg, name, entry: calls.append((name, entry["mac"])) or True devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) devices.remove("STUDIO-B-TB") self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) self.snap.dante_map = {"STUDIO-B-TB": {"mac": "00:1d:c1:89:1a:0a", "switch": "10.66.1.102", "port": "0/16", "shared": 1, "poe": True}} _, t = run_steps(self.engine, self.snap, SUNDAY, 4) self.engine.run_fixes("dante", self.snap, None, t) time.sleep(0.2) self.assertEqual(calls, [("STUDIO-B-TB", "00:1d:c1:89:1a:0a")]) tiles = self.engine.step(self.snap, time.time()) self.assertEqual(tiles["dante"]["x"], "DANTE\nCHECKING\nPOWER CYCLE") def test_dante_map_poller(self): class Comp: def variable(self, label, name): return {"STUDIO-B-MON_ip": "169.254.133.24", "NA-Tracks_ip": "169.254.99.70"}.get(name) self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["STUDIO-B-MON", "NA-Tracks"]) self.snap.dante_map = {"BOH-WING": {"mac": "x", "poe": False}} # an older entry is kept arp = lambda: {"169.254.133.24": "00:1d:c1:51:18:84", "169.254.99.70": "00:e0:4c:ff:b6:8a"} survey = lambda switches, macs: {"00:1d:c1:51:18:84": {"switch": "10.66.1.102", "port": "0/15", "shared": 1, "poe": True}} path = os.path.join(tempfile.mkdtemp(), "dante-map.json") sp.poll_dante_map(self.snap, CONFIG, Comp(), path, arp=arp, survey=survey) self.assertEqual(self.snap.dante_map["STUDIO-B-MON"]["port"], "0/15") self.assertTrue(self.snap.dante_map["STUDIO-B-MON"]["poe"]) self.assertNotIn("poe", self.snap.dante_map["NA-Tracks"]) # a Mac: not on a PoE port self.assertIn("BOH-WING", self.snap.dante_map) self.assertEqual(sp.load_dante_map(path), self.snap.dante_map) self.assertEqual(sp.normalize_mac("0:1d:c1:89:1a:a"), "00:1d:c1:89:1a:0a") def test_ingest_dvs_and_no_auto_fix(self): devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) devices.remove("STUDIO-A-Ingest") self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) self.assertEqual(self.plan("dante"), [("fix", "dvs_ingest")]) # A failure with no fix must not claim there is nothing wrong. self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"] if p["name"] != "SuperRack-LiveBox"] _, t = run_steps(self.engine, self.snap, SUNDAY, 4) self.engine.run_fixes("midi", self.snap, None, t) self.assertEqual(self.engine.notes["midi"]["text"], "NO AUTO\nFIX") def test_hold_on_a_green_button_says_so(self): _, t = run_steps(self.engine, self.snap, SUNDAY, 2) self.engine.run_fixes("x32", self.snap, None, t) tiles = self.engine.step(self.snap, t + 1) self.assertEqual((tiles["x32"]["x"], level(tiles["x32"])), ("X32\nNOTHING\nTO FIX", "idle")) tiles = self.engine.step(self.snap, t + 5) self.assertEqual(tiles["x32"]["x"], "X32\nOK") def test_remote_script_round_trip(self): script = sp.remote_script(["pgrep -xq Live", "pgrep -xq 'Church Translate'"]) self.assertIn("(pgrep -xq 'Church Translate') >/dev/null 2>&1; echo SP1=$?", script) self.assertEqual(sp.parse_remote("SP0=0\nSP1=1\n", 2), [True, False]) self.assertEqual(sp.parse_remote("", 2), [None, None]) def test_every_fix_is_valid(self): for f in CONFIG["fixes"].values(): self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck")) class FakeDeck: """A pretend HyperDeck on a local port: answers like a real one, stays silent, or greets only.""" def __init__(self, mode): import socket as _s self.mode, self.received = mode, b"" self.srv = _s.socket() self.srv.bind(("127.0.0.1", 0)) self.srv.listen(5) self.port = self.srv.getsockname()[1] threading.Thread(target=self.serve, daemon=True).start() def serve(self): while True: try: conn, _ = self.srv.accept() except OSError: return with conn: conn.settimeout(4) if self.mode != "silent": conn.sendall(b"500 connection info:\r\nprotocol version: 1.11\r\nmodel: HyperDeck Studio 4K Pro\r\n\r\n") try: data = conn.recv(1024) self.received += data if self.mode in ("record", "preview") and b"transport info" in data: conn.sendall(b"208 transport info:\r\nstatus: %s\r\nspeed: 100\r\n\r\n" % self.mode.encode()) more = conn.recv(1024) self.received += more except OSError: pass def close(self): self.srv.close() class TestHyperDeckProbe(unittest.TestCase): def test_probe_answers(self): for mode, expected in (("record", "record"), ("preview", "preview"), ("greet", "answered"), ("silent", "hung")): deck = FakeDeck(mode) try: self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=deck.port), expected, mode) finally: deck.close() deck = FakeDeck("preview") port = deck.port deck.close() self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=port), "unreachable") def test_only_a_silent_deck_is_rebooted(self): deck = FakeDeck("silent") try: chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port} self.assertTrue(sp.execute_fix(CONFIG, "hyperdeck", chk)) time.sleep(0.3) self.assertIn(b"reboot", deck.received) finally: deck.close() def test_a_recording_deck_is_never_rebooted(self): deck = FakeDeck("record") try: cfg = json.loads(json.dumps(CONFIG)) cfg["companion"]["port"] = 1 # no Companion here: the reconnect just fails chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port} sp.execute_fix(cfg, "hyperdeck", chk) time.sleep(0.3) self.assertNotIn(b"reboot", deck.received) finally: deck.close() class TestNetgearParsing(unittest.TestCase): BIG = """ VLAN ID MAC Address Interface IfIndex Status 66 00:1D:C1:51:34:C8 0/14 14 Learned 66 00:1D:C1:51:18:84 0/15 15 Learned 66 00:E0:4C:9A:1A:F3 0/26 26 Learned 66 00:0E:DD:E2:8E:C2 0/26 26 Learned 66 00:0E:DD:E2:8F:60 0/26 26 Learned""" SMALL = """ 66 00:1D:C1:51:34:C8 0/10 10 Learned 66 00:1D:C1:51:18:84 0/10 10 Learned 66 00:E0:4C:9A:1A:F3 0/3 3 Learned""" def test_access_port_is_where_the_device_has_fewest_neighbours(self): import netgear best = netgear.best_ports({"10.66.1.100": self.SMALL, "10.66.1.102": self.BIG}, ["00:1d:c1:51:34:c8", "00:e0:4c:9a:1a:f3", "00:0e:dd:e2:8e:c2"]) self.assertEqual(best["00:1d:c1:51:34:c8"], ("10.66.1.102", "0/14", 1)) # not the uplink 0/10 self.assertEqual(best["00:e0:4c:9a:1a:f3"], ("10.66.1.100", "0/3", 1)) self.assertEqual(best["00:0e:dd:e2:8e:c2"][2], 3) # shared port: too many self.assertGreater(3, netgear.MAX_SHARED) # 3 = uplink self.assertTrue(netgear.poe_delivering("0/14 Yes 32000 1 1500 29 54 Delivering Power No Error")) self.assertFalse(netgear.poe_delivering("0/14 Yes 32000 0 0 0 0 Searching No Error")) def test_never_cycles_an_uplink(self): import netgear orig = netgear.survey netgear.survey = lambda switches, macs: {macs[0]: {"switch": "s", "port": "0/26", "shared": 3, "poe": False}} try: with self.assertRaises(RuntimeError): netgear.poe_cycle_device(["s"], "00:0e:dd:e2:8e:c2") finally: netgear.survey = orig class FakeCompanion(BaseHTTPRequestHandler): """Just enough of Companion 5's HTTP API.""" store = {} def log_message(self, *a): pass def _send(self, status, body): data = body.encode() if isinstance(body, str) else body self.send_response(status) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def do_GET(self): s = self.store path = urllib.parse.urlparse(self.path).path if path == "/api/connections": return self._send(200, json.dumps([{"id": c["id"], "label": l, "enabled": c["enabled"], "status": {"category": c["category"], "message": c["message"]}} for l, c in s["conns"].items()])) m = re.match(r"^/api/variable/([^/]+)/([^/]+)/value$", path) if m: key = (urllib.parse.unquote(m.group(1)), urllib.parse.unquote(m.group(2))) v = s["vars"].get(key) return self._send(404, "Not found") if v is None else self._send(200, str(v)) m = re.match(r"^/api/custom-variable/([^/]+)/value$", path) if m: v = s["custom"].get(m.group(1)) return self._send(404, "Not found") if v is None else self._send(200, v) self._send(404, "") def do_POST(self): s = self.store u = urllib.parse.urlparse(self.path) m = re.match(r"^/api/custom-variable/([^/]+)/value$", u.path) if m: if m.group(1) not in s["custom"]: return self._send(404, "Not found") s["custom"][m.group(1)] = urllib.parse.parse_qs(u.query, keep_blank_values=True)["value"][0] return self._send(200, "ok") m = re.match(r"^/api/connections/([^/]+)/restart$", u.path) if m: s["restarts"].append(m.group(1)) return self._send(200, "{}") self._send(404, "") class TestAgainstFakeCompanion(unittest.TestCase): @classmethod def setUpClass(cls): snap = healthy_snapshot(time.time()) FakeCompanion.store = { "conns": snap.conns, "vars": dict(snap.vars), "custom": {"sp_state": "", "sp_hb": "", "sp_cmd": "", "flash_tick": "0"}, "restarts": [], } cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion) threading.Thread(target=cls.server.serve_forever, daemon=True).start() cls.port = cls.server.server_address[1] @classmethod def tearDownClass(cls): cls.server.shutdown() def test_client_round_trip_and_long_state(self): comp = sp.Companion("127.0.0.1", self.port) self.assertEqual(len(comp.connections()), len(FakeCompanion.store["conns"])) self.assertEqual(comp.variable("HyperDeck_1", "recordingTime"), "28:00:00") self.assertIsNone(comp.variable("HyperDeck_1", "nope")) snap = healthy_snapshot(time.time()) tiles = sp.Engine(CONFIG).step(snap, time.time()) js = json.dumps(tiles, separators=(",", ":"), sort_keys=True) self.assertEqual(comp.set_custom("sp_state", js), (True, 200)) self.assertEqual(json.loads(comp.get_custom("sp_state"))["all"]["x"], "ALL OK") self.assertEqual(comp.set_custom("missing", "x"), (False, 404)) print("\n sp_state is %d bytes, %d in the URL" % (len(js), len(urllib.parse.quote(js, safe="")))) def test_commands(self): comp = sp.Companion("127.0.0.1", self.port) store = FakeCompanion.store snap = healthy_snapshot(SUNDAY) engine = sp.Engine(CONFIG) engine.executor = lambda cfg, fix_id, chk: self.fail("no real fix may run in a test") now = SUNDAY store["conns"]["VX1000"]["category"] = "error" try: sp.poll_connections(comp, snap) run_steps(engine, snap, now, 4) before = len(store["restarts"]) # hd3 is green, so its hold does nothing; LED is failing on VX1000, so that restarts. sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now + 5) self.assertEqual(store["restarts"][before:], ["id-VX1000"]) finally: store["conns"]["VX1000"]["category"] = "good" self.assertFalse(engine.service.on(now)) sp.handle_commands("service toggle", engine, comp, snap, now) self.assertTrue(engine.service.on(now)) def test_end_to_end_loop(self): """The real run() loop: it writes state and heartbeat, and consumes button commands.""" cfg = json.loads(json.dumps(CONFIG)) cfg["companion"]["port"] = self.port # keep it offline: no pings, no web, no launchd, no bonjour lookups cfg["network"]["gateway"] = None keep = ("conn", "hyperdeck", "dante", "tascam", "shure", "psm") for t in cfg["tiles"]: if "checks" in t: t["checks"] = [c for c in t["checks"] if c["type"] in keep] or [{"type": "conn", "conn": "x32"}] store = FakeCompanion.store store["conns"]["ATEM"]["category"] = "error" store["custom"]["sp_cmd"] = ";ack atem" data_dir = tempfile.mkdtemp() originals = sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map sp.poll_dante_map = lambda *a, **k: None # never log in to a switch from a test sp.poll_mdns = lambda snap, seconds=3: None sp.poll_system = lambda snap, cfg: None sp.poll_midi = lambda snap, cfg: None # CoreMIDI has its own test sp.poll_remote = lambda snap, cfg: None # never SSH anywhere from a test stop = threading.Event() loop = threading.Thread(target=sp.run, args=(cfg, data_dir), kwargs={"stop": stop}, daemon=True) try: loop.start() deadline = time.time() + 12 state = {} while time.time() < deadline: raw = store["custom"]["sp_state"] state = json.loads(raw) if raw else {} if state.get("atem", {}).get("x") == "ATEM\nOFFLINE": break time.sleep(0.25) finally: stop.set() loop.join(timeout=15) sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map = originals store["conns"]["ATEM"]["category"] = "good" self.assertFalse(loop.is_alive()) # it shuts down cleanly self.assertTrue(store["custom"]["sp_hb"].isdigit()) self.assertLess(abs(int(store["custom"]["sp_hb"]) - time.time()), 5) self.assertEqual(store["custom"]["sp_cmd"], "") # the command was consumed self.assertIn("atem", state) self.assertIn("ATEM", state["atem"]["x"]) class TestPage(unittest.TestCase): def setUp(self): self.export = make_page.build(CONFIG) self.controls = self.export["page"]["controls"] def test_layout(self): self.assertEqual(self.export["type"], "page") self.assertEqual(self.export["oldPageNumber"], 19) self.assertEqual(self.controls["0"]["0"], {"type": "pageup"}) # only the page number follows ALL self.assertEqual(self.controls["2"]["0"], {"type": "pagedown"}) for (r, kind, action) in (("1", "pagenum", "set_page"),): key = self.controls[r]["0"] self.assertEqual(key["type"], "button-layered") self.assertTrue(key["options"]["notes"].startswith("STATUS nav key (%s)" % kind)) step = key["steps"]["0"] self.assertEqual(step["action_sets"]["down"], []) tap, hold = step["action_sets"]["up"][0], step["action_sets"]["1000"][0] self.assertEqual((tap["definitionId"], tap["options"]["page"]["value"]), (action, 1)) # page 1 outright, not "startup" self.assertEqual((hold["definitionId"], hold["options"]["page"]["value"]), ("set_page", 19)) self.assertEqual(step["options"]["runWhileHeld"], [1000]) box = [l for l in key["style"]["layers"] if l["type"] == "box"][0] self.assertTrue(box["color"]["isExpression"]) self.assertIn("'$.all.b'", box["color"]["value"]) # follows the ALL button positioned = [t for t in CONFIG["tiles"] if "pos" in t] n = sum(1 for r in self.controls.values() for c in r.values() if c["type"] == "button-layered") self.assertEqual(n, len(positioned) + 1) # + the page number key # The master is where the smallest Stream Deck can see it. self.assertEqual(CONFIG["tiles"][0]["pos"], [0, 1]) def test_each_button_reads_its_own_tile(self): for t in CONFIG["tiles"]: if "pos" not in t: continue r, c = (str(x) for x in t["pos"]) layers = {l["id"]: l for l in self.controls[r][c]["style"]["layers"]} for lid, key in (("box0", "color"), ("text0", "color"), ("text0", "text"), ("text0", "fontsize")): expr = layers[lid][key] self.assertTrue(expr["isExpression"]) self.assertIn("'$.%s." % t["id"], expr["value"]) self.assertIn("$(custom:sp_hb) > 15", expr["value"]) self.assertIn("$(custom:flash_tick) == 1", layers["box0"]["color"]["value"]) def test_actions(self): hd1 = self.controls["1"]["1"]["steps"]["0"] up = hd1["action_sets"]["up"][0]["options"]["value"] self.assertEqual(up, {"value": "$(custom:sp_cmd);ack hd1", "isExpression": False}) self.assertEqual(hd1["action_sets"]["2000"][0]["options"]["value"]["value"], "$(custom:sp_cmd);restart hd1") self.assertEqual(hd1["options"]["runWhileHeld"], [2000]) master = self.controls["0"]["1"]["steps"]["0"] self.assertIn("ackall", master["action_sets"]["up"][0]["options"]["value"]["value"]) self.assertNotIn("2000", master["action_sets"]) ingest = self.controls["3"]["5"]["steps"]["0"] self.assertIn("2000", ingest["action_sets"]) # restarts its services self.assertIn("2000", self.controls["0"]["2"]["steps"]["0"]["action_sets"]) # REC group net = self.controls["3"]["4"]["steps"]["0"] self.assertNotIn("2000", net["action_sets"]) # nothing to fix there ids = [a["id"] for r in self.controls.values() for c in r.values() for s in c.get("steps", {}).values() for al in s["action_sets"].values() for a in al] self.assertEqual(len(ids), len(set(ids))) def test_stale_text_and_file(self): master = {l["id"]: l for l in self.controls["0"]["1"]["style"]["layers"]} self.assertIn("`${'WATCHDOG\\nDOWN\\n'}${$(internal:instance_errors)} ERR`", master["text0"]["text"]["value"]) # No string '+' anywhere: in Companion 5 expressions it adds numbers. for r in self.controls.values(): for c in r.values(): for layer in c.get("style", {}).get("layers", []): for p in layer.values(): if isinstance(p, dict) and p.get("isExpression"): self.assertNotRegex(p["value"], r"'\s*\+|\+\s*'") path = os.path.join(tempfile.mkdtemp(), "p.companionconfig") sys.argv = ["make_page.py", path] make_page.main() on_disk = make_page.build(sp.load_config(os.path.join(ROOT, "config.json"))) with gzip.open(path) as f: self.assertEqual(json.load(f), json.loads(json.dumps(on_disk))) class TestCoreMIDI(unittest.TestCase): def test_listener_loopback(self): import coremidi src = coremidi.VirtualSource("status-panel-test") try: time.sleep(0.3) lis = coremidi.Listener("status-panel-test") lis.poll() self.assertTrue(lis.connected) src.send([0x90, 36, 100]) src.send([0xF8]) # clock is ignored time.sleep(0.3) self.assertEqual((lis.last_msg, lis.count), ("NOTE 36", 1)) snap = sp.Snapshot() snap.listeners["status-panel-test"] = lis res = sp.check_midi_rx({"source": "status-panel-test", "name": "IN"}, snap, {"now": time.time(), "service": True})[0][1] self.assertEqual(res.info, "IN 0s") finally: src.close() time.sleep(0.3) lis.poll() self.assertFalse(lis.connected) def test_describe(self): import coremidi self.assertEqual(coremidi.describe([0xB0, 7, 64]), "CC 7=64") self.assertEqual(coremidi.describe([0xC2, 5]), "PC 5") self.assertEqual(coremidi.describe([0x90, 60, 0]), "OFF 60") self.assertEqual(coremidi.describe([0xF1, 0x20]), "MTC") if __name__ == "__main__": unittest.main()