christianandClaude Opus 5.5
f7346d6f5e
The Art-Net heartbeat moves from APPS to a VISTA tile ("VISTA / SILENT";
"VISTA SILENT" on SHOW and ALL), hold starts Vista if it is not running.
SHOW now groups MIDI, APPS and VISTA. Tested live before the move: closing
Vista and Resolume alarmed, and each hold brought them back (Vista 5 s,
Arena 9 s after the hold).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1161 lines
61 KiB
Python
1161 lines
61 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}
|
|
snap.launchd_time = now
|
|
snap.procs = {"/Applications/ProdDash.app/Contents/MacOS/ProdDash": True}
|
|
snap.http = {"https://central.resi.io/": True, "http://10.6.7.13:8080/": True}
|
|
snap.disk = {"/": 97}
|
|
return snap
|
|
|
|
|
|
def run_steps(engine, snap, start, seconds, step=1.0):
|
|
tiles, t = None, start
|
|
while t <= start + seconds:
|
|
snap.conns_time = t
|
|
tiles = engine.step(snap, t)
|
|
t += step
|
|
return tiles, t
|
|
|
|
|
|
def level(tile):
|
|
for name, ((b, f), _) in sp.COLORS.items():
|
|
if (tile["b"], tile["f"]) == (b, f):
|
|
return name
|
|
return "?"
|
|
|
|
|
|
class TestChecks(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = SUNDAY
|
|
self.snap = healthy_snapshot(self.now)
|
|
self.ctx = {"now": self.now, "service": True, "any_recording": False}
|
|
|
|
def test_all_green_when_healthy(self):
|
|
tiles = sp.Engine(CONFIG).step(self.snap, self.now)
|
|
bad = {k: v["x"] for k, v in tiles.items() if level(v) not in ("ok", "service", "off")}
|
|
self.assertEqual(bad, {})
|
|
self.assertEqual(tiles["all"]["x"], "ALL OK")
|
|
self.assertEqual(tiles["hd1"]["x"], "HD1\n28h")
|
|
self.assertEqual(tiles["dante"]["x"], "DANTE\n19/19")
|
|
|
|
def test_connection_states(self):
|
|
c = self.snap.conns["ATEM"]
|
|
c["category"], c["message"] = "error", "Network error: connect ETIMEDOUT 10.1.10.2:9910"
|
|
self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "NO ANSWER")
|
|
c["message"] = "connect EHOSTUNREACH"
|
|
self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "UNREACHABLE")
|
|
c["enabled"] = False
|
|
self.assertEqual((sp.conn_result(self.snap, "ATEM", self.now).level,
|
|
sp.conn_result(self.snap, "ATEM", self.now).msg), (sp.WARN, "DISABLED"))
|
|
self.assertEqual(sp.conn_result(self.snap, "nope", self.now).msg, "NOT IN COMPANION")
|
|
self.assertEqual(sp.conn_result(self.snap, "x32", self.now + 60).msg, "NO DATA")
|
|
|
|
def test_hyperdeck_disk_and_record_sync(self):
|
|
s = self.snap
|
|
s.vars[("HyperDeck_2", "slotId")] = "-"
|
|
s.vars[("HyperDeck_2", "recordingTime")] = "--:--:--"
|
|
out = dict(sp.check_hyperdeck({"conn": "HyperDeck_2"}, s, self.ctx))
|
|
self.assertEqual(out["disk"].msg, "NO DISK")
|
|
self.assertTrue(out["disk"].service_only)
|
|
s.vars[("HyperDeck_3", "recordingTime")] = "0:20:00"
|
|
self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].msg, "LOW 20m")
|
|
s.vars[("HyperDeck_3", "recordingTime")] = "0:04:00"
|
|
self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].level, sp.ERR)
|
|
ctx = dict(self.ctx, any_recording=True)
|
|
self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))["rec"].msg, "NOT REC")
|
|
s.vars[("HyperDeck_1", "status")] = "Record"
|
|
out = dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))
|
|
self.assertEqual(out["rec"].level, sp.OK)
|
|
self.assertEqual(out["conn"].info, "REC 28h")
|
|
|
|
def test_optional_deck(self):
|
|
for i in (2, 4):
|
|
self.snap.vars[("HyperDeck_%d" % i, "slotId")] = "-"
|
|
self.snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "--:--:--"
|
|
for i in (1, 3, 5, 6):
|
|
self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual((level(tiles["hd2"]), tiles["hd2"]["x"]), ("off", "HD2\nNO DISK"))
|
|
self.assertEqual(level(tiles["rec"]), "ok")
|
|
self.assertEqual(tiles["all"]["x"], "ALL OK")
|
|
self.snap.vars[("HyperDeck_2", "slotId")] = "1" # disk goes in: now it is watched
|
|
self.snap.vars[("HyperDeck_2", "recordingTime")] = "20:00:00"
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual(tiles["hd2"]["x"], "HD2\nNOT REC")
|
|
self.assertEqual(level(tiles["hd2"]), "err")
|
|
|
|
def test_one_deck_not_recording_blinks(self):
|
|
engine = sp.Engine(CONFIG)
|
|
for i in (1, 2, 3, 5, 6):
|
|
self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
|
|
tiles, _ = run_steps(engine, self.snap, self.now, 4)
|
|
self.assertEqual(tiles["hd4"]["x"], "HD4\nNOT REC")
|
|
self.assertEqual(level(tiles["hd4"]), "err")
|
|
self.assertEqual(tiles["all"]["x"], "1 DOWN\nHD4 NOT REC")
|
|
self.assertEqual(tiles["rec"]["x"], "REC\nHD4 NOT REC")
|
|
|
|
def test_dante(self):
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["NA-Tracks"])
|
|
out = dict(sp.check_dante({"conn": "audinate-dantecontroller",
|
|
"devices": {"NA-Tracks": "DANTE", "BOH-WING": "WING"}}, self.snap, self.ctx))
|
|
self.assertEqual(out["BOH-WING"].msg, "WING GONE")
|
|
self.assertEqual(out["NA-Tracks"].level, sp.OK)
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = None
|
|
out = dict(sp.check_dante({"conn": "audinate-dantecontroller", "devices": ["BOH-WING"]}, self.snap, self.ctx))
|
|
self.assertEqual(out["conn"].msg, "NO DATA")
|
|
|
|
def test_dante_button_names_the_missing_device(self):
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("NA-Tracks")
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual(tiles["dante"]["x"], "DANTE\nTRACKS GONE")
|
|
self.assertEqual(level(tiles["dante"]), "err")
|
|
self.assertEqual(tiles["all"]["x"], "1 DOWN\nDANTE TRACKS GONE")
|
|
|
|
def test_shure_psm_tascam(self):
|
|
s = self.snap
|
|
s.vars[("Axient_1", "ch_1_battery_bars")] = "1"
|
|
s.vars[("Axient_1", "ch_2_battery_bars")] = "2"
|
|
s.vars[("Axient_1", "ch_3_battery_bars")] = "Unknown"
|
|
s.vars[("Axient_1", "ch_4_interference_status")] = "DETECTED"
|
|
s.vars[("Axient_1", "ch_3_interference_status")] = "DETECTED" # transmitter off: ignored
|
|
out = dict(sp.check_shure({"conn": "Axient_1", "channels": 4}, s, self.ctx))
|
|
self.assertEqual((out["ch1_batt"].level, out["ch1_batt"].msg), (sp.ERR, "VOX 1 BATT"))
|
|
self.assertEqual(out["ch2_batt"].level, sp.WARN)
|
|
self.assertEqual(out["ch3_batt"].level, sp.OK)
|
|
self.assertEqual(out["ch3_rf"].level, sp.OK)
|
|
self.assertEqual(out["ch4_rf"].msg, "VOX 4 INTERF")
|
|
s.vars[("PSM_2", "ch_1_rf_mute")] = "MUTED"
|
|
self.assertEqual(dict(sp.check_psm({"conn": "PSM_2", "channels": 2}, s, self.ctx))["ch1_mute"].msg,
|
|
"PACK RF MUTE")
|
|
s.vars[("tascam", "error")] = "Media Error"
|
|
self.assertEqual(dict(sp.check_tascam({"conn": "tascam"}, s, self.ctx))["error"].msg, "MEDIA ERROR")
|
|
|
|
def test_rf_group_names_the_mic(self):
|
|
self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "1"
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual(tiles["rf"]["x"], "RF\nVOX 3 BATT")
|
|
self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "5"
|
|
self.snap.conns["PSM_1"]["category"] = "error"
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertIn("PSM 1 OFFLINE", tiles["rf"]["x"])
|
|
|
|
def test_bonjour_parsing_morning_and_afternoon(self):
|
|
out = ("Browsing for _apple-midi._udp.local.\n"
|
|
" 9:20:51.321 Add 3 13 local. _apple-midi._udp. NA Tracks\n"
|
|
"19:20:51.321 Add 3 27 local. _apple-midi._udp. Lighting PC\n"
|
|
"19:21:02.001 Add 3 27 local. _apple-midi._udp. Gone Soon\n"
|
|
"19:21:09.001 Rmv 1 27 local. _apple-midi._udp. Gone Soon\n")
|
|
self.assertEqual(sp.parse_dns_sd(out), {"NA Tracks", "Lighting PC"})
|
|
|
|
def test_vista_heartbeat(self):
|
|
self.snap.artnet_since = self.now - 60
|
|
self.snap.artnet_last = {"10.1.10.206": self.now - 1}
|
|
chk = {"source": "10.1.10.206", "max_silence_seconds": 5, "msg": "VISTA SILENT"}
|
|
self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].level, sp.OK)
|
|
self.snap.artnet_last = {"10.1.10.206": self.now - 8}
|
|
self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].msg, "VISTA SILENT")
|
|
self.snap.artnet_since = self.now - 2 # just started listening: no verdict
|
|
self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].level, sp.OK)
|
|
self.snap.artnet_since = None # could not listen: no verdict
|
|
self.assertEqual(sp.check_artnet(chk, self.snap, self.ctx)[0][1].level, sp.OK)
|
|
# 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_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"]), "warn")
|
|
tiles, _ = run_steps(engine, self.snap, t + 200, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nOK")
|
|
|
|
def test_racks_powering_off_is_not_a_network_event(self):
|
|
engine = sp.Engine(CONFIG)
|
|
run_steps(engine, self.snap, SUNDAY, 2)
|
|
for label in ("Axient_1", "Axient_2", "ULXD-4", "PSM_1", "PSM_2", "PSM_3"):
|
|
self.snap.conns[label]["category"] = "error"
|
|
tiles, _ = run_steps(engine, self.snap, SUNDAY + 3, 4)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nOK")
|
|
self.assertEqual(tiles["rf"]["x"].split("\n")[-1], "+5")
|
|
|
|
def test_gateway_and_link(self):
|
|
self.snap.ping["10.1.0.1"] = 2
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nNO GATEWAY")
|
|
self.snap.link = False
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nLINK DOWN\n+1")
|
|
|
|
def test_restart_targets(self):
|
|
engine = sp.Engine(CONFIG)
|
|
self.assertEqual(engine.restart_targets("hd3"), ["HyperDeck_3"])
|
|
self.assertEqual(engine.restart_targets("led"), ["VX1000", "UHDJr"])
|
|
self.assertEqual(engine.restart_targets("apps"), ["Propresenter-API", "ableton", "BibleTV"])
|
|
self.assertEqual(engine.restart_targets("dante"), []) # shared connection: never
|
|
self.assertEqual(engine.restart_targets("ingest"), [])
|
|
|
|
|
|
class TestFixes(unittest.TestCase):
|
|
"""Hold = fix only what that button is failing on. Nothing here runs a real fix."""
|
|
|
|
def setUp(self):
|
|
self.snap = healthy_snapshot(SUNDAY)
|
|
self.engine = sp.Engine(CONFIG)
|
|
# Safety net: a hold in these tests must never reach a real machine.
|
|
self.engine.executor = self.engine.poe_executor = lambda *a: True
|
|
|
|
def plan(self, tile):
|
|
run_steps(self.engine, self.snap, SUNDAY, 4)
|
|
return [(kind, what) for _, kind, what, _ in self.engine.fix_plan(tile, self.snap)]
|
|
|
|
def test_green_buttons_do_nothing(self):
|
|
for tile in ("apps", "dante", "rec", "hd1", "x32", "ingest", "show"):
|
|
self.assertEqual(self.plan(tile), [], tile)
|
|
|
|
def test_propresenter_is_restarted_once(self):
|
|
self.snap.conns["Propresenter-API"]["category"] = "error"
|
|
self.snap.remote["prop\0pgrep -xq ProPresenter"] = False
|
|
self.assertEqual(self.plan("apps"), [("fix", "propresenter")])
|
|
self.assertEqual(self.plan("show"), [("fix", "propresenter")]) # the group does the same
|
|
|
|
def test_apps_on_other_macs(self):
|
|
self.snap.remote["tracks\0pgrep -xq Live"] = False
|
|
self.snap.remote["translation\0pgrep -xq 'Church Translate'"] = False
|
|
self.assertEqual(self.plan("apps"), [("fix", "ableton"), ("fix", "translate")])
|
|
|
|
def test_bible_app_is_restarted_on_the_windows_pc(self):
|
|
self.snap.http["http://10.6.7.13:8080/"] = False
|
|
self.assertEqual(self.plan("apps"), [("fix", "bibleapp")])
|
|
fix = CONFIG["fixes"]["bibleapp"]
|
|
self.assertEqual(CONFIG["hosts"][fix["host"]]["ssh"], "media@10.6.7.13")
|
|
self.assertIn('schtasks /Run /TN "\\StatusPanel\\Start BibleApp"', fix["run"]) # cmd.exe, not sh
|
|
|
|
def test_lighting_and_livestream_pcs(self):
|
|
self.snap.conns["OBS-Youtube"]["category"] = "error"
|
|
self.snap.conns["arena"]["category"] = "error"
|
|
self.snap.artnet_since, self.snap.artnet_last = SUNDAY - 100, {"10.1.10.206": SUNDAY - 100}
|
|
self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"]
|
|
if p["name"] != "Lighting PC"]
|
|
self.snap.mdns = self.snap.mdns - {"Lighting PC"}
|
|
self.assertEqual(self.plan("obs"), [("fix", "obs")])
|
|
self.assertEqual(self.plan("arena"), [("fix", "arena")])
|
|
self.assertEqual(self.plan("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_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_an_off_pc_is_woken(self):
|
|
from unittest import mock
|
|
woke = []
|
|
done = mock.Mock(returncode=255, stdout="", stderr="ssh: connect to host 10.1.9.126 port 22: timed out")
|
|
with mock.patch.object(sp.subprocess, "run", return_value=done), \
|
|
mock.patch.object(sp, "wake_on_lan", side_effect=lambda mac, bc=None: woke.append((mac, bc))):
|
|
self.assertEqual(sp.execute_fix(CONFIG, "obs", {"conn": "OBS-Youtube"}), sp.WAKING)
|
|
self.assertEqual(woke, [("6c:24:08:d7:fb:57", "10.1.255.255")])
|
|
self.assertFalse(sp.execute_fix(CONFIG, "arena", {"conn": "arena"})) # no MAC: just fails
|
|
self.assertEqual(len(woke), 1)
|
|
# the button says so
|
|
self.snap.conns["OBS-Youtube"]["category"] = "error"
|
|
self.engine.executor = lambda cfg, fix_id, chk: sp.WAKING
|
|
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
|
|
self.engine.run_fixes("obs", self.snap, None, t)
|
|
time.sleep(0.2)
|
|
self.assertEqual(self.engine.step(self.snap, time.time())["obs"]["x"], "OBS\nWAKING PC\nOBS")
|
|
|
|
def test_magic_packet(self):
|
|
from unittest import mock
|
|
with mock.patch.object(sp.socket, "socket") as sock:
|
|
packet = sp.wake_on_lan("6c:24:08:d7:fb:57", "10.1.255.255")
|
|
self.assertEqual(packet, b"\xff" * 6 + bytes.fromhex("6c2408d7fb57") * 16)
|
|
sent = [c.args[1] for c in sock.return_value.__enter__.return_value.sendto.call_args_list]
|
|
self.assertEqual(sent, [("10.1.255.255", 9), ("255.255.255.255", 9)])
|
|
|
|
def test_companion_reconnects_after_the_app_starts(self):
|
|
from unittest import mock
|
|
comp = mock.Mock()
|
|
comp.connections.return_value = [{"label": "arena", "id": "abc"}]
|
|
ok = mock.Mock(returncode=0, stdout="", stderr="")
|
|
with mock.patch.object(sp.subprocess, "run", return_value=ok), \
|
|
mock.patch.object(sp, "Companion", return_value=comp):
|
|
self.assertTrue(sp.execute_fix(CONFIG, "arena", {"conn": "arena"}))
|
|
comp.restart_connection.assert_called_once_with("abc")
|
|
comp.reset_mock()
|
|
self.assertTrue(sp.execute_fix(CONFIG, "vista", {"type": "artnet"})) # no "reconnect"
|
|
comp.restart_connection.assert_not_called()
|
|
|
|
def test_unreachable_mac_is_not_a_verdict(self):
|
|
self.snap.remote["tracks\0pgrep -xq Live"] = None
|
|
self.assertEqual(self.plan("apps"), [])
|
|
|
|
def test_dante_reopens_dvs_on_the_right_mac(self):
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("NA-ProPresenter-Mac")
|
|
devices.remove("BOH-WING") # hardware: re-check via Companion's Dante module
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller"), ("fix", "dvs_prop")])
|
|
|
|
def test_rec_reboots_only_hung_decks_not_recording_ones(self):
|
|
for i in (2, 3, 5):
|
|
self.snap.conns["HyperDeck_%d" % i]["category"] = "error"
|
|
self.snap.vars[("HyperDeck_3", "status")] = "Record" # last known state: recording
|
|
self.assertEqual(self.plan("rec"), [("fix", "hyperdeck"), ("fix", "hyperdeck")])
|
|
decks = [(chk["conn"], chk["ip"]) for _, _, _, chk in self.engine.fix_plan("rec", self.snap)]
|
|
self.assertEqual(decks, [("HyperDeck_2", "10.1.10.12"), ("HyperDeck_5", "10.1.10.15")])
|
|
# the address comes from Companion and is remembered if the connection loses it
|
|
self.snap.vars[("HyperDeck_2", "ip")] = None
|
|
decks = [(chk["conn"], chk["ip"]) for _, _, _, chk in self.engine.fix_plan("rec", self.snap)]
|
|
self.assertEqual(decks[0], ("HyperDeck_2", "10.1.10.12"))
|
|
self.assertEqual(self.plan("hd3"), [])
|
|
|
|
def test_disabled_connection_is_switched_on_never_rebooted(self):
|
|
# 2026-10-08: HyperDeck_1 disabled in Companion, a hold sent the deck a reboot.
|
|
self.snap.conns["HyperDeck_1"]["enabled"] = False
|
|
self.snap.conns["x32"]["enabled"] = False
|
|
self.assertEqual(self.plan("hd1"), [("enable", "HyperDeck_1")])
|
|
self.assertEqual(self.plan("rec"), [("enable", "HyperDeck_1")])
|
|
self.assertEqual(self.plan("x32"), [("enable", "x32")])
|
|
|
|
def test_not_recording_is_not_fixed_by_a_reboot(self):
|
|
for i in (1, 3):
|
|
self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
|
|
self.assertEqual(self.plan("hd5"), []) # NOT REC: a person decides
|
|
|
|
def test_connection_without_a_fix_is_restarted(self):
|
|
self.snap.conns["x32"]["category"] = "error"
|
|
self.assertEqual(self.plan("x32"), [("restart", "x32")])
|
|
|
|
def test_ingest_services(self):
|
|
del self.snap.launchd["org.waterschurch.resi-propresenter"]
|
|
self.snap.procs["/Applications/ProdDash.app/Contents/MacOS/ProdDash"] = False
|
|
self.assertEqual(self.plan("ingest"), [("fix", "bridge"), ("fix", "proddash")])
|
|
|
|
def test_cooldown_and_executor(self):
|
|
ran = []
|
|
self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) # record, never run
|
|
self.snap.remote["tracks\0pgrep -xq Live"] = False
|
|
run_steps(self.engine, self.snap, SUNDAY, 4)
|
|
self.engine.run_fixes("apps", self.snap, None, SUNDAY + 10)
|
|
self.engine.run_fixes("apps", self.snap, None, SUNDAY + 20) # held again 10 s later: skipped
|
|
self.engine.run_fixes("apps", self.snap, None, SUNDAY + 80)
|
|
time.sleep(0.2)
|
|
self.assertEqual(ran, ["ableton", "ableton"])
|
|
|
|
def test_hold_feedback_on_the_button(self):
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("NA-ProPresenter-Mac")
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
gate = threading.Event()
|
|
outcome = {"ok": True}
|
|
|
|
def slow_fix(cfg, fix_id, chk):
|
|
gate.wait(2)
|
|
return outcome["ok"]
|
|
self.engine.executor = slow_fix
|
|
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
|
|
self.engine.run_fixes("dante", self.snap, None, t)
|
|
tiles = self.engine.step(self.snap, t)
|
|
self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nFIXING\nDVS", "fixing"))
|
|
gate.set()
|
|
time.sleep(0.2)
|
|
tiles = self.engine.step(self.snap, time.time())
|
|
self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nCHECKING\nDVS", "fixing"))
|
|
tiles = self.engine.step(self.snap, time.time() + 25) # still gone: back to the alarm
|
|
self.assertEqual(tiles["dante"]["x"], "DANTE\nPROP MAC GONE")
|
|
# held again within a minute
|
|
self.engine.run_fixes("dante", self.snap, None, t + 30)
|
|
self.assertEqual(self.engine.notes["dante"]["text"], "JUST TRIED\nWAIT 1 MIN")
|
|
# a fix whose command fails says so
|
|
outcome["ok"] = False
|
|
self.engine.run_fixes("dante", self.snap, None, t + 100)
|
|
time.sleep(0.2)
|
|
tiles = self.engine.step(self.snap, time.time())
|
|
self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nFIX FAILED\nDVS", "failed"))
|
|
|
|
def test_dante_hold_power_cycles_any_poe_device_from_the_map(self):
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("STUDIO-B-MON") # PoE box (per the map)
|
|
devices.remove("BOH-WING") # console: not PoE
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
self.snap.dante_map = {
|
|
"STUDIO-B-MON": {"mac": "00:1d:c1:51:18:84", "switch": "10.66.1.102", "port": "0/15", "shared": 1, "poe": True},
|
|
"BOH-WING": {"mac": "00:1d:c1:03:ec:3a", "switch": "10.66.1.102", "port": "0/3", "shared": 1, "poe": False},
|
|
}
|
|
self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller"), ("poe", "STUDIO-B-MON")])
|
|
self.snap.dante_map = {} # not mapped yet: just re-check via the module
|
|
self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller")])
|
|
|
|
def test_poe_hold_hands_the_mapped_entry_to_the_switch_code(self):
|
|
calls = []
|
|
self.engine.poe_executor = lambda cfg, name, entry: calls.append((name, entry["mac"])) or True
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("STUDIO-B-TB")
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
self.snap.dante_map = {"STUDIO-B-TB": {"mac": "00:1d:c1:89:1a:0a", "switch": "10.66.1.102",
|
|
"port": "0/16", "shared": 1, "poe": True}}
|
|
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
|
|
self.engine.run_fixes("dante", self.snap, None, t)
|
|
time.sleep(0.2)
|
|
self.assertEqual(calls, [("STUDIO-B-TB", "00:1d:c1:89:1a:0a")])
|
|
tiles = self.engine.step(self.snap, time.time())
|
|
self.assertEqual(tiles["dante"]["x"], "DANTE\nCHECKING\nPOWER CYCLE")
|
|
|
|
def test_dante_map_poller(self):
|
|
class Comp:
|
|
def variable(self, label, name):
|
|
return {"STUDIO-B-MON_ip": "169.254.133.24", "NA-Tracks_ip": "169.254.99.70"}.get(name)
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["STUDIO-B-MON", "NA-Tracks"])
|
|
self.snap.dante_map = {"BOH-WING": {"mac": "x", "poe": False}} # an older entry is kept
|
|
arp = lambda: {"169.254.133.24": "00:1d:c1:51:18:84", "169.254.99.70": "00:e0:4c:ff:b6:8a"}
|
|
survey = lambda switches, macs: {"00:1d:c1:51:18:84": {"switch": "10.66.1.102", "port": "0/15",
|
|
"shared": 1, "poe": True}}
|
|
path = os.path.join(tempfile.mkdtemp(), "dante-map.json")
|
|
sp.poll_dante_map(self.snap, CONFIG, Comp(), path, arp=arp, survey=survey)
|
|
self.assertEqual(self.snap.dante_map["STUDIO-B-MON"]["port"], "0/15")
|
|
self.assertTrue(self.snap.dante_map["STUDIO-B-MON"]["poe"])
|
|
self.assertNotIn("poe", self.snap.dante_map["NA-Tracks"]) # a Mac: not on a PoE port
|
|
self.assertIn("BOH-WING", self.snap.dante_map)
|
|
self.assertEqual(sp.load_dante_map(path), self.snap.dante_map)
|
|
self.assertEqual(sp.normalize_mac("0:1d:c1:89:1a:a"), "00:1d:c1:89:1a:0a")
|
|
|
|
def test_ingest_dvs_and_no_auto_fix(self):
|
|
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
|
|
devices.remove("STUDIO-A-Ingest")
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
|
self.assertEqual(self.plan("dante"), [("fix", "dvs_ingest")])
|
|
# A failure with no fix must not claim there is nothing wrong.
|
|
self.snap.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"))
|
|
|
|
|
|
class FakeDeck:
|
|
"""A pretend HyperDeck on a local port: answers like a real one, stays silent, or greets only."""
|
|
|
|
def __init__(self, mode):
|
|
import socket as _s
|
|
self.mode, self.received = mode, b""
|
|
self.srv = _s.socket()
|
|
self.srv.bind(("127.0.0.1", 0))
|
|
self.srv.listen(5)
|
|
self.port = self.srv.getsockname()[1]
|
|
threading.Thread(target=self.serve, daemon=True).start()
|
|
|
|
def serve(self):
|
|
while True:
|
|
try:
|
|
conn, _ = self.srv.accept()
|
|
except OSError:
|
|
return
|
|
with conn:
|
|
conn.settimeout(4)
|
|
if self.mode != "silent":
|
|
conn.sendall(b"500 connection info:\r\nprotocol version: 1.11\r\nmodel: HyperDeck Studio 4K Pro\r\n\r\n")
|
|
try:
|
|
data = conn.recv(1024)
|
|
self.received += data
|
|
if self.mode in ("record", "preview") and b"transport info" in data:
|
|
conn.sendall(b"208 transport info:\r\nstatus: %s\r\nspeed: 100\r\n\r\n" % self.mode.encode())
|
|
more = conn.recv(1024)
|
|
self.received += more
|
|
except OSError:
|
|
pass
|
|
|
|
def close(self):
|
|
self.srv.close()
|
|
|
|
|
|
class TestHyperDeckProbe(unittest.TestCase):
|
|
def test_probe_answers(self):
|
|
for mode, expected in (("record", "record"), ("preview", "preview"), ("greet", "answered"), ("silent", "hung")):
|
|
deck = FakeDeck(mode)
|
|
try:
|
|
self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=deck.port), expected, mode)
|
|
finally:
|
|
deck.close()
|
|
deck = FakeDeck("preview")
|
|
port = deck.port
|
|
deck.close()
|
|
self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=port), "unreachable")
|
|
|
|
def test_only_a_silent_deck_is_rebooted(self):
|
|
deck = FakeDeck("silent")
|
|
try:
|
|
chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port}
|
|
self.assertTrue(sp.execute_fix(CONFIG, "hyperdeck", chk))
|
|
time.sleep(0.3)
|
|
self.assertIn(b"reboot", deck.received)
|
|
finally:
|
|
deck.close()
|
|
|
|
def test_a_recording_deck_is_never_rebooted(self):
|
|
deck = FakeDeck("record")
|
|
try:
|
|
cfg = json.loads(json.dumps(CONFIG))
|
|
cfg["companion"]["port"] = 1 # no Companion here: the reconnect just fails
|
|
chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port}
|
|
sp.execute_fix(cfg, "hyperdeck", chk)
|
|
time.sleep(0.3)
|
|
self.assertNotIn(b"reboot", deck.received)
|
|
finally:
|
|
deck.close()
|
|
|
|
|
|
class TestNetgearParsing(unittest.TestCase):
|
|
BIG = """ VLAN ID MAC Address Interface IfIndex Status
|
|
66 00:1D:C1:51:34:C8 0/14 14 Learned
|
|
66 00:1D:C1:51:18:84 0/15 15 Learned
|
|
66 00:E0:4C:9A:1A:F3 0/26 26 Learned
|
|
66 00:0E:DD:E2:8E:C2 0/26 26 Learned
|
|
66 00:0E:DD:E2:8F:60 0/26 26 Learned"""
|
|
SMALL = """ 66 00:1D:C1:51:34:C8 0/10 10 Learned
|
|
66 00:1D:C1:51:18:84 0/10 10 Learned
|
|
66 00:E0:4C:9A:1A:F3 0/3 3 Learned"""
|
|
|
|
def test_access_port_is_where_the_device_has_fewest_neighbours(self):
|
|
import netgear
|
|
best = netgear.best_ports({"10.66.1.100": self.SMALL, "10.66.1.102": self.BIG},
|
|
["00:1d:c1:51:34:c8", "00:e0:4c:9a:1a:f3", "00:0e:dd:e2:8e:c2"])
|
|
self.assertEqual(best["00:1d:c1:51:34:c8"], ("10.66.1.102", "0/14", 1)) # not the uplink 0/10
|
|
self.assertEqual(best["00:e0:4c:9a:1a:f3"], ("10.66.1.100", "0/3", 1))
|
|
self.assertEqual(best["00:0e:dd:e2:8e:c2"][2], 3) # shared port: too many
|
|
self.assertGreater(3, netgear.MAX_SHARED) # 3 = uplink
|
|
self.assertTrue(netgear.poe_delivering("0/14 Yes 32000 1 1500 29 54 Delivering Power No Error"))
|
|
self.assertFalse(netgear.poe_delivering("0/14 Yes 32000 0 0 0 0 Searching No Error"))
|
|
|
|
def test_never_cycles_an_uplink(self):
|
|
import netgear
|
|
orig = netgear.survey
|
|
netgear.survey = lambda switches, macs: {macs[0]: {"switch": "s", "port": "0/26", "shared": 3, "poe": False}}
|
|
try:
|
|
with self.assertRaises(RuntimeError):
|
|
netgear.poe_cycle_device(["s"], "00:0e:dd:e2:8e:c2")
|
|
finally:
|
|
netgear.survey = orig
|
|
|
|
|
|
class FakeCompanion(BaseHTTPRequestHandler):
|
|
"""Just enough of Companion 5's HTTP API."""
|
|
store = {}
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def _send(self, status, body):
|
|
data = body.encode() if isinstance(body, str) else body
|
|
self.send_response(status)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
s = self.store
|
|
path = urllib.parse.urlparse(self.path).path
|
|
if path == "/api/connections":
|
|
return self._send(200, json.dumps([{"id": c["id"], "label": l, "enabled": c["enabled"],
|
|
"status": {"category": c["category"], "message": c["message"]}}
|
|
for l, c in s["conns"].items()]))
|
|
m = re.match(r"^/api/variable/([^/]+)/([^/]+)/value$", path)
|
|
if m:
|
|
key = (urllib.parse.unquote(m.group(1)), urllib.parse.unquote(m.group(2)))
|
|
v = s["vars"].get(key)
|
|
return self._send(404, "Not found") if v is None else self._send(200, str(v))
|
|
m = re.match(r"^/api/custom-variable/([^/]+)/value$", path)
|
|
if m:
|
|
v = s["custom"].get(m.group(1))
|
|
return self._send(404, "Not found") if v is None else self._send(200, v)
|
|
self._send(404, "")
|
|
|
|
def do_POST(self):
|
|
s = self.store
|
|
u = urllib.parse.urlparse(self.path)
|
|
m = re.match(r"^/api/custom-variable/([^/]+)/value$", u.path)
|
|
if m:
|
|
if m.group(1) not in s["custom"]:
|
|
return self._send(404, "Not found")
|
|
s["custom"][m.group(1)] = urllib.parse.parse_qs(u.query, keep_blank_values=True)["value"][0]
|
|
return self._send(200, "ok")
|
|
m = re.match(r"^/api/connections/([^/]+)/restart$", u.path)
|
|
if m:
|
|
s["restarts"].append(m.group(1))
|
|
return self._send(200, "{}")
|
|
self._send(404, "")
|
|
|
|
|
|
class TestAgainstFakeCompanion(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
snap = healthy_snapshot(time.time())
|
|
FakeCompanion.store = {
|
|
"conns": snap.conns,
|
|
"vars": dict(snap.vars),
|
|
"custom": {"sp_state": "", "sp_hb": "", "sp_cmd": "", "flash_tick": "0"},
|
|
"restarts": [],
|
|
}
|
|
cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion)
|
|
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
|
|
cls.port = cls.server.server_address[1]
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
|
|
def test_client_round_trip_and_long_state(self):
|
|
comp = sp.Companion("127.0.0.1", self.port)
|
|
self.assertEqual(len(comp.connections()), len(FakeCompanion.store["conns"]))
|
|
self.assertEqual(comp.variable("HyperDeck_1", "recordingTime"), "28:00:00")
|
|
self.assertIsNone(comp.variable("HyperDeck_1", "nope"))
|
|
snap = healthy_snapshot(time.time())
|
|
tiles = sp.Engine(CONFIG).step(snap, time.time())
|
|
js = json.dumps(tiles, separators=(",", ":"), sort_keys=True)
|
|
self.assertEqual(comp.set_custom("sp_state", js), (True, 200))
|
|
self.assertEqual(json.loads(comp.get_custom("sp_state"))["all"]["x"], "ALL OK")
|
|
self.assertEqual(comp.set_custom("missing", "x"), (False, 404))
|
|
print("\n sp_state is %d bytes, %d in the URL" % (len(js), len(urllib.parse.quote(js, safe=""))))
|
|
|
|
def test_commands(self):
|
|
comp = sp.Companion("127.0.0.1", self.port)
|
|
store = FakeCompanion.store
|
|
snap = healthy_snapshot(SUNDAY)
|
|
engine = sp.Engine(CONFIG)
|
|
engine.executor = lambda cfg, fix_id, chk: self.fail("no real fix may run in a test")
|
|
now = SUNDAY
|
|
store["conns"]["VX1000"]["category"] = "error"
|
|
try:
|
|
sp.poll_connections(comp, snap)
|
|
run_steps(engine, snap, now, 4)
|
|
before = len(store["restarts"])
|
|
# hd3 is green, so its hold does nothing; LED is failing on VX1000, so that restarts.
|
|
sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now + 5)
|
|
self.assertEqual(store["restarts"][before:], ["id-VX1000"])
|
|
finally:
|
|
store["conns"]["VX1000"]["category"] = "good"
|
|
self.assertFalse(engine.service.on(now))
|
|
sp.handle_commands("service toggle", engine, comp, snap, now)
|
|
self.assertTrue(engine.service.on(now))
|
|
|
|
def test_end_to_end_loop(self):
|
|
"""The real run() loop: it writes state and heartbeat, and consumes button commands."""
|
|
cfg = json.loads(json.dumps(CONFIG))
|
|
cfg["companion"]["port"] = self.port
|
|
# keep it offline: no pings, no web, no launchd, no bonjour lookups
|
|
cfg["network"]["gateway"] = None
|
|
keep = ("conn", "hyperdeck", "dante", "tascam", "shure", "psm")
|
|
for t in cfg["tiles"]:
|
|
if "checks" in t:
|
|
t["checks"] = [c for c in t["checks"] if c["type"] in keep] or [{"type": "conn", "conn": "x32"}]
|
|
store = FakeCompanion.store
|
|
store["conns"]["ATEM"]["category"] = "error"
|
|
store["custom"]["sp_cmd"] = ";ack atem"
|
|
data_dir = tempfile.mkdtemp()
|
|
originals = (sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map,
|
|
sp.poll_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) + 1) # + the page number key
|
|
# The master is where the smallest Stream Deck can see it.
|
|
self.assertEqual(CONFIG["tiles"][0]["pos"], [0, 1])
|
|
|
|
def test_each_button_reads_its_own_tile(self):
|
|
for t in CONFIG["tiles"]:
|
|
if "pos" not in t:
|
|
continue
|
|
r, c = (str(x) for x in t["pos"])
|
|
layers = {l["id"]: l for l in self.controls[r][c]["style"]["layers"]}
|
|
for lid, key in (("box0", "color"), ("text0", "color"), ("text0", "text"), ("text0", "fontsize")):
|
|
expr = layers[lid][key]
|
|
self.assertTrue(expr["isExpression"])
|
|
self.assertIn("'$.%s." % t["id"], expr["value"])
|
|
self.assertIn("$(custom:sp_hb) > 15", expr["value"])
|
|
self.assertIn("$(custom:flash_tick) == 1", layers["box0"]["color"]["value"])
|
|
|
|
def test_actions(self):
|
|
hd1 = self.controls["1"]["1"]["steps"]["0"]
|
|
up = hd1["action_sets"]["up"][0]["options"]["value"]
|
|
self.assertEqual(up, {"value": "$(custom:sp_cmd);ack hd1", "isExpression": False})
|
|
self.assertEqual(hd1["action_sets"]["2000"][0]["options"]["value"]["value"], "$(custom:sp_cmd);restart hd1")
|
|
self.assertEqual(hd1["options"]["runWhileHeld"], [2000])
|
|
master = self.controls["0"]["1"]["steps"]["0"]
|
|
self.assertIn("ackall", master["action_sets"]["up"][0]["options"]["value"]["value"])
|
|
self.assertNotIn("2000", master["action_sets"])
|
|
ingest = self.controls["3"]["5"]["steps"]["0"]
|
|
self.assertIn("2000", ingest["action_sets"]) # restarts its services
|
|
self.assertIn("2000", self.controls["0"]["2"]["steps"]["0"]["action_sets"]) # REC group
|
|
net = self.controls["3"]["4"]["steps"]["0"]
|
|
self.assertNotIn("2000", net["action_sets"]) # nothing to fix there
|
|
ids = [a["id"] for r in self.controls.values() for c in r.values()
|
|
for s in c.get("steps", {}).values() for al in s["action_sets"].values() for a in al]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
|
|
def test_stale_text_and_file(self):
|
|
master = {l["id"]: l for l in self.controls["0"]["1"]["style"]["layers"]}
|
|
self.assertIn("`${'WATCHDOG\\nDOWN\\n'}${$(internal:instance_errors)} ERR`", master["text0"]["text"]["value"])
|
|
# No string '+' anywhere: in Companion 5 expressions it adds numbers.
|
|
for r in self.controls.values():
|
|
for c in r.values():
|
|
for layer in c.get("style", {}).get("layers", []):
|
|
for p in layer.values():
|
|
if isinstance(p, dict) and p.get("isExpression"):
|
|
self.assertNotRegex(p["value"], r"'\s*\+|\+\s*'")
|
|
path = os.path.join(tempfile.mkdtemp(), "p.companionconfig")
|
|
sys.argv = ["make_page.py", path]
|
|
make_page.main()
|
|
on_disk = make_page.build(sp.load_config(os.path.join(ROOT, "config.json")))
|
|
with gzip.open(path) as f:
|
|
self.assertEqual(json.load(f), json.loads(json.dumps(on_disk)))
|
|
|
|
|
|
class TestCoreMIDI(unittest.TestCase):
|
|
def test_listener_loopback(self):
|
|
import coremidi
|
|
src = coremidi.VirtualSource("status-panel-test")
|
|
try:
|
|
time.sleep(0.3)
|
|
lis = coremidi.Listener("status-panel-test")
|
|
lis.poll()
|
|
self.assertTrue(lis.connected)
|
|
src.send([0x90, 36, 100])
|
|
src.send([0xF8]) # clock is ignored
|
|
time.sleep(0.3)
|
|
self.assertEqual((lis.last_msg, lis.count), ("NOTE 36", 1))
|
|
snap = sp.Snapshot()
|
|
snap.listeners["status-panel-test"] = lis
|
|
res = sp.check_midi_rx({"source": "status-panel-test", "name": "IN"}, snap,
|
|
{"now": time.time(), "service": True})[0][1]
|
|
self.assertEqual(res.info, "IN 0s")
|
|
finally:
|
|
src.close()
|
|
time.sleep(0.3)
|
|
lis.poll()
|
|
self.assertFalse(lis.connected)
|
|
|
|
def test_describe(self):
|
|
import coremidi
|
|
self.assertEqual(coremidi.describe([0xB0, 7, 64]), "CC 7=64")
|
|
self.assertEqual(coremidi.describe([0xC2, 5]), "PC 5")
|
|
self.assertEqual(coremidi.describe([0x90, 60, 0]), "OFF 60")
|
|
self.assertEqual(coremidi.describe([0xF1, 0x20]), "MTC")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|