christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
14fce2cf9a73ef7ab105367adb2885f2805cb52e
status-indicators/tests/test_status_panel.py
T
christianandClaude Fable 5.1 14fce2cf9a Audio failover: press FOH Backup by itself when Studio B dies; Companion button presses as fixes 
A new fix kind "companion_press" presses any Companion button over its HTTP API (page, row, col).
A fix marked "failover" runs by itself within seconds of its check failing: inside the "failover"
windows (Sunday 06:30-13:30), even while ALARMS is muted, once per fault, never back the other way.
The hidden STB AUDIO key (inside AUDIO) watches Studio B's ping every second, Pro Tools over SSH
every 2 s and its Dante device; each check's fix is the FOH Backup press. The keys say ON FOH BACKUP
until STREAM STB on the Restart page switches back. Settings page: the press builder, the failover
tick and undo, the Failover calendar, help entry "A fallback button".

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 12:40:24 -04:00

2159 lines
123 KiB
Python
Raw Blame History

"""Offline tests: no Companion, no gear. Run: python3 -m unittest discover -s tests"""
import gzip
import json
import logging
import os
import re
import subprocess
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()
CONFIG["auto_fix"]["log_dir"] = tempfile.mkdtemp(prefix="sp-autolog-") # never the real Desktop
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,
"org.waterschurch.lobby-captions-display": 901, "org.waterschurch.digico-relay": 902}
snap.launchd_time = now
snap.procs = {"/Applications/ProdDash.app/Contents/MacOS/ProdDash": True}
snap.remote = {"studiob\0pgrep -xq 'Pro Tools'": True} # the failover's check has a verdict
snap.http = {"https://central.resi.io/": True, "http://10.6.7.13:8080/": True, "http://10.1.14.225:1025/": 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), (b2, f2)) in sp.COLORS.items():
if (tile["b"], tile["f"], tile["b2"], tile["f2"]) == (b, f, b2, f2):
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")
and not k.startswith("rs_")} # Restart page keys sit grey
self.assertEqual(bad, {})
self.assertEqual((tiles["rs_dvs_tracks"]["x"], level(tiles["rs_dvs_tracks"])), ("DVS TRACKS\nHOLD 2s", "idle"))
self.assertEqual(tiles["all"]["x"], "ALL OK")
self.assertEqual(tiles["hd1"]["x"], "HD1\n28h")
self.assertEqual(tiles["dante"]["x"], "DANTE\n19/19")
def test_spotify(self):
chk = next(c for t in CONFIG["tiles"] for c in t.get("checks", []) if c["type"] == "spotify")
s = self.snap
msg = lambda ctx=self.ctx: sp.check_spotify(chk, s, ctx)[0][1]
self.assertEqual(msg().level, sp.OK) # not read yet
s.spotify = {"running": False}
self.assertEqual(msg().msg, "SPOTIFY CLOSED")
self.assertEqual(msg({"now": WEDNESDAY}).msg, "SPOTIFY CLOSED") # alarms like every other key, any day
s.spotify = {"running": True, "state": "paused", "track": "spotify:track:a"}
self.assertEqual(msg().msg, "NOT PLAYING")
s.spotify["state"] = "playing"
s.spotify_tracks = {"spotify:track:a", "spotify:track:b"}
s.spotify_recent.extend(["spotify:track:x", "spotify:track:y"])
self.assertEqual(msg().level, sp.OK) # list incomplete: 2 unknown is fine
s.spotify_recent.append("spotify:track:z")
self.assertEqual(msg().msg, "WRONG PLAYLIST")
s.spotify_recent.append("spotify:track:a")
self.assertEqual(msg().level, sp.OK)
s.spotify_complete = True
s.spotify_recent.append("spotify:track:q")
self.assertEqual(msg().msg, "WRONG PLAYLIST") # full list: one unknown is enough
fix = CONFIG["fixes"][chk["fix"]]
self.assertEqual(fix["kind"], "spotify")
self.assertEqual(sp.spotify_playlist(CONFIG, chk), CONFIG["spotify"]["playlist"])
self.assertIn(chk["fix"], CONFIG["auto_fix"]["fixes"])
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)
# its own button: "VISTA / SILENT", and the groups say "VISTA SILENT"
self.snap.artnet_since, self.snap.artnet_last = self.now - 60, {"10.1.10.206": self.now - 8}
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
self.assertEqual(tiles["vista"]["x"], "VISTA\nSILENT")
self.assertEqual(tiles["show"]["x"], "SHOW\nVISTA SILENT")
self.assertEqual(tiles["apps"]["x"], "APPS\nOK")
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_companion_midi_and_prop_tv(self):
engine = sp.Engine(CONFIG)
self.snap.conns["lighting-midi"]["category"] = "error"
self.snap.http["http://10.1.14.225:1025/"] = False
tiles, _ = run_steps(engine, self.snap, self.now, 4)
self.assertEqual(tiles["midi"]["x"], "MIDI\nCOMPANION MIDI OFFLINE")
self.assertEqual(tiles["apps"]["x"], "APPS\nPROP TV DOWN")
self.assertEqual(dict(sp.http_checks(CONFIG))["http://10.1.14.225:1025/"], 5)
def test_pro_tools_solo_mode_on_studio_b(self):
chk = [c for c in CONFIG["tiles"][[t["id"] for t in CONFIG["tiles"]].index("apps")]["checks"]
if c.get("msg") == "PT SOLO NOT AFL"][0]
self.snap.remote["studiob\0" + chk["cmd"]] = False
engine = sp.Engine(CONFIG)
tiles, _ = run_steps(engine, self.snap, self.now, 4)
self.assertEqual(tiles["apps"]["x"], "APPS\nPT SOLO NOT AFL")
self.assertEqual(engine.fix_plan("apps", self.snap)[0][1:3], ("fix", "solo_afl"))
self.assertIn("solo_afl", CONFIG["auto_fix"]["fixes"])
self.assertIn('"bad":true', chk["cmd"])
def test_atem_software_midi_control_must_stay_off(self):
key = ("com.blackmagic-design.ATEM Software Control", "midiEnabled")
engine = sp.Engine(CONFIG)
self.snap.prefs[key] = "0"
self.assertEqual(run_steps(engine, self.snap, self.now, 4)[0]["atem"]["x"], "ATEM\nOK")
self.snap.prefs[key] = "1"
self.assertEqual(run_steps(engine, self.snap, self.now + 5, 4)[0]["atem"]["x"], "ATEM\nATEM SW MIDI ON")
def test_atem_reference_from_the_restore_file(self):
d = tempfile.mkdtemp()
xml = os.path.join(d, "r.xml")
open(xml, "w").write("""<Profile product="ATEM 4 M/E Constellation 4K Plus"><FairlightAudioMixer>
<AudioInputs><AudioInput id="16"><AudioSources><AudioSource id="-65280" faderGain="0" mixOption="On"/></AudioSources></AudioInput>
<AudioInput id="1301"><AudioSources><AudioSource id="-65280" faderGain="-inf" mixOption="Off"/></AudioSources></AudioInput>
<AudioInput id="7"><AudioSources><AudioSource id="-65280" faderGain="-2.5" mixOption="AudioFollowVideo"/></AudioSources></AudioInput>
</AudioInputs></FairlightAudioMixer><CameraControl>
<Parameter device="2" category="Video" parameter="ManualWhiteBalance" temperature="4500" tint="0"/>
<Parameter device="6" category="ColorCorrection" parameter="LiftAdjust" red="0" green="0" blue="0" luma="0.18994140625"/>
<Parameter device="1" category="Lens" parameter="ApertureNormalised" value="0"/>
<Parameter device="1" property="ApertureCoarse" value="1"/></CameraControl></Profile>""")
ref = sp.atem_reference(xml)
self.assertEqual(ref["audio"], {"16/-65280": {"mixOption": 2, "faderGain": 0},
"1301/-65280": {"mixOption": 1}, # -inf: fader not compared
"7/-65280": {"mixOption": 4, "faderGain": -250}})
self.assertEqual(ref["camera"], {"2/1/2": [4500.0, 0.0], "6/8/0": [0.0, 0.0, 0.0, 0.18994140625]})
def test_atem_audio_and_camera_checks(self):
engine = sp.Engine(CONFIG)
self.snap.atem = {"connected": True, "audio": [], "camera": [], "time": self.now}
self.assertEqual(run_steps(engine, self.snap, self.now, 4)[0]["atem"]["x"], "ATEM\nOK")
self.snap.atem = {"connected": True, "audio": ["input 16: Off, should be On"], "camera": ["camera 2 ..."],
"time": self.now + 5}
tiles, _ = run_steps(engine, self.snap, self.now + 5, 4)
self.assertEqual(tiles["atem"]["x"], "ATEM\nATEM AUDIO WRONG\n+1")
self.assertEqual(sorted(w for _, k, w, _ in engine.fix_plan("atem", self.snap)), ["atem_audio", "atem_camera"])
self.snap.atem = None # watcher not reporting
self.assertEqual(run_steps(engine, self.snap, self.now + 10, 2)[0]["atem"]["x"], "ATEM\nOK")
for f in ("atem_audio", "atem_camera", "atem_midi"):
self.assertIn(f, CONFIG["auto_fix"]["fixes"])
self.assertIn("midiEnabled -bool false", CONFIG["fixes"]["atem_midi"]["run"])
def test_atem_fix_goes_through_the_watcher(self):
class W:
def __init__(self): self.parts = []
def fix(self, part): self.parts.append(part); return True
w, old = W(), sp.ATEM_WATCHER
sp.ATEM_WATCHER = w
try:
self.assertTrue(sp.execute_fix(CONFIG, "atem_camera", {}))
finally:
sp.ATEM_WATCHER = old
self.assertEqual(w.parts, ["camera"])
def test_propresenter_yellow_warning_is_ignored(self):
engine = sp.Engine(CONFIG)
self.snap.conns["Propresenter-API"].update(category="warning", message="Socket closed")
tiles, _ = run_steps(engine, self.snap, self.now, 5)
self.assertEqual(tiles["apps"]["x"], "APPS\nOK")
self.snap.conns["Propresenter-API"]["category"] = "error" # a real failure still alarms
tiles, _ = run_steps(engine, self.snap, self.now + 6, 5)
self.assertTrue(tiles["apps"]["x"].startswith("APPS\nPROPRES "), tiles["apps"]["x"])
self.snap.conns["Propresenter-API"].update(category="warning", enabled=False) # switched off
tiles, _ = run_steps(engine, self.snap, self.now + 12, 5)
self.assertEqual(tiles["apps"]["x"], "APPS\nPROPRES DISABLED")
def test_shure_receivers_are_pinged_at_companions_address(self):
chk = {"type": "ping", "conn": "Axient_1", "host": "10.1.1.101", "name": "AXIENT 1"}
self.assertEqual(sp.ping_target(chk, self.snap), "10.1.1.101") # fallback
self.snap.conn_hosts = {"Axient_1": "10.1.1.201"} # moved in Companion
self.assertEqual(sp.ping_target(chk, self.snap), "10.1.1.201")
self.assertIn("10.1.1.201", sp.ping_hosts(CONFIG, self.snap))
self.snap.ping["10.1.1.201"] = 2
self.assertEqual(sp.check_ping(chk, self.snap, self.ctx)[0][1].msg, "AXIENT 1 NO PING")
# every Shure receiver has one, and the RF group says which
self.snap.conn_hosts = {}
self.snap.ping["10.1.1.105"] = 3
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
self.assertEqual(tiles["rf"]["x"], "RF\nPSM 3 NO PING")
def test_companion_db_path_skips_a_dead_newer_folder(self):
import sqlite3
root = tempfile.mkdtemp(prefix="sp-companion-")
for v in ("v5.0", "v5.1"):
os.makedirs(os.path.join(root, v))
good = os.path.join(root, "v5.0", "db.sqlite")
db = sqlite3.connect(good); db.execute("CREATE TABLE instances (id TEXT, value TEXT)"); db.commit(); db.close()
open(os.path.join(root, "v5.1", "db.sqlite"), "w").write("not a database") # an abandoned upgrade
from unittest import mock
real = os.path.expanduser
with mock.patch.object(sp.os.path, "expanduser", lambda p: root if "companion" in p else real(p)):
self.assertEqual(sp.companion_db_path({"companion": {"build": "5.0.7+9763"}}), good)
self.assertEqual(sp.companion_db_path({"companion": {}}), good) # newest that opens
self.assertEqual(sp.companion_db_path({"companion": {"db": "/x/y.sqlite"}}), "/x/y.sqlite")
def test_companion_conn_hosts_reads_its_database_read_only(self):
import sqlite3
d = tempfile.mkdtemp()
path = os.path.join(d, "db.sqlite")
db = sqlite3.connect(path)
db.execute("CREATE TABLE instances (id STRING, value STRING)")
rows = [("a", {"label": "Axient_1", "config": {"host": "10.1.1.101", "port": 2202}}),
("b", {"label": "OBS-Youtube", "config": {"host": " 10.1.9.126 "}}),
("c", {"label": "no-host", "config": {}}), ("d", "not json")]
db.executemany("INSERT INTO instances VALUES (?, ?)",
[(i, v if isinstance(v, str) else json.dumps(v)) for i, v in rows])
db.commit()
db.close()
self.assertEqual(sp.companion_conn_hosts(path), {"Axient_1": "10.1.1.101", "OBS-Youtube": "10.1.9.126"})
self.assertEqual(sp.companion_conn_hosts(os.path.join(d, "missing.sqlite")), {})
def test_fx_lights_midi_on_the_tracks_mac(self):
engine = sp.Engine(CONFIG)
self.assertEqual(engine.step(self.snap, self.now)["midi"]["x"], "MIDI\nOK") # not read yet: no verdict
self.snap.remote_midi["tracks"] = {"NA Tracks": {"port": 5004, "peers": [
{"name": "SuperRack-LiveBox", "address": "", "state": "running"},
{"name": "Studio B Mac Studio", "address": "", "state": "running"},
{"name": "Lighting PC", "address": "", "state": "running"}]}}
self.assertEqual(engine.step(self.snap, self.now)["midi"]["x"], "MIDI\nOK")
self.snap.remote_midi["tracks"]["NA Tracks"]["peers"].pop()
tiles, _ = run_steps(engine, self.snap, self.now, 4)
self.assertEqual(tiles["midi"]["x"], "MIDI\nFX LIGHTS MIDI LOST")
self.snap.remote_midi["tracks"] = None # Tracks Mac unreachable: its ping says so
tiles, _ = run_steps(engine, self.snap, self.now + 5, 1)
self.assertEqual(tiles["midi"]["x"], "MIDI\nOK")
def test_remote_midi_script_runs_coremidi(self):
r = subprocess.run(["/usr/bin/python3", "-"], input=sp.remote_midi_script(),
capture_output=True, text=True, timeout=30)
self.assertEqual(r.returncode, 0, r.stderr)
self.assertIsInstance(sp.parse_remote_midi(r.stdout), dict)
self.assertIsNone(sp.parse_remote_midi("ssh: connect to host timed out"))
self.assertEqual(sp.remote_midi_hosts(CONFIG), ["prop", "tracks"])
def test_a_midi_drop_shows_within_a_second_or_two(self):
engine = sp.Engine(CONFIG)
run_steps(engine, self.snap, self.now, 2)
self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"] if p["name"] != "Lighting PC"]
tiles, _ = run_steps(engine, self.snap, self.now + 3, 1.0, step=0.5)
self.assertEqual(tiles["midi"]["x"], "MIDI\nLIGHT PC MIDI LOST") # 1 s, not 3
def test_remote_midi_stream_script_runs_and_streams(self):
p = subprocess.Popen(["/usr/bin/python3", "-u", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
p.stdin.write(sp.remote_midi_stream_script(0.1))
p.stdin.close()
first = p.stdout.readline()
p.kill()
self.assertIsInstance(sp.parse_remote_midi(first), dict)
def test_a_silent_remote_stream_gives_no_verdict(self):
self.snap.remote_midi["tracks"] = {"NA Tracks": {"port": 5004, "peers": []}}
self.snap.remote_midi_time["tracks"] = self.now - 30
self.assertEqual(sp.check_midi_peer({"type": "midi_peer", "host": "tracks", "session": "NA Tracks",
"peer": "Lighting PC"}, self.snap, self.ctx)[0][1].level, sp.OK)
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"]), "notice") # flashes amber
self.assertTrue(tiles["all"]["x"].startswith("1 WARN\nNET BLIP ")) # a warning, not a DOWN
tiles, _ = run_steps(engine, self.snap, t + 200, 1)
self.assertTrue(tiles["net"]["x"].startswith("NET\nBLIP ")) # still up after 3+ minutes
tiles, _ = run_steps(engine, self.snap, t + 320, 1)
self.assertEqual(tiles["net"]["x"], "NET\nOK") # gone by itself after 5
def test_network_blip_press_and_hold(self):
engine = sp.Engine(CONFIG)
t = SUNDAY
run_steps(engine, self.snap, t, 2)
drop = ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam")
for label in drop:
self.snap.conns[label]["category"] = "error"
_, t = run_steps(engine, self.snap, t + 3, 4)
for label in drop:
self.snap.conns[label]["category"] = "good"
tiles, t = run_steps(engine, self.snap, t, 1)
engine.ack("net") # press: stop the flashing
tiles, t = run_steps(engine, self.snap, t, 1)
self.assertEqual((tiles["net"]["x"].split("\n")[1][:5], level(tiles["net"])), ("BLIP ", "warn"))
engine.run_fixes("net", self.snap, None, t) # hold: gone now
tiles, t = run_steps(engine, self.snap, t + 4, 1)
self.assertEqual((tiles["net"]["x"], tiles["all"]["x"]), ("NET\nOK", "ALL OK"))
engine.run_fixes("net", self.snap, None, t)
self.assertEqual(engine.notes["net"]["text"], "NOTHING\nTO CLEAR")
def test_helper_connections_no_button_watches_are_not_a_blip(self):
engine = sp.Engine(CONFIG)
for i in range(1, 7):
self.snap.conns["tcp-HyperDeck-%d" % i] = {"id": "t%d" % i, "enabled": True, "category": "good", "message": None}
run_steps(engine, self.snap, SUNDAY, 2)
for i in range(1, 7): # what happened at 07:45:55
self.snap.conns["tcp-HyperDeck-%d" % i]["category"] = "error"
tiles, _ = run_steps(engine, self.snap, SUNDAY + 3, 4)
self.assertEqual(tiles["net"]["x"], "NET\nOK")
self.assertIsNone(engine.burst)
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"), [])
def mock_comp():
class Comp:
def __init__(self):
self.restarted, self.enabled = [], []
def restart_connection(self, cid):
self.restarted.append(cid)
return True
def enable_connection(self, cid):
self.enabled.append(cid)
return True
return Comp()
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 (or write a real log).
self.engine.executor = self.engine.poe_executor = lambda *a: True
self.engine.autolog = sp.AutoLog(tempfile.mkdtemp())
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("vista"), [("fix", "vista")])
self.assertEqual(self.plan("apps"), [])
# Restarting rtpMIDI drops every session and does not rejoin them (2026-10-08): the hold
# starts rtpMIDI only if it is not running, and reconnects the peer from the Mac's side.
self.assertEqual(sorted(self.plan("midi")), [("fix", "midi_reconnect"), ("fix", "rtpmidi_start")])
run = CONFIG["fixes"]["rtpmidi_start"]["run"]
self.assertTrue(run.startswith("(sc query rtpMIDIService | find"), run)
self.assertNotIn("Restart-Service", run)
self.assertNotIn("net stop", run)
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_shure_back_online_clears_and_holds_do_nothing(self):
self.snap.conns["Axient_1"]["category"] = "error"
run_steps(self.engine, self.snap, SUNDAY, 4)
self.assertEqual(self.plan("rf"), [("restart", "Axient_1")])
self.engine.events.clear()
self.snap.conns["Axient_1"]["category"] = "good" # back: now reported per channel
tiles, _ = run_steps(self.engine, self.snap, SUNDAY + 5, 2)
self.assertEqual(tiles["rf"]["x"], "RF\nOK")
self.assertIn(("CLEAR", "AXIENT 1 OFFLINE"), self.engine.events)
self.assertEqual(self.plan("rf"), []) # nothing left to "fix"
self.assertFalse(self.engine.failing("rf"))
def test_restart_page_keys_run_even_when_all_is_fine(self):
ran = []
self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) or True
run_steps(self.engine, self.snap, SUNDAY, 4) # everything green
self.assertEqual(self.engine.run_force("rs_dvs_tracks", self.snap, None, SUNDAY + 5), "dvs_tracks")
time.sleep(0.2)
self.assertEqual(ran, ["dvs_tracks"])
self.assertEqual(self.engine.notes["rs_dvs_tracks"]["text"], "DONE")
self.assertIsNone(self.engine.run_force("rs_dvs_tracks", self.snap, None, SUNDAY + 20)) # 60 s guard
self.assertEqual(self.engine.notes["rs_dvs_tracks"]["text"], "JUST DONE\nWAIT 1 MIN")
self.assertIsNone(self.engine.run_force("rs_nope", self.snap, None, SUNDAY + 90))
# the hold on the status page and the Restart key share the guard
self.engine.run_force("rs_prodcom", self.snap, None, SUNDAY + 100)
self.engine.notes.clear()
self.snap.http["http://127.0.0.1:24480/api/v1/ping"] = False
run_steps(self.engine, self.snap, SUNDAY + 100, 4)
self.engine.run_fixes("ingest", self.snap, None, SUNDAY + 106)
self.assertEqual(self.engine.notes["ingest"]["text"], "JUST TRIED\nWAIT 1 MIN")
# what each key restarts
keys = {k["id"]: k["fix"] for k in CONFIG["restart_page"]["keys"]}
self.assertEqual(keys["rs_ableton"], "ableton_restart")
self.assertIn("pkill -TERM -x Live", CONFIG["fixes"]["ableton_restart"]["run"])
self.assertIn("kickstart -k system/com.audinate.dante.DanteVirtualSoundcard",
open(os.path.join(os.path.dirname(os.path.abspath(sp.__file__)), "remote-access", "mac-dvs-helper.sh.template")).read())
self.assertIn("open -g /Applications/ProdCom.app", CONFIG["fixes"]["dvs_ingest"]["run"]) # ProdCom after DVS
def auto_setup(self):
ran = []
self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) or True
comp = mock_comp()
return ran, comp
def test_auto_fix_sunday_service_gentle_fixes_only(self):
ran, comp = self.auto_setup()
self.snap.remote["tracks\0pgrep -xq Live"] = False # Ableton closed: gentle
self.snap.http["http://10.6.7.13:8080/"] = False # Bible app: restart = a hold
self.snap.conns["x32"]["category"] = "error" # a reconnect: gentle
t = SUNDAY # Sunday 10:00, alarms on
run_steps(self.engine, self.snap, t, 10)
self.assertEqual(self.engine.auto_fix(self.snap, comp, t + 10), []) # not 20 s yet
done = self.engine.auto_fix(self.snap, comp, t + 30)
time.sleep(0.2)
self.assertEqual(sorted(done), ["apps", "x32"])
self.assertEqual(ran, ["ableton"]) # never "bibleapp"
self.assertEqual(comp.restarted, ["id-x32"])
# while it works, the key says it is fixing itself
gate = threading.Event()
self.engine.executor = lambda cfg, fix_id, chk: gate.wait(2) or True
self.engine.auto_fix(self.snap, comp, t + 130) # second try, 90 s later
self.assertEqual(self.engine.notes["apps"]["text"], "AUTO-FIXING\nABLETON")
gate.set()
def test_auto_fix_writes_a_daily_log(self):
ran, comp = self.auto_setup()
t = SUNDAY
self.snap.remote["tracks\0pgrep -xq Live"] = False
run_steps(self.engine, self.snap, t, 25)
self.engine.auto_fix(self.snap, comp, t + 25)
time.sleep(0.2)
self.snap.remote["tracks\0pgrep -xq Live"] = True # Ableton is back
run_steps(self.engine, self.snap, t + 40, 1)
path = self.engine.autolog.path(t)
self.assertTrue(path.endswith("2026-10-11 Sunday Auto-Fix.log"))
lines = open(path).read().splitlines()
self.assertTrue(lines[0].startswith("Status panel auto-fixes - Sunday 11 October 2026"))
body = [l for l in lines if l[:2].isdigit()]
self.assertIn("PANEL auto-fix on until 13:30", body.pop(0))
self.assertEqual(body[0][:8], "08:00:25")
self.assertIn("APPS problem: ABLETON NOT RUNNING - fixing: start Ableton", body[0])
self.assertIn("(try 1 of 2)", body[0])
self.assertIn("-> done, checking it comes back", body[1])
self.assertIn("OK again: ABLETON NOT RUNNING (15 s after the auto-fix)", body[2])
def test_auto_fix_logs_go_after_a_month(self):
d = tempfile.mkdtemp()
names = ["2026-09-10 Thursday Auto-Fix.log", "2026-09-11 Friday Auto-Fix.log", # a month old
"2026-09-12 Saturday Auto-Fix.log", "2026-10-04 Sunday Auto-Fix.log", # not yet
"notes.txt", "2026-01-01 Thursday Auto-Fix.log.bak"] # not ours
for n in names:
open(os.path.join(d, n), "w").close()
removed = sp.AutoLog(d).cleanup(SUNDAY) # Sunday 11 Oct 2026
self.assertEqual(sorted(removed), names[:2])
self.assertEqual(sorted(os.listdir(d)), sorted(names[2:]))
open(os.path.join(d, "2026-01-31 Saturday Auto-Fix.log"), "w").close() # month ends: 28 Feb
self.assertEqual(sp.AutoLog(d).cleanup(datetime(2026, 2, 27, 9).timestamp()), [])
self.assertEqual(sp.AutoLog(d).cleanup(datetime(2026, 2, 28, 9).timestamp()),
["2026-01-31 Saturday Auto-Fix.log"])
def test_auto_fix_tries_twice_then_leaves_it_to_you(self):
ran, comp = self.auto_setup()
self.snap.remote["tracks\0pgrep -xq Live"] = False
t = SUNDAY
run_steps(self.engine, self.snap, t, 25)
for dt in (25, 60, 120, 230, 400): # 90 s apart, at most twice
self.engine.auto_fix(self.snap, comp, t + dt)
time.sleep(0.2)
self.assertEqual(ran, ["ableton", "ableton"])
self.snap.remote["tracks\0pgrep -xq Live"] = True # fixed (by hand)...
run_steps(self.engine, self.snap, t + 401, 2)
self.engine.auto_fix(self.snap, comp, t + 410)
self.assertIn(("fix", "ableton", ""), self.engine.auto_tries) # a blink does not reset it
self.engine.auto_fix(self.snap, comp, t + 1100)
self.assertEqual(self.engine.auto_tries, {}) # gone 10 min: fresh tries
def test_auto_fix_count_survives_its_own_fault_blinking_out(self):
ran, comp = self.auto_setup()
t = SUNDAY
self.snap.conns["x32"]["category"] = "error"
run_steps(self.engine, self.snap, t, 25)
for dt in range(25, 900, 30): # each reconnect: error -> starting -> error
self.snap.conns["x32"]["category"] = None if dt % 60 == 55 else "error"
run_steps(self.engine, self.snap, t + dt, 1)
self.engine.auto_fix(self.snap, comp, t + dt + 1)
self.assertEqual(comp.restarted, ["id-x32", "id-x32"]) # still only twice
def test_auto_fix_leaves_dante_alone_when_many_devices_vanish(self):
ran, comp = self.auto_setup()
t = SUNDAY
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(
[d for d in devices if d not in ("NA-Tracks", "NA-ProPresenter-Mac", "BOH-WING")])
run_steps(self.engine, self.snap, t, 30)
self.engine.auto_fix(self.snap, comp, t + 30)
self.assertEqual((ran, comp.restarted), ([], [])) # Ingest's view of Dante: hands off
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(
[d for d in devices if d != "NA-Tracks"]) # just the Tracks Mac's DVS
run_steps(self.engine, self.snap, t + 31, 1)
run_steps(self.engine, self.snap, t + 600, 30)
self.engine.auto_fix(self.snap, comp, t + 630)
time.sleep(0.2)
self.assertEqual(ran, ["dvs_tracks"])
self.assertNotIn("dvs_ingest", CONFIG["auto_fix"]["fixes"]) # it closes ProdCom
def test_midi_reconnects_run_one_after_another(self):
calls = []
def slow(cfg, fix_id, chk):
calls.append((threading.get_ident(), chk["peer"]))
time.sleep(0.05)
return True
self.engine.executor = slow
self.snap.midi["NA Ingest"]["peers"] = [] # Lighting PC and SuperRack gone
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
self.engine.run_fixes("midi", self.snap, None, t)
time.sleep(0.4)
self.assertEqual(sorted(p for _, p in calls), ["Lighting PC", "SuperRack-LiveBox"])
self.assertEqual(len({tid for tid, _ in calls}), 1) # one worker, not two at once
def test_restart_key_waits_while_auto_fix_starts_that_app(self):
ran, comp = self.auto_setup()
self.snap.remote["tracks\0pgrep -xq Live"] = False
run_steps(self.engine, self.snap, SUNDAY, 25)
self.engine.auto_fix(self.snap, comp, SUNDAY + 25) # auto-fix starts Ableton
self.assertIsNone(self.engine.run_force("rs_ableton", self.snap, None, SUNDAY + 30))
self.assertEqual(self.engine.notes["rs_ableton"]["text"], "STARTING\nWAIT 1 MIN")
time.sleep(0.2)
self.assertEqual(ran, ["ableton"]) # not killed while starting
def test_a_restarted_dead_connection_is_not_a_network_drop(self):
t = SUNDAY
run_steps(self.engine, self.snap, t, 2)
for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"): # dead for a long time
self.snap.conns[label]["category"] = "error"
run_steps(self.engine, self.snap, t + 100, 2)
self.engine.burst, self.engine.conn_drops = None, sp.deque()
for cat in (None, "error"): # auto restarts: starting, error
for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"):
self.snap.conns[label]["category"] = cat
tiles, t2 = run_steps(self.engine, self.snap, t + 200, 2)
self.assertEqual(tiles["net"]["x"], "NET\nOK")
def test_auto_fix_waits_out_the_network_and_respects_alarms_off(self):
ran, comp = self.auto_setup()
self.snap.remote["tracks\0pgrep -xq Live"] = False
t = SUNDAY
run_steps(self.engine, self.snap, t, 25)
self.engine.burst = {"start": t, "conns": set(), "cleared": None} # a network blip going on
self.assertEqual(self.engine.auto_fix(self.snap, comp, t + 30), [])
self.engine.burst = None
self.engine.service.set(False, t + 31) # ALARMS pressed: maintenance
self.assertEqual(self.engine.auto_fix(self.snap, comp, t + 40), [])
self.engine.service.set(True, t + 41)
weekday = SUNDAY - 3 * 86400 # Thursday: holds only
run_steps(self.engine, self.snap, weekday, 25)
self.assertEqual(self.engine.auto_fix(self.snap, comp, weekday + 30), [])
self.assertEqual(ran, [])
def test_restart_keys_flash_when_their_thing_is_down(self):
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
devices.remove("NA-Tracks") # DVS on the Tracks Mac gone
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
self.snap.remote["tracks\0pgrep -xq Live"] = False # Ableton not running
self.snap.conns["tascam"]["category"] = "error"
tiles, _ = run_steps(self.engine, self.snap, SUNDAY, 4)
flashing = {k for k, v in tiles.items() if k.startswith("rs_") and level(v) == "err"}
self.assertEqual(flashing, {"rs_dvs_tracks", "rs_ableton", "rs_tascam"})
self.assertEqual(tiles["rs_dvs_tracks"]["x"], "DVS TRACKS\nHOLD TO FIX")
self.assertEqual(level(tiles["rs_prodcom"]), "idle")
self.engine.ack("all") # acked on the status page:
tiles = self.engine.step(self.snap, SUNDAY + 5) # still shows what to hold
self.assertEqual(level(tiles["rs_ableton"]), "err")
def test_ableton_reopens_the_set_it_had_open(self):
for fix_id in ("ableton", "ableton_restart"):
run = CONFIG["fixes"][fix_id]["run"]
self.assertIn("Log.txt", run)
self.assertIn("grep -v '/Templates/'", run)
self.assertIn('open -a \'Ableton Live 11 Suite\' "$S"', run)
self.assertTrue(CONFIG["fixes"]["ableton"]["run"].startswith("pgrep -xq Live || {"))
def test_tascam_hold_always_reconnects(self):
ran = []
self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) or "HUNG - POWER\nCYCLE IT"
run_steps(self.engine, self.snap, SUNDAY, 4) # TASCAM looks fine
self.engine.run_fixes("tascam", self.snap, None, SUNDAY + 5)
time.sleep(0.2)
self.assertEqual(ran, ["tascam_reconnect"])
tiles = self.engine.step(self.snap, time.time())
self.assertEqual((tiles["tascam"]["x"], level(tiles["tascam"])), ("TASCAM\nHUNG - POWER\nCYCLE IT", "failed"))
def test_reconnect_and_verify(self):
from unittest import mock
fix = CONFIG["fixes"]["tascam_reconnect"]
conn = lambda cat, enabled=True: [{"label": "tascam", "id": "t1", "enabled": enabled, "status": {"category": cat}}]
def run(before, after, pings=True, restart_ok=True, enable_ok=True):
comp = mock.Mock()
comp.connections.side_effect = [before, after]
comp.restart_connection.return_value = restart_ok
comp.enable_connection.return_value = enable_ok
with mock.patch.object(sp, "Companion", return_value=comp), \
mock.patch.object(sp, "companion_conn_hosts", return_value={"tascam": "10.1.11.74"}), \
mock.patch.object(sp, "ping_once", return_value=pings):
return sp.reconnect_and_verify(CONFIG, fix, "x", sleep=lambda s: None), comp
r, comp = run(conn("error"), conn("good"))
self.assertTrue(r)
comp.restart_connection.assert_called_once_with("t1")
self.assertEqual(run(conn("error"), conn("error"))[0], "HUNG - POWER\nCYCLE IT") # pings: hung
self.assertEqual(run(conn("error"), conn("error"), pings=False)[0], "NO NETWORK\nCHECK POWER")
self.assertEqual(run(conn("good"), conn(None))[0], "STILL\nCONNECTING") # no verdict yet
self.assertFalse(run(conn("error"), conn("error"), restart_ok=False)[0])
r, comp = run(conn(None, enabled=False), conn("good")) # switched off
self.assertTrue(r)
comp.enable_connection.assert_called_once_with("t1")
comp.restart_connection.assert_not_called()
self.assertEqual(run(conn(None, enabled=False), conn(None), enable_ok=False)[0], "DISABLED IN\nCOMPANION")
self.assertFalse(run([], [])[0])
def test_stale_or_unstamped_holds_are_dropped(self):
ran = []
self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) or True
now = SUNDAY + 100
sp.handle_commands(";force rs_bridge %d;force rs_prodcom %d;force rs_proddash" % (now - 300, now - 2),
self.engine, None, self.snap, now)
time.sleep(0.2)
self.assertEqual(ran, ["prodcom"]) # only the fresh one: not the old or unstamped
def test_a_second_press_keeps_saying_restarting(self):
gate = threading.Event()
self.engine.executor = lambda cfg, fix_id, chk: gate.wait(2) or True
self.engine.run_force("rs_bridge", self.snap, None, SUNDAY)
self.engine.run_force("rs_bridge", self.snap, None, SUNDAY + 3)
self.assertEqual(self.engine.notes["rs_bridge"]["text"], "RESTARTING")
gate.set()
def test_group_button_names_the_device_that_needs_a_power_cycle(self):
self.engine.executor = lambda cfg, fix_id, chk: "HUNG - POWER\nCYCLE IT"
self.snap.conns["tascam"]["category"] = "error"
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
self.engine.run_fixes("rec", self.snap, None, t)
time.sleep(0.2)
self.assertEqual(self.engine.notes["rec"]["text"], "TASCAM HUNG - POWER\nCYCLE IT")
def test_windows_starts_check_the_app_really_started(self):
for fix_id, exe in (("obs", "obs64.exe"), ("arena", "Arena.exe"), ("vista", "Vista.exe"), ("bibleapp", "Bible App.exe")):
run = CONFIG["fixes"][fix_id]["run"]
self.assertTrue(run.rstrip(")").endswith('find /I "%s" >nul' % exe), fix_id)
self.assertIn("ping -n 9 127.0.0.1", run)
def test_ingest_resi_and_tascam_checks(self):
self.snap.http["http://127.0.0.1:24480/api/v1/ping"] = False
self.assertEqual(self.plan("ingest"), [("fix", "prodcom")])
self.snap.http["http://127.0.0.1:24480/api/v1/ping"] = True
self.snap.launchd.pop("org.waterschurch.lobby-captions-display")
self.assertEqual(self.plan("ingest"), [("fix", "caption_screen")]) # the screen only
self.snap.launchd["org.waterschurch.lobby-captions-display"] = 1
self.snap.ping["10.1.10.202"] = 3
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 4)
self.assertEqual(tiles["resi"]["x"], "RESI\nENCODER NO PING")
self.snap.conn_hosts = {"tascam": "10.1.11.74"}
self.snap.ping["10.1.11.74"] = 3
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 4)
self.assertEqual(tiles["tascam"]["x"], "TASCAM\nTASCAM NO PING")
def test_midi_reconnect_where_the_peer_was_last_seen(self):
others = [{"name": "SuperRack-LiveBox", "address": "10.1.13.140:5004", "state": "running"},
{"name": "Studio B Mac Studio", "address": "\tStudio B Mac Studio\tlocal.", "state": "running"}]
tracks = {"NA Tracks": {"port": 5004, "peers": others + [
{"name": "Lighting PC", "address": "10.1.10.77:5004", "state": "running"}]}}
sp.remember_peers(self.snap, "tracks", tracks) # seen at a new address earlier
tracks["NA Tracks"]["peers"] = list(others)
self.snap.remote_midi["tracks"] = tracks
self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"]
if p["name"] != "Lighting PC"]
run_steps(self.engine, self.snap, SUNDAY, 4)
plan = self.engine.fix_plan("midi", self.snap)
got = sorted((c.get("host", ""), c["session"], c["address"]) for _, kind, what, c in plan)
self.assertEqual(got, [("", "NA Ingest", "10.1.10.206:5004"), # never seen: config address
("tracks", "NA Tracks", "10.1.10.77:5004")]) # seen: last address
self.assertEqual(self.plan("show"), [("fix", "midi_reconnect")] * 2) # one per session
def test_midi_reconnect_runs_on_the_right_mac(self):
from unittest import mock
chk = {"type": "midi_peer", "host": "tracks", "session": "NA Tracks", "peer": "Lighting PC",
"address": "10.1.10.206:5004"}
done = mock.Mock(returncode=0, stdout='SPMIDI"asked"\n', stderr="")
with mock.patch.object(sp.subprocess, "run", return_value=done) as run:
self.assertTrue(sp.execute_fix(CONFIG, "midi_reconnect", chk))
argv, kw = run.call_args.args[0], run.call_args.kwargs
self.assertIn("waterschurchmedia@10.1.3.227", argv)
self.assertIn("ns['connect_peer']('NA Tracks', 'Lighting PC', '10.1.10.206:5004')", kw["input"])
failed = mock.Mock(returncode=255, stdout="", stderr="ssh: timed out")
with mock.patch.object(sp.subprocess, "run", return_value=failed):
self.assertFalse(sp.execute_fix(CONFIG, "midi_reconnect", chk))
def test_reconnect_leaves_other_peers_alone(self):
import coremidi
sess = {"dns-sd-name": "NA Tracks", "flags": 1, "join-policy": 2, "port": 5004, "peers": [
{"address": "10.1.13.140:5004", "name": "SuperRack-LiveBox", "state": "running"},
{"address": "\tStudio B Mac Studio\tlocal.", "name": "Studio B Mac Studio", "state": "running"}]}
new = coremidi.with_peer(sess, "Lighting PC", "10.1.10.206:5004")
self.assertEqual(new["peers"][:2], sess["peers"])
self.assertEqual(new["peers"][2], {"address": "10.1.10.206:5004", "name": "Lighting PC"})
self.assertEqual({k: v for k, v in new.items() if k != "peers"}, {k: v for k, v in sess.items() if k != "peers"})
self.assertIsNone(coremidi.with_peer(new, "Lighting PC", "10.1.10.206:5004")) # already there
back = coremidi.without_peer(new, "Lighting PC")
self.assertEqual(back, sess)
self.assertIsNone(coremidi.without_peer(sess, "Lighting PC")) # not there
def test_a_peer_stuck_initializing_is_kicked(self):
import coremidi
from unittest import mock
sess = {"dns-sd-name": "NA Ingest", "port": 5009, "peers": [
{"name": "Lighting PC", "address": "10.1.10.206:5004", "state": "running"},
{"name": "SuperRack-LiveBox", "address": "10.1.13.140:5004", "state": "initializing"}]}
self.assertTrue(coremidi.stuck_peer(sess, "SuperRack-LiveBox"))
self.assertFalse(coremidi.stuck_peer(sess, "Lighting PC"))
writes = []
with mock.patch.object(coremidi, "_session_entities", return_value=[(1, sess)]), \
mock.patch.object(coremidi, "_string_prop", return_value=None), \
mock.patch.object(coremidi, "_set_dict_prop", side_effect=lambda e, k, v: writes.append(v) or 0):
self.assertEqual(coremidi.connect_peer("NA Ingest", "SuperRack-LiveBox", "10.1.13.140:5004",
settle=lambda s: None), "kicked")
self.assertEqual(coremidi.connect_peer("NA Ingest", "Lighting PC", "10.1.10.206:5004"), "already")
self.assertEqual([[p["name"] for p in w["peers"]] for w in writes],
[["Lighting PC"], ["Lighting PC", "SuperRack-LiveBox"]]) # dropped, then added back
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.assertEqual(sp.execute_fix(CONFIG, "arena", {"conn": "arena"}), # no MAC: says so
"ARENA NOT\nREACHABLE")
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_an_unreachable_pc_is_named_not_failed(self):
from unittest import mock
gone = mock.Mock(returncode=255, stdout="", stderr="ssh: connect to host 10.6.7.13 port 22: Operation timed out")
with mock.patch.object(sp.subprocess, "run", return_value=gone):
self.assertEqual(sp.execute_fix(CONFIG, "bibleapp", {}), "BIBLE APP NOT\nREACHABLE")
refused = mock.Mock(returncode=1, stdout="", stderr="") # reached it, the command failed
with mock.patch.object(sp.subprocess, "run", return_value=refused):
self.assertFalse(sp.execute_fix(CONFIG, "bibleapp", {}))
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_hd_hold_works_even_when_it_looks_green(self):
ran = []
self.engine.executor = lambda cfg, fix_id, chk: ran.append((fix_id, chk.get("conn"), chk.get("ip"))) or True
run_steps(self.engine, self.snap, SUNDAY, 4) # every deck fine
self.engine.run_fixes("hd4", self.snap, None, SUNDAY + 5)
self.snap.vars[("HyperDeck_3", "status")] = "Record"
self.engine.run_fixes("hd3", self.snap, None, SUNDAY + 5) # recording: left alone
time.sleep(0.2)
self.assertEqual(ran, [("hyperdeck", "HyperDeck_4", "10.1.10.14")])
self.assertEqual(self.engine.notes["hd3"]["text"], "RECORDING\nLEFT ALONE")
self.engine.run_fixes("hd5", self.snap, None, SUNDAY + 6) # another deck: its own guard
self.engine.run_fixes("hd4", self.snap, None, SUNDAY + 20) # same deck within a minute
time.sleep(0.2)
self.assertEqual([r[1] for r in ran], ["HyperDeck_4", "HyperDeck_5"])
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)
self.snap.conns_time = time.time() # fresh data at the real clock
tiles = self.engine.step(self.snap, time.time())
self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nCHECKING\nDVS", "fixing"))
self.snap.conns_time = time.time() + 25 # fresh data at the real clock
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)
self.snap.conns_time = time.time() # fresh data at the real clock
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.mdns = self.snap.mdns - {"NA Tracks"} # an advert gone: nothing to reconnect
_, 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_windows_remote_checks_use_conditional_echoes(self):
line = sp.remote_script(["tasklist | find /I \"Vista.exe\"", "sc query rtpMIDIService | find \"RUNNING\""], "cmd")
self.assertNotIn("%ERRORLEVEL%", line)
self.assertIn("&& echo SP0=0 || echo SP0=1", line)
self.assertIn("&& echo SP1=0 || echo SP1=1", line)
self.assertEqual(sp.parse_remote("SP0=0\r\nSP1=1\r\n", 2), [True, False])
self.assertEqual(sp.host_shell(CONFIG, "lighting"), "cmd")
self.assertEqual(sp.host_shell(CONFIG, "tracks"), "sh")
def test_overnight_windows(self):
sat_2350 = datetime(2026, 10, 10, 23, 50).timestamp()
sun_0020 = datetime(2026, 10, 11, 0, 20).timestamp()
sun_0100 = datetime(2026, 10, 11, 1, 0).timestamp()
w = [{"days": ["sat"], "start": "18:00", "end": "00:30"}]
self.assertTrue(sp.in_windows(w, sat_2350))
self.assertTrue(sp.in_windows(w, sun_0020)) # past midnight, still Saturday's service
self.assertFalse(sp.in_windows(w, sun_0100))
self.assertFalse(sp.in_windows([{"days": ["sun"], "start": "18:00", "end": "00:30"}], sat_2350))
def test_check_identity_is_about_the_thing_not_the_position(self):
midi = [c for c in CONFIG["tiles"][9]["checks"] if c["type"] == "midi_peer"]
self.assertEqual(len({sp.check_identity(c) for c in midi}), len(midi)) # six peers, six identities
self.assertEqual(sp.check_identity({"type": "midi_peer", "host": "tracks", "session": "NA Tracks", "peer": "Lighting PC"}),
"tracks/NA Tracks/Lighting PC")
self.assertEqual(sp.check_identity({"type": "ping", "conn": "Axient_1", "host": "10.1.1.101"}), "Axient_1")
self.assertEqual(sp.check_identity({"type": "conn", "conn": "ATEM"}), "ATEM")
def test_acknowledgements_survive_a_restart(self):
state = os.path.join(tempfile.mkdtemp(prefix="sp-state-"), "state.json")
engine = sp.Engine(CONFIG, state)
engine.executor = engine.poe_executor = lambda *a: True
snap = healthy_snapshot(SUNDAY)
snap.conns["ATEM"]["category"] = "error"
run_steps(engine, snap, SUNDAY, 5)
engine.ack("atem")
self.assertEqual(level(run_steps(engine, snap, SUNDAY + 6, 1)[0]["atem"]), "ack")
saved = json.load(open(state))
self.assertIn("atem|ATEM|conn", saved["acks"])
again = sp.Engine(CONFIG, state) # the watchdog restarts
again.executor = again.poe_executor = lambda *a: True
tiles, _ = run_steps(again, snap, SUNDAY + 10, 5)
self.assertEqual(level(tiles["atem"]), "ack") # still acknowledged, no new flashing
snap.conns["ATEM"]["category"] = "good"
run_steps(again, snap, SUNDAY + 20, 2)
self.assertEqual(json.load(open(state))["acks"], []) # cleared: forgotten
def test_run_this_fix_now_goes_through_the_hold_path(self):
ran = []
self.engine.executor = lambda cfg, fix_id, chk: ran.append((fix_id, chk.get("conn"), chk.get("ip"), chk.get("peer"))) or True
snap = self.snap
run_steps(self.engine, snap, SUNDAY, 2)
comp = mock_comp()
now = SUNDAY + 3
self.engine.run_test("hd2", "hyperdeck", "0", snap, comp, now) # the check's number...
self.engine.run_test("midi", "midi_reconnect", "tracks/NA Tracks/Studio B Mac Studio", snap, comp, now) # ...or its identity
self.engine.run_test("midi", "midi_reconnect", "9", snap, comp, now) # no such check
self.engine.run_test("atem", "atem_audio", "1", snap, comp, now) # never from the page
self.engine.run_test("hd2", "propresenter", "0", snap, comp, now) # not attached to that key
for t in self.engine.fix_threads:
t.join(2)
self.assertEqual(ran, [("hyperdeck", "HyperDeck_2", "10.1.10.12", None),
("midi_reconnect", None, None, "Studio B Mac Studio")])
self.assertEqual(self.engine.notes["hd2"]["text"], "DONE")
def test_a_restart_lets_the_running_fix_finish_and_starts_nothing_new(self):
gate = threading.Event()
self.engine.executor = lambda cfg, fix_id, chk: gate.wait(3) or True
run_steps(self.engine, self.snap, SUNDAY, 2)
self.engine.run_force("rs_bridge", self.snap, mock_comp(), SUNDAY + 3)
self.assertEqual(len(self.engine.running), 1) # published for the settings page
self.engine.stopping = True # SIGTERM arrived
self.engine.run_force("rs_captions", self.snap, mock_comp(), SUNDAY + 4)
self.assertEqual(self.engine.notes["rs_captions"]["text"], "RESTARTING\nSOON")
self.assertEqual(self.engine.run_fixes("ingest", self.snap, mock_comp(), SUNDAY + 4), [])
gate.set()
for t in self.engine.fix_threads:
t.join(2)
self.assertEqual(self.engine.running, {})
self.assertEqual(sp.exit_timeout(CONFIG), 90) # the longest fix + its verify + 10
def test_a_hyperdeck_switched_off_is_always_switched_back_on(self):
calls = []
class Comp:
def connections(self):
return [{"label": "HyperDeck_2", "id": "x", "enabled": True, "status": {"category": "good"}}]
def disable_connection(self, cid):
calls.append("off")
return True
def enable_connection(self, cid):
calls.append("on")
return True
import unittest.mock as mock
def boom(seconds):
raise RuntimeError("killed mid-way")
with mock.patch.object(sp, "Companion", lambda *a, **k: Comp()):
with self.assertRaises(RuntimeError):
sp.reconnect_and_verify(CONFIG, CONFIG["fixes"]["hyperdeck"], "x", sleep=boom, conn="HyperDeck_2", toggle=True)
self.assertEqual(calls, ["off", "on"])
def test_spotify_fix_opens_and_plays(self):
calls = []
class R:
def __init__(self, code, out=""):
self.returncode, self.stdout, self.stderr = code, out, ""
def run(argv, **kw):
calls.append(argv)
if argv[1] == "-xq":
return R(1) # Spotify is closed
if "player state" in argv[-1]:
return R(0, "playing\n")
return R(0)
import unittest.mock as mock
with mock.patch.object(sp.time, "sleep", lambda s: None):
ok = sp.play_spotify(CONFIG, CONFIG["fixes"]["spotify_play"], {}, "spotify", run=run)
self.assertTrue(ok)
self.assertEqual(calls[1][:3], ["/usr/bin/open", "-g", "-a"])
self.assertIn(CONFIG["spotify"]["playlist"], calls[2][-1])
def test_never_auto_fixes_stay_a_hold(self):
cfg = json.loads(json.dumps(CONFIG))
cfg["fixes"]["ableton"]["never_auto"] = True
cfg["auto_fix"]["fixes"].remove("ableton")
engine = sp.Engine(cfg, None)
self.assertNotIn("ableton", {f for f in engine.auto.get("fixes", []) if not engine.cfg["fixes"][f].get("never_auto")})
cfg["auto_fix"]["fixes"].append("ableton")
path = os.path.join(tempfile.mkdtemp(prefix="sp-cfg-"), "config.json")
json.dump(cfg, open(path, "w"))
with self.assertRaises(ValueError):
sp.load_config(path)
def test_snapshot_summary_is_plain_json(self):
snap = self.snap
snap.conns["ATEM"]["category"] = "error"
run_steps(self.engine, snap, SUNDAY, 5)
out = sp.snapshot_summary(snap, self.engine, SUNDAY + 5)
json.dumps(out) # nothing in it that JSON cannot hold
self.assertEqual(out["conns"]["ATEM"]["category"], "error")
self.assertEqual(out["faults"]["atem"][0]["msg"], "OFFLINE")
self.assertEqual(out["faults"]["atem"][0]["check"], "ATEM")
self.assertEqual(out["live"], [False, ""])
self.assertEqual(out["midi"]["NA Ingest"]["peers"][0]["name"], "Lighting PC")
def test_every_fix_is_valid(self):
for f in CONFIG["fixes"].values():
self.assertIn(f.get("kind", "ssh"), sp.FIX_KINDS)
def run_failover(engine, snap, start, seconds, step=0.5):
"""Like run_steps, but the failover looks after every step, as the run loop does."""
tiles, t = None, start
while t <= start + seconds:
snap.conns_time = t
tiles = engine.step(snap, t)
engine.failover(snap, None, t)
t += step
return tiles, t
STB_PING, STB_PT = "10.1.13.175", "studiob\0pgrep -xq 'Pro Tools'"
class TestFailover(unittest.TestCase):
"""Studio B dies on a Sunday: the FOH Backup button is pressed by itself, once, and the keys say so.
Nothing here presses a real button, posts a real alert or writes a real log."""
def setUp(self):
self.snap = healthy_snapshot(SUNDAY)
self.engine = sp.Engine(CONFIG)
self.ran, self.sent = [], []
self.engine.executor = lambda cfg, fix_id, chk: self.ran.append(fix_id) or True
self.engine.poe_executor = lambda *a: True
self.engine.notify = lambda url, text: self.sent.append(text)
self.engine.autolog = sp.AutoLog(tempfile.mkdtemp())
def fired_at(self):
self.engine.fix_threads and [t.join(2) for t in self.engine.fix_threads]
rec = self.engine.failed.get("stream_foh_backup")
return rec and rec["at"]
def test_config_wires_studio_b_to_foh_backup(self):
tile = next(t for t in CONFIG["tiles"] if t["id"] == "stb_audio")
self.assertNotIn("pos", tile) # hidden: AUDIO shows it
self.assertIn("stb_audio", next(t for t in CONFIG["tiles"] if t["id"] == "audio")["members"])
for c in tile["checks"]:
fixes = set(c["fix"].values()) if isinstance(c["fix"], dict) else {c["fix"]}
self.assertEqual(fixes, {"stream_foh_backup"}, c["type"])
fix = CONFIG["fixes"]["stream_foh_backup"]
self.assertEqual((fix["kind"], fix["page"], fix["row"], fix["col"]), ("companion_press", 4, 0, 5))
self.assertTrue(fix["failover"])
self.assertEqual(fix["undo"], "stream_studio_b")
self.assertFalse(CONFIG["fixes"]["stream_studio_b"].get("failover"))
self.assertNotIn("stream_foh_backup", CONFIG["auto_fix"]["fixes"])
keys = {k["id"]: k["fix"] for k in CONFIG["restart_page"]["keys"]}
self.assertEqual((keys["rs_stream_stb"], keys["rs_stream_foh"]), ("stream_studio_b", "stream_foh_backup"))
self.assertEqual(CONFIG["failover"]["after_seconds"], 2)
self.assertEqual(self.engine.failover_windows, [{"days": ["sun"], "start": "06:30", "end": "13:30"}])
def test_studio_b_dies_foh_backup_is_pressed_once(self):
t0 = SUNDAY
run_failover(self.engine, self.snap, t0 - 5, 4) # all fine for a while
self.snap.ping[STB_PING] = 2 # two 1-second pings missed
tiles, t = run_failover(self.engine, self.snap, t0, 6)
at = self.fired_at()
self.assertEqual(self.ran, ["stream_foh_backup"])
key = next(k for k, (tid, chk, sub) in self.engine.meta.items() if chk.get("host") == STB_PING)
since = self.engine.trackers[key].since
self.assertEqual(since, t0 + 1) # 1 s debounce
self.assertEqual(at, since + 2) # + 2 s after_seconds
self.assertLessEqual(at - t0, 5)
tiles, t = run_failover(self.engine, self.snap, t, 60) # the fault lasts a minute
self.assertEqual(self.ran, ["stream_foh_backup"]) # still once
self.assertNotIn("stream_studio_b", self.ran) # never switched back by itself
# the keys say where the stream is
self.assertEqual(tiles["audio"]["x"], "AUDIO\nSTUDIO B NO PING\nON FOH BACKUP")
self.assertEqual(level(tiles["audio"]), "err")
self.engine.notes.pop("stb_audio", None) # under the fix's own feedback
tiles = self.engine.step(self.snap, t)
self.assertEqual(tiles["stb_audio"]["x"], "STB AUDIO\nSTUDIO B NO PING\nON FOH BACKUP")
self.assertEqual(tiles["all"]["x"], "1 DOWN\nSTUDIO B NO PING")
# the log, the alert, and a hold right after
body = [l for l in open(self.engine.autolog.path(t0)).read().splitlines() if l[:2].isdigit()]
self.assertTrue(any("STB AUDIO FAILOVER: STUDIO B NO PING - press the FOH Backup button" in l for l in body), body)
self.assertTrue(any("-> done, checking it comes back" in l for l in body), body)
self.assertEqual(len(self.sent), 1)
self.assertTrue(self.sent[0].startswith("NA Ingest: STB AUDIO STUDIO B NO PING - switched over by itself: press the FOH Backup"))
self.assertIn("hold STREAM STB on the Restart page", self.sent[0])
self.engine.run_fixes("audio", self.snap, None, at + 10)
self.assertEqual(self.engine.notes["audio"]["text"], "JUST TRIED\nWAIT 1 MIN")
self.assertIsNone(self.engine.run_force("rs_stream_foh", self.snap, None, at + 11))
self.assertEqual(self.engine.notes["rs_stream_foh"]["text"], "JUST DONE\nWAIT 1 MIN")
out = sp.snapshot_summary(self.snap, self.engine, t)
json.dumps(out)
self.assertEqual(out["failover"]["stream_foh_backup"]["note"], "ON FOH BACKUP")
self.assertEqual(out["failover"]["stream_foh_backup"]["faults"], ["STUDIO B NO PING"])
def test_pro_tools_crash_is_as_fast(self):
t0 = SUNDAY
self.snap.remote[STB_PT] = False
run_failover(self.engine, self.snap, t0, 6)
self.assertEqual(self.ran, ["stream_foh_backup"])
self.assertLessEqual(self.fired_at() - t0, 5)
tiles = self.engine.step(self.snap, t0 + 7)
self.assertEqual(tiles["audio"]["x"], "AUDIO\nPRO TOOLS NOT RUNNING\nON FOH BACKUP")
def test_fires_with_alarms_muted(self):
self.engine.service.set(False, SUNDAY) # maintenance: auto-fix stops
self.snap.ping[STB_PING] = 2
run_failover(self.engine, self.snap, SUNDAY, 6)
self.assertEqual(self.ran, ["stream_foh_backup"]) # the failover does not
def test_only_inside_the_window(self):
for when in (WEDNESDAY, datetime(2026, 10, 11, 15, 0).timestamp()): # a weekday; Sunday after the window
snap = healthy_snapshot(when)
snap.ping[STB_PING] = 2
run_failover(self.engine, snap, when, 10)
self.assertEqual(self.ran, [], datetime.fromtimestamp(when))
self.assertEqual(self.engine.failed, {})
def test_rearms_when_the_fault_clears_and_the_note_fades(self):
t0 = SUNDAY
self.snap.ping[STB_PING] = 2
tiles, t = run_failover(self.engine, self.snap, t0, 6)
self.snap.ping[STB_PING] = 0 # Studio B answers again
tiles, t = run_failover(self.engine, self.snap, t, 5)
self.assertEqual(tiles["audio"]["x"], "AUDIO\nOK\nON FOH BACKUP") # still on the backup
self.assertEqual(level(tiles["audio"]), "ok")
self.assertIsNotNone(self.engine.failed["stream_foh_backup"]["cleared"])
self.snap.ping[STB_PING] = 2 # ...and dies again 5 s later
tiles, t = run_failover(self.engine, self.snap, t, 4)
self.assertEqual(self.ran, ["stream_foh_backup"]) # was clear under 10 s: not armed yet
tiles, t = run_failover(self.engine, self.snap, t, 3) # 10 s since it cleared: armed, fires
self.assertEqual(self.ran, ["stream_foh_backup", "stream_foh_backup"])
self.snap.ping[STB_PING] = 0
tiles, t = run_failover(self.engine, self.snap, t, 2)
tiles, t = run_failover(self.engine, self.snap, t + 1000, 1) # note_seconds later: forgotten
self.assertEqual(tiles["audio"]["x"], "AUDIO\nOK")
self.assertEqual(self.engine.failed, {})
def test_a_check_that_cannot_ask_is_not_a_clear(self):
t0 = SUNDAY
self.snap.remote[STB_PT] = False # Pro Tools gone
tiles, t = run_failover(self.engine, self.snap, t0, 6)
self.assertEqual(self.ran, ["stream_foh_backup"])
self.snap.remote[STB_PT] = None # ssh could not ask for a while
tiles, t = run_failover(self.engine, self.snap, t, 30)
self.assertIsNone(self.engine.failed["stream_foh_backup"]["cleared"]) # not re-armed on a hiccup
self.snap.remote[STB_PT] = False # the answer is back: still gone
tiles, t = run_failover(self.engine, self.snap, t, 10)
self.assertEqual(self.ran, ["stream_foh_backup"]) # no second press
self.assertEqual(tiles["rs_stream_foh"]["x"], "STREAM FOH\nON FOH BACKUP") # pressed already: no HOLD TO FIX
self.assertEqual(level(tiles["rs_stream_foh"]), "ok")
self.assertEqual(tiles["rs_stream_stb"]["x"], "STREAM STB\nHOLD 2s")
def test_studio_b_rednet_is_never_power_cycled_by_itself(self):
"""Only STUDIO-B-RedNet-TNX vanishes: the failover presses FOH Backup; Sunday auto-fix on the DANTE
key must not power-cycle the RedNet's PoE port (that stays a hold), nor touch anything else."""
comp = mock_comp()
self.snap.dante_map = {"STUDIO-B-RedNet-TNX": {"mac": "00:1d:c1:aa:bb:cc", "poe": True, "switch": "10.1.60.2", "port": "0/14"}}
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps([d for d in devices if d != "STUDIO-B-RedNet-TNX"])
poe = []
self.engine.poe_executor = lambda cfg, name, entry: poe.append(name) or True
tiles, t = run_failover(self.engine, self.snap, SUNDAY, 30)
self.assertEqual(self.ran, ["stream_foh_backup"])
self.assertEqual(self.engine.auto_fix(self.snap, comp, t), [])
time.sleep(0.2)
self.assertEqual((poe, comp.restarted, self.ran), ([], [], ["stream_foh_backup"]))
self.assertEqual([(k, w) for _, k, w, _ in self.engine.fix_plan("dante", self.snap)], [("poe", "STUDIO-B-RedNet-TNX")]) # a hold would
def test_switching_back_by_hand_clears_the_note(self):
t0 = SUNDAY
gate = threading.Event() # the Studio B press takes a moment
def press(cfg, fix_id, chk):
if fix_id == "stream_studio_b":
gate.wait(2)
self.ran.append(fix_id)
return True
self.engine.executor = press
self.snap.ping[STB_PING] = 2
tiles, t = run_failover(self.engine, self.snap, t0, 6)
self.assertIn("ON FOH BACKUP", tiles["audio"]["x"])
self.assertEqual(self.engine.run_force("rs_stream_stb", self.snap, None, t), "stream_studio_b")
self.assertEqual(self.engine.notes["rs_stream_stb"]["text"], "PRESSING")
gate.set()
time.sleep(0.2)
self.assertEqual(self.engine.notes["rs_stream_stb"]["text"], "DONE")
self.assertEqual(self.engine.failed, {})
tiles = self.engine.step(self.snap, t + 0.5)
self.assertEqual(tiles["audio"]["x"], "AUDIO\nSTUDIO B NO PING")
tiles, t = run_failover(self.engine, self.snap, t + 1, 4) # still dead: it fires again
self.assertEqual(self.ran, ["stream_foh_backup", "stream_studio_b", "stream_foh_backup"])
self.assertIn("ON FOH BACKUP", tiles["audio"]["x"])
def test_held_back_when_several_dante_devices_vanish(self):
t0 = SUNDAY
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(
[d for d in devices if d not in ("STUDIO-B-RedNet-TNX", "NA-Tracks", "BOH-WING")])
tiles, t = run_failover(self.engine, self.snap, t0, 12)
self.assertEqual(self.ran, []) # Ingest's view of Dante: hands off
self.assertEqual(tiles["stb_audio"]["x"], "STB AUDIO\nSTB TNX GONE") # seen, no note
body = open(self.engine.autolog.path(t0)).read()
self.assertIn("failover held back: 3 Dante devices gone at once", body)
self.assertEqual(body.count("held back"), 1) # said once, not every half second
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(
[d for d in devices if d != "STUDIO-B-RedNet-TNX"]) # only Studio B's own box
run_failover(self.engine, self.snap, t, 3)
self.assertEqual(self.ran, ["stream_foh_backup"])
def test_state_survives_a_restart(self):
state = os.path.join(tempfile.mkdtemp(prefix="sp-state-"), "state.json")
engine = sp.Engine(CONFIG, state)
engine.executor = lambda cfg, fix_id, chk: self.ran.append(fix_id) or True
engine.notify = lambda url, text: None
engine.autolog = sp.AutoLog(tempfile.mkdtemp())
self.snap.ping[STB_PING] = 2
tiles, t = run_failover(engine, self.snap, SUNDAY, 6)
self.assertEqual(self.ran, ["stream_foh_backup"])
self.assertEqual(json.load(open(state))["failover"]["stream_foh_backup"]["tile"], "stb_audio")
again = sp.Engine(CONFIG, state) # the watchdog restarts mid-fault
again.executor = lambda cfg, fix_id, chk: self.ran.append(fix_id) or True
again.notify = lambda url, text: None
again.autolog = sp.AutoLog(tempfile.mkdtemp())
tiles, t = run_failover(again, self.snap, t, 10)
self.assertEqual(self.ran, ["stream_foh_backup"]) # not pressed a second time
self.assertEqual(tiles["audio"]["x"], "AUDIO\nSTUDIO B NO PING\nON FOH BACKUP")
def test_nothing_new_starts_while_stopping(self):
self.snap.ping[STB_PING] = 2
self.engine.stopping = True
run_failover(self.engine, self.snap, SUNDAY, 6)
self.assertEqual(self.ran, [])
def test_load_config_checks_the_button_and_the_undo(self):
def load(change):
cfg = json.loads(json.dumps(CONFIG))
change(cfg)
path = os.path.join(tempfile.mkdtemp(prefix="sp-cfg-"), "config.json")
with open(path, "w") as f:
json.dump(cfg, f)
return sp.load_config(path)
load(lambda c: None)
for bad in (lambda c: c["fixes"]["stream_foh_backup"].update(page=0),
lambda c: c["fixes"]["stream_foh_backup"].update(row=True),
lambda c: c["fixes"]["stream_foh_backup"].update(col="5"),
lambda c: c["fixes"]["stream_foh_backup"].pop("col"),
lambda c: c["fixes"]["stream_foh_backup"].update(never_auto=True),
lambda c: c["fixes"]["stream_foh_backup"].update(undo="nope"),
lambda c: c["failover"]["windows"].append({"days": ["sun"], "start": "6:30", "end": "13:30"})):
with self.assertRaises(ValueError):
load(bad)
def test_studio_b_is_pinged_every_second(self):
self.assertEqual(sp.ping_hosts(CONFIG, fast=True), [STB_PING])
slow = sp.ping_hosts(CONFIG)
self.assertNotIn(STB_PING, slow)
self.assertIn("10.1.0.1", slow) # the gateway stays in the slow list
self.assertIn("10.1.3.227", slow)
self.assertEqual(sp.PING_INTERVAL, 3)
def test_remote_groups_run_when_due_each_on_its_own(self):
from unittest import mock
groups = sp.remote_groups(CONFIG)
self.assertEqual(groups[("studiob", 2)], ["pgrep -xq 'Pro Tools'"])
self.assertTrue(all(k[1] == 10 for k in groups if k != ("studiob", 2)))
calls = []
fake = lambda argv, **kw: calls.append(argv) or mock.Mock(returncode=0, stdout="SP0=0\nSP1=0\n", stderr="")
t = SUNDAY + 100
with mock.patch.object(sp.subprocess, "run", fake):
last = {k: t - 3 for k in groups} # every group ran 3 s ago
for th in sp.poll_remote(self.snap, CONFIG, now=t, last=last):
th.join(2)
self.assertEqual(len(calls), 1) # only the 2-second group is due
self.assertIn("studiob@10.1.13.175", calls[0])
self.assertIn("pgrep -xq 'Pro Tools'", calls[0][-1])
self.assertTrue(self.snap.remote[STB_PT])
self.assertEqual(last[("studiob", 2)], t)
self.assertEqual(sp.poll_remote(self.snap, CONFIG, now=t + 1, last=last), []) # nothing due yet
calls.clear()
sp.poll_remote(self.snap, CONFIG) # run_once: everything, and waits
self.assertEqual(len(calls), len(groups))
self.assertEqual(sp.REMOTE_INTERVAL, 10)
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")
body = self.rfile.read(int(self.headers.get("Content-Length") or 0)).decode()
query = urllib.parse.parse_qs(u.query, keep_blank_values=True)
if not body and "value" not in query:
return self._send(400, "No value") # like Companion 5.0.7 with an empty body
s["custom"][m.group(1)] = body if body else query["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, "{}")
m = re.match(r"^/api/location/(\d+)/(\d+)/(\d+)/press$", u.path)
if m:
where = tuple(int(x) for x in m.groups())
s["presses"].append(where)
if where in s["buttons"]:
return self._send(200, "ok")
return self._send(204, "No control") # like Companion 5.0.7 for an empty cell
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": [],
"presses": [],
"buttons": {(4, 0, 5), (4, 0, 6)}, # the Audio page's FOH Backup and Studio B
}
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_press_a_button(self):
comp = sp.Companion("127.0.0.1", self.port)
store = FakeCompanion.store
self.assertEqual(comp.press(4, 0, 5), 200)
self.assertEqual(comp.press(4, 1, 1), 204) # an empty cell
self.assertEqual(store["presses"][-2:], [(4, 0, 5), (4, 1, 1)])
cfg = json.loads(json.dumps(CONFIG))
cfg["companion"]["port"] = self.port
self.assertTrue(sp.execute_fix(cfg, "stream_foh_backup", {}))
self.assertEqual(store["presses"][-1], (4, 0, 5))
cfg["fixes"]["stream_foh_backup"]["col"] = 7
self.assertEqual(sp.execute_fix(cfg, "stream_foh_backup", {}), "NO BUTTON\nTHERE")
self.assertEqual(store["presses"][-1], (4, 0, 7))
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"])
# x32 is green, so its hold does nothing; LED is failing on VX1000, so that restarts.
sp.handle_commands(";restart x32;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.pop("atem", None) # never connect to the real switcher from a test
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.start_remote_midi_streams)
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, **k: [] # never SSH anywhere from a test
sp.start_remote_midi_streams = lambda snap, cfg, stop: []
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,
sp.start_remote_midi_streams) = 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) + 2) # + the page number and RESTART keys
jump = self.controls["2"]["6"]["steps"]["0"]["action_sets"]["up"][0]
self.assertEqual((jump["definitionId"], jump["options"]["page"]["value"]), ("set_page", 20))
# The master is where the smallest Stream Deck can see it.
self.assertEqual(CONFIG["tiles"][0]["pos"], [0, 1])
def test_restart_page(self):
export = make_page.build_restart(CONFIG)
self.assertEqual((export["oldPageNumber"], export["page"]["name"]), (20, "20\\nRestart"))
c = export["page"]["controls"]
holds = {}
for r in c.values():
for ctl in r.values():
sets = ctl.get("steps", {}).get("0", {}).get("action_sets", {})
if "2000" in sets:
holds[sets["2000"][0]["options"]["value"]["value"]] = sets["up"]
self.assertEqual(len(holds), len(CONFIG["restart_page"]["keys"]))
self.assertIn("$(custom:sp_cmd);force rs_tascam $(internal:time_unix)", holds)
self.assertTrue(all(up == [] for up in holds.values())) # a tap does nothing
back = c["2"]["6"]["steps"]["0"]["action_sets"]["up"][0] # where the user put it
self.assertEqual((back["definitionId"], back["options"]["page"]["value"]), ("set_page", 19))
def test_push_page_looks_at_what_is_on_the_page(self):
import push_page
nav = {"0": {"0": {"type": "pageup"}}, "1": {"0": {"type": "button-layered",
"options": {"notes": "STATUS nav key (pagenum): x"}}}}
self.assertTrue(push_page.safe_target({"page": {"name": "PAGE", "controls": nav}}, "23\\nRestart")[0])
ours = {"page": {"name": "whatever", "controls": dict(nav, **{"3": {"7": {"type": "button-layered",
"options": {"notes": "STATUS tile 'led' - drawn by status_panel.py on NA Ingest."}},
"6": {"type": "button-layered", "options": {"notes": "STATUS jump key - Tap: open the Restart page."}}}})}}
self.assertTrue(push_page.safe_target(ours, "23\\nRestart")[0]) # our keys, whatever the page is called
busy = {"page": {"name": "23\\nRestart", "controls": dict(nav, **{"2": {"3": {"type": "button"}}})}}
ok, why = push_page.safe_target(busy, "23\\nRestart")
self.assertFalse(ok) # a key someone made by hand: refused
self.assertIn("row 2 col 3", why) # ...and named by its place
self.assertIn("move or delete them first", why)
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 $(internal:time_unix)")
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.assertEqual(net["action_sets"]["2000"][0]["options"]["value"]["value"], # hold: clear a blip
"$(custom:sp_cmd);restart net $(internal:time_unix)")
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 View Git Blame Copy Permalink