Add the Companion status page and its watchdog for NA Ingest.
status_panel.py runs on NA Ingest as a LaunchAgent. Every second it reads Companion's connection statuses and module variables, adds what Companion cannot see (network MIDI peers, Bonjour sessions, pings, launchd services, disk, Resi cloud, the Bible TV app), and writes one JSON custom variable that page 19 renders. Buttons acknowledge alarms, mute them, and restart a stuck Companion connection on a two second hold. If the watchdog stops, the page turns purple instead of freezing on stale green. make_page.py builds the page export from config.json; tests/ runs offline against a fake Companion and a CoreMIDI loopback. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,617 @@
|
||||
"""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, "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\n10/10")
|
||||
|
||||
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.OFF)
|
||||
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_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_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["lightpc"]["x"], "LIGHT PC\nMIDI OK")
|
||||
|
||||
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["lightpc"]["x"], "LIGHT PC\nMIDI LOST")
|
||||
self.assertEqual(tiles["tracks"]["x"], "TRACKS\nMIDI SESS 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["bibletv"]["x"], "BIBLE TV\nAPP DOWN")
|
||||
|
||||
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_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("tracks"), ["ableton"]) # not the shared Dante one
|
||||
self.assertEqual(engine.restart_targets("ingest"), [])
|
||||
|
||||
|
||||
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)
|
||||
snap = sp.Snapshot()
|
||||
sp.poll_connections(comp, snap)
|
||||
engine = sp.Engine(CONFIG)
|
||||
now = SUNDAY
|
||||
sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now)
|
||||
self.assertEqual(FakeCompanion.store["restarts"][-3:], ["id-HyperDeck_3", "id-VX1000", "id-UHDJr"])
|
||||
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_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
|
||||
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 = 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"})
|
||||
self.assertEqual(self.controls["2"]["0"], {"type": "pagedown"})
|
||||
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))
|
||||
# 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.assertNotIn("2000", ingest["action_sets"]) # nothing to restart 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()
|
||||
Reference in New Issue
Block a user