christianandClaude Opus 5.5
3417f00f1b
The Bible TV PC's Wi-Fi dropped for ~30 s at 06:17 on 2026-10-11; the hold could not reach it and showed FIX FAILED, though nothing on the PC was broken. An ssh fix that cannot reach its machine now shows e.g. "BIBLE APP NOT REACHABLE". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1472 lines
81 KiB
Python
1472 lines
81 KiB
Python
"""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()
|
|
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}
|
|
snap.launchd_time = now
|
|
snap.procs = {"/Applications/ProdDash.app/Contents/MacOS/ProdDash": True}
|
|
snap.http = {"https://central.resi.io/": True, "http://10.6.7.13:8080/": True, "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_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_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_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_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_racks_powering_off_is_not_a_network_event(self):
|
|
engine = sp.Engine(CONFIG)
|
|
run_steps(engine, self.snap, SUNDAY, 2)
|
|
for label in ("Axient_1", "Axient_2", "ULXD-4", "PSM_1", "PSM_2", "PSM_3"):
|
|
self.snap.conns[label]["category"] = "error"
|
|
tiles, _ = run_steps(engine, self.snap, SUNDAY + 3, 4)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nOK")
|
|
self.assertEqual(tiles["rf"]["x"].split("\n")[-1], "+5")
|
|
|
|
def test_gateway_and_link(self):
|
|
self.snap.ping["10.1.0.1"] = 2
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nNO GATEWAY")
|
|
self.snap.link = False
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nLINK DOWN\n+1")
|
|
|
|
def test_restart_targets(self):
|
|
engine = sp.Engine(CONFIG)
|
|
self.assertEqual(engine.restart_targets("hd3"), ["HyperDeck_3"])
|
|
self.assertEqual(engine.restart_targets("led"), ["VX1000", "UHDJr"])
|
|
self.assertEqual(engine.restart_targets("apps"), ["Propresenter-API", "ableton", "BibleTV"])
|
|
self.assertEqual(engine.restart_targets("dante"), []) # shared connection: never
|
|
self.assertEqual(engine.restart_targets("ingest"), [])
|
|
|
|
|
|
class TestFixes(unittest.TestCase):
|
|
"""Hold = fix only what that button is failing on. Nothing here runs a real fix."""
|
|
|
|
def setUp(self):
|
|
self.snap = healthy_snapshot(SUNDAY)
|
|
self.engine = sp.Engine(CONFIG)
|
|
# Safety net: a hold in these tests must never reach a real machine.
|
|
self.engine.executor = self.engine.poe_executor = lambda *a: True
|
|
|
|
def plan(self, tile):
|
|
run_steps(self.engine, self.snap, SUNDAY, 4)
|
|
return [(kind, what) for _, kind, what, _ in self.engine.fix_plan(tile, self.snap)]
|
|
|
|
def test_green_buttons_do_nothing(self):
|
|
for tile in ("apps", "dante", "rec", "hd1", "x32", "ingest", "show"):
|
|
self.assertEqual(self.plan(tile), [], tile)
|
|
|
|
def test_propresenter_is_restarted_once(self):
|
|
self.snap.conns["Propresenter-API"]["category"] = "error"
|
|
self.snap.remote["prop\0pgrep -xq ProPresenter"] = False
|
|
self.assertEqual(self.plan("apps"), [("fix", "propresenter")])
|
|
self.assertEqual(self.plan("show"), [("fix", "propresenter")]) # the group does the same
|
|
|
|
def test_apps_on_other_macs(self):
|
|
self.snap.remote["tracks\0pgrep -xq Live"] = False
|
|
self.snap.remote["translation\0pgrep -xq 'Church Translate'"] = False
|
|
self.assertEqual(self.plan("apps"), [("fix", "ableton"), ("fix", "translate")])
|
|
|
|
def test_bible_app_is_restarted_on_the_windows_pc(self):
|
|
self.snap.http["http://10.6.7.13:8080/"] = False
|
|
self.assertEqual(self.plan("apps"), [("fix", "bibleapp")])
|
|
fix = CONFIG["fixes"]["bibleapp"]
|
|
self.assertEqual(CONFIG["hosts"][fix["host"]]["ssh"], "media@10.6.7.13")
|
|
self.assertIn('schtasks /Run /TN "\\StatusPanel\\Start BibleApp"', fix["run"]) # cmd.exe, not sh
|
|
|
|
def test_lighting_and_livestream_pcs(self):
|
|
self.snap.conns["OBS-Youtube"]["category"] = "error"
|
|
self.snap.conns["arena"]["category"] = "error"
|
|
self.snap.artnet_since, self.snap.artnet_last = SUNDAY - 100, {"10.1.10.206": SUNDAY - 100}
|
|
self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"]
|
|
if p["name"] != "Lighting PC"]
|
|
self.snap.mdns = self.snap.mdns - {"Lighting PC"}
|
|
self.assertEqual(self.plan("obs"), [("fix", "obs")])
|
|
self.assertEqual(self.plan("arena"), [("fix", "arena")])
|
|
self.assertEqual(self.plan("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")).read())
|
|
self.assertIn("open -g /Applications/ProdCom.app", CONFIG["fixes"]["dvs_ingest"]["run"]) # ProdCom after DVS
|
|
|
|
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)
|
|
tiles = self.engine.step(self.snap, time.time())
|
|
self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nCHECKING\nDVS", "fixing"))
|
|
tiles = self.engine.step(self.snap, time.time() + 25) # still gone: back to the alarm
|
|
self.assertEqual(tiles["dante"]["x"], "DANTE\nPROP MAC GONE")
|
|
# held again within a minute
|
|
self.engine.run_fixes("dante", self.snap, None, t + 30)
|
|
self.assertEqual(self.engine.notes["dante"]["text"], "JUST TRIED\nWAIT 1 MIN")
|
|
# a fix whose command fails says so
|
|
outcome["ok"] = False
|
|
self.engine.run_fixes("dante", self.snap, None, t + 100)
|
|
time.sleep(0.2)
|
|
tiles = self.engine.step(self.snap, time.time())
|
|
self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nFIX FAILED\nDVS", "failed"))
|
|
|
|
def test_dante_hold_power_cycles_any_poe_device_from_the_map(self):
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("STUDIO-B-MON") # PoE box (per the map)
|
|
devices.remove("BOH-WING") # console: not PoE
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
self.snap.dante_map = {
|
|
"STUDIO-B-MON": {"mac": "00:1d:c1:51:18:84", "switch": "10.66.1.102", "port": "0/15", "shared": 1, "poe": True},
|
|
"BOH-WING": {"mac": "00:1d:c1:03:ec:3a", "switch": "10.66.1.102", "port": "0/3", "shared": 1, "poe": False},
|
|
}
|
|
self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller"), ("poe", "STUDIO-B-MON")])
|
|
self.snap.dante_map = {} # not mapped yet: just re-check via the module
|
|
self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller")])
|
|
|
|
def test_poe_hold_hands_the_mapped_entry_to_the_switch_code(self):
|
|
calls = []
|
|
self.engine.poe_executor = lambda cfg, name, entry: calls.append((name, entry["mac"])) or True
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("STUDIO-B-TB")
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
self.snap.dante_map = {"STUDIO-B-TB": {"mac": "00:1d:c1:89:1a:0a", "switch": "10.66.1.102",
|
|
"port": "0/16", "shared": 1, "poe": True}}
|
|
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
|
|
self.engine.run_fixes("dante", self.snap, None, t)
|
|
time.sleep(0.2)
|
|
self.assertEqual(calls, [("STUDIO-B-TB", "00:1d:c1:89:1a:0a")])
|
|
tiles = self.engine.step(self.snap, time.time())
|
|
self.assertEqual(tiles["dante"]["x"], "DANTE\nCHECKING\nPOWER CYCLE")
|
|
|
|
def test_dante_map_poller(self):
|
|
class Comp:
|
|
def variable(self, label, name):
|
|
return {"STUDIO-B-MON_ip": "169.254.133.24", "NA-Tracks_ip": "169.254.99.70"}.get(name)
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["STUDIO-B-MON", "NA-Tracks"])
|
|
self.snap.dante_map = {"BOH-WING": {"mac": "x", "poe": False}} # an older entry is kept
|
|
arp = lambda: {"169.254.133.24": "00:1d:c1:51:18:84", "169.254.99.70": "00:e0:4c:ff:b6:8a"}
|
|
survey = lambda switches, macs: {"00:1d:c1:51:18:84": {"switch": "10.66.1.102", "port": "0/15",
|
|
"shared": 1, "poe": True}}
|
|
path = os.path.join(tempfile.mkdtemp(), "dante-map.json")
|
|
sp.poll_dante_map(self.snap, CONFIG, Comp(), path, arp=arp, survey=survey)
|
|
self.assertEqual(self.snap.dante_map["STUDIO-B-MON"]["port"], "0/15")
|
|
self.assertTrue(self.snap.dante_map["STUDIO-B-MON"]["poe"])
|
|
self.assertNotIn("poe", self.snap.dante_map["NA-Tracks"]) # a Mac: not on a PoE port
|
|
self.assertIn("BOH-WING", self.snap.dante_map)
|
|
self.assertEqual(sp.load_dante_map(path), self.snap.dante_map)
|
|
self.assertEqual(sp.normalize_mac("0:1d:c1:89:1a:a"), "00:1d:c1:89:1a:0a")
|
|
|
|
def test_ingest_dvs_and_no_auto_fix(self):
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("STUDIO-A-Ingest")
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
self.assertEqual(self.plan("dante"), [("fix", "dvs_ingest")])
|
|
# A failure with no fix must not claim there is nothing wrong.
|
|
self.snap.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_every_fix_is_valid(self):
|
|
for f in CONFIG["fixes"].values():
|
|
self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck", "midi_connect", "reconnect"))
|
|
|
|
|
|
class FakeDeck:
|
|
"""A pretend HyperDeck on a local port: answers like a real one, stays silent, or greets only."""
|
|
|
|
def __init__(self, mode):
|
|
import socket as _s
|
|
self.mode, self.received = mode, b""
|
|
self.srv = _s.socket()
|
|
self.srv.bind(("127.0.0.1", 0))
|
|
self.srv.listen(5)
|
|
self.port = self.srv.getsockname()[1]
|
|
threading.Thread(target=self.serve, daemon=True).start()
|
|
|
|
def serve(self):
|
|
while True:
|
|
try:
|
|
conn, _ = self.srv.accept()
|
|
except OSError:
|
|
return
|
|
with conn:
|
|
conn.settimeout(4)
|
|
if self.mode != "silent":
|
|
conn.sendall(b"500 connection info:\r\nprotocol version: 1.11\r\nmodel: HyperDeck Studio 4K Pro\r\n\r\n")
|
|
try:
|
|
data = conn.recv(1024)
|
|
self.received += data
|
|
if self.mode in ("record", "preview") and b"transport info" in data:
|
|
conn.sendall(b"208 transport info:\r\nstatus: %s\r\nspeed: 100\r\n\r\n" % self.mode.encode())
|
|
more = conn.recv(1024)
|
|
self.received += more
|
|
except OSError:
|
|
pass
|
|
|
|
def close(self):
|
|
self.srv.close()
|
|
|
|
|
|
class TestHyperDeckProbe(unittest.TestCase):
|
|
def test_probe_answers(self):
|
|
for mode, expected in (("record", "record"), ("preview", "preview"), ("greet", "answered"), ("silent", "hung")):
|
|
deck = FakeDeck(mode)
|
|
try:
|
|
self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=deck.port), expected, mode)
|
|
finally:
|
|
deck.close()
|
|
deck = FakeDeck("preview")
|
|
port = deck.port
|
|
deck.close()
|
|
self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=port), "unreachable")
|
|
|
|
def test_the_hold_switches_the_connection_off_and_on_and_never_reboots(self):
|
|
from unittest import mock
|
|
deck = FakeDeck("silent") # even a deck that would once have been rebooted
|
|
try:
|
|
comp = mock.Mock()
|
|
comp.connections.side_effect = [
|
|
[{"label": "HyperDeck_1", "id": "h1", "enabled": True, "status": {"category": "error"}}],
|
|
[{"label": "HyperDeck_1", "id": "h1", "enabled": True, "status": {"category": "good"}}]]
|
|
comp.disable_connection.return_value = comp.enable_connection.return_value = True
|
|
chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port}
|
|
with mock.patch.object(sp, "Companion", return_value=comp), mock.patch.object(sp.time, "sleep"):
|
|
self.assertTrue(sp.execute_fix(CONFIG, "hyperdeck", chk))
|
|
self.assertEqual([c[0] for c in comp.method_calls if c[0].endswith("_connection")],
|
|
["disable_connection", "enable_connection"])
|
|
comp.restart_connection.assert_not_called()
|
|
time.sleep(0.3)
|
|
self.assertNotIn(b"reboot", deck.received)
|
|
finally:
|
|
deck.close()
|
|
|
|
def test_a_deck_that_stays_offline_says_so(self):
|
|
from unittest import mock
|
|
fix = CONFIG["fixes"]["hyperdeck"]
|
|
for pings, expect in ((True, "STILL OFFLINE\nHOLD AGAIN"), (False, "NO NETWORK\nCHECK POWER")):
|
|
comp = mock.Mock()
|
|
comp.connections.side_effect = [
|
|
[{"label": "HyperDeck_2", "id": "h2", "enabled": True, "status": {"category": "error"}}],
|
|
[{"label": "HyperDeck_2", "id": "h2", "enabled": True, "status": {"category": "error"}}]]
|
|
with mock.patch.object(sp, "Companion", return_value=comp), \
|
|
mock.patch.object(sp, "ping_once", return_value=pings) as ping:
|
|
self.assertEqual(sp.reconnect_and_verify(CONFIG, fix, "x", sleep=lambda s: None,
|
|
conn="HyperDeck_2", host="10.1.10.12", toggle=True), expect)
|
|
ping.assert_called_once_with("10.1.10.12") # the deck's own address, from Companion
|
|
|
|
|
|
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, "{}")
|
|
self._send(404, "")
|
|
|
|
|
|
class TestAgainstFakeCompanion(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
snap = healthy_snapshot(time.time())
|
|
FakeCompanion.store = {
|
|
"conns": snap.conns,
|
|
"vars": dict(snap.vars),
|
|
"custom": {"sp_state": "", "sp_hb": "", "sp_cmd": "", "flash_tick": "0"},
|
|
"restarts": [],
|
|
}
|
|
cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion)
|
|
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
|
|
cls.port = cls.server.server_address[1]
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
|
|
def test_client_round_trip_and_long_state(self):
|
|
comp = sp.Companion("127.0.0.1", self.port)
|
|
self.assertEqual(len(comp.connections()), len(FakeCompanion.store["conns"]))
|
|
self.assertEqual(comp.variable("HyperDeck_1", "recordingTime"), "28:00:00")
|
|
self.assertIsNone(comp.variable("HyperDeck_1", "nope"))
|
|
snap = healthy_snapshot(time.time())
|
|
tiles = sp.Engine(CONFIG).step(snap, time.time())
|
|
js = json.dumps(tiles, separators=(",", ":"), sort_keys=True)
|
|
self.assertEqual(comp.set_custom("sp_state", js), (True, 200))
|
|
self.assertEqual(json.loads(comp.get_custom("sp_state"))["all"]["x"], "ALL OK")
|
|
self.assertEqual(comp.set_custom("missing", "x"), (False, 404))
|
|
print("\n sp_state is %d bytes, %d in the URL" % (len(js), len(urllib.parse.quote(js, safe=""))))
|
|
|
|
def test_commands(self):
|
|
comp = sp.Companion("127.0.0.1", self.port)
|
|
store = FakeCompanion.store
|
|
snap = healthy_snapshot(SUNDAY)
|
|
engine = sp.Engine(CONFIG)
|
|
engine.executor = lambda cfg, fix_id, chk: self.fail("no real fix may run in a test")
|
|
now = SUNDAY
|
|
store["conns"]["VX1000"]["category"] = "error"
|
|
try:
|
|
sp.poll_connections(comp, snap)
|
|
run_steps(engine, snap, now, 4)
|
|
before = len(store["restarts"])
|
|
# 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["companion"]["port"] = self.port
|
|
# keep it offline: no pings, no web, no launchd, no bonjour lookups
|
|
cfg["network"]["gateway"] = None
|
|
keep = ("conn", "hyperdeck", "dante", "tascam", "shure", "psm")
|
|
for t in cfg["tiles"]:
|
|
if "checks" in t:
|
|
t["checks"] = [c for c in t["checks"] if c["type"] in keep] or [{"type": "conn", "conn": "x32"}]
|
|
store = FakeCompanion.store
|
|
store["conns"]["ATEM"]["category"] = "error"
|
|
store["custom"]["sp_cmd"] = ";ack atem"
|
|
data_dir = tempfile.mkdtemp()
|
|
originals = (sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map,
|
|
sp.poll_remote_midi)
|
|
sp.poll_dante_map = lambda *a, **k: None # never log in to a switch from a test
|
|
sp.poll_mdns = lambda snap, seconds=3: None
|
|
sp.poll_system = lambda snap, cfg: None
|
|
sp.poll_midi = lambda snap, cfg: None # CoreMIDI has its own test
|
|
sp.poll_remote = lambda snap, cfg: None # never SSH anywhere from a test
|
|
sp.poll_remote_midi = lambda snap, cfg: None
|
|
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.poll_remote_midi) = originals
|
|
store["conns"]["ATEM"]["category"] = "good"
|
|
self.assertFalse(loop.is_alive()) # it shuts down cleanly
|
|
self.assertTrue(store["custom"]["sp_hb"].isdigit())
|
|
self.assertLess(abs(int(store["custom"]["sp_hb"]) - time.time()), 5)
|
|
self.assertEqual(store["custom"]["sp_cmd"], "") # the command was consumed
|
|
self.assertIn("atem", state)
|
|
self.assertIn("ATEM", state["atem"]["x"])
|
|
|
|
|
|
class TestPage(unittest.TestCase):
|
|
def setUp(self):
|
|
self.export = make_page.build(CONFIG)
|
|
self.controls = self.export["page"]["controls"]
|
|
|
|
def test_layout(self):
|
|
self.assertEqual(self.export["type"], "page")
|
|
self.assertEqual(self.export["oldPageNumber"], 19)
|
|
self.assertEqual(self.controls["0"]["0"], {"type": "pageup"}) # 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["3"]["0"]["steps"]["0"]["action_sets"]["up"][0]
|
|
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])
|
|
busy = {"page": {"name": "PAGE", "controls": dict(nav, **{"2": {"3": {"type": "button"}}})}}
|
|
self.assertFalse(push_page.safe_target(busy, "23\\nRestart")[0]) # pages 22 and 97
|
|
busy["page"]["name"] = "23\\nRestart"
|
|
self.assertTrue(push_page.safe_target(busy, "23\\nRestart")[0]) # our own page again
|
|
|
|
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()
|