christianandClaude Opus 5.5
38b6616e3f
It now reads 19/19 to match Dante Controller. NA-Tracks, SuperRack and the ProPresenter Mac still alarm on their own buttons, so a drop is not flagged twice; DANTE just shows 18/19 while one of them is missing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
627 lines
30 KiB
Python
627 lines
30 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 sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.parse
|
|
from datetime import datetime
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, ROOT)
|
|
|
|
import make_page # noqa: E402
|
|
import status_panel as sp # noqa: E402
|
|
|
|
sp.log.addHandler(logging.NullHandler())
|
|
sp.log.propagate = False
|
|
|
|
CONFIG = sp.load_config(os.path.join(ROOT, "config.json"))
|
|
# The tests pin their own schedule so they do not depend on the site's choice.
|
|
CONFIG["service_windows"] = [{"days": ["sun"], "start": "07:00", "end": "12:30"}]
|
|
CONFIG["tiles"] = [dict(t, label="SERVICE") if t["id"] == "service" else t for t in CONFIG["tiles"]]
|
|
# A Sunday 08:00 (inside the service window) and a Wednesday 15:00 (outside it).
|
|
SUNDAY = datetime(2026, 10, 11, 8, 0).timestamp()
|
|
WEDNESDAY = datetime(2026, 10, 7, 15, 0).timestamp()
|
|
|
|
|
|
def healthy_snapshot(now):
|
|
"""Everything in config.json present and fine."""
|
|
snap = sp.Snapshot()
|
|
labels = {c["conn"] for t in CONFIG["tiles"] for c in t.get("checks", []) if c.get("conn")}
|
|
snap.conns = {l: {"id": "id-" + l, "enabled": True, "category": "good", "message": None} for l in labels}
|
|
snap.conns_time = now
|
|
for i in range(1, 7):
|
|
snap.vars[("HyperDeck_%d" % i, "status")] = "Preview"
|
|
snap.vars[("HyperDeck_%d" % i, "slotId")] = "1"
|
|
snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "28:00:00"
|
|
devices = set()
|
|
for t in CONFIG["tiles"]:
|
|
for c in t.get("checks", []):
|
|
if c["type"] == "dante":
|
|
devices |= set(c["devices"])
|
|
snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(sorted(devices))
|
|
snap.vars[("tascam", "error")] = "No Error"
|
|
snap.vars[("tascam", "caution")] = "No Caution"
|
|
for label, n in (("Axient_1", 4), ("Axient_2", 4), ("ULXD-4", 1)):
|
|
for ch in range(1, n + 1):
|
|
snap.vars[(label, "ch_%d_name" % ch)] = "VOX %d" % ch
|
|
snap.vars[(label, "ch_%d_battery_bars" % ch)] = "5"
|
|
snap.vars[(label, "ch_%d_interference_status" % ch)] = "NONE"
|
|
for label in ("PSM_1", "PSM_2", "PSM_3"):
|
|
for ch in (1, 2):
|
|
snap.vars[(label, "ch_%d_name" % ch)] = "PACK"
|
|
snap.vars[(label, "ch_%d_rf_mute" % ch)] = "UNMUTED"
|
|
snap.midi = {"NA Ingest": {"peers": [{"name": "Lighting PC", "address": "", "state": "running"},
|
|
{"name": "SuperRack-LiveBox", "address": "", "state": "running"}]}}
|
|
snap.midi_time = now
|
|
snap.mdns, snap.mdns_time = {"NA Tracks", "Lighting PC", "NA Ingest"}, now
|
|
snap.ping = {"10.1.3.227": 0, "10.6.7.13": 0, "10.1.0.1": 0}
|
|
snap.link = True
|
|
snap.launchd = {"org.waterschurch.resi-propresenter": 884, "org.waterschurch.lobby-captions": 900}
|
|
snap.launchd_time = now
|
|
snap.procs = {"/Applications/ProdDash.app/Contents/MacOS/ProdDash": True}
|
|
snap.http = {"https://central.resi.io/": True, "http://10.6.7.13:8080/": True}
|
|
snap.disk = {"/": 97}
|
|
return snap
|
|
|
|
|
|
def run_steps(engine, snap, start, seconds, step=1.0):
|
|
tiles, t = None, start
|
|
while t <= start + seconds:
|
|
snap.conns_time = t
|
|
tiles = engine.step(snap, t)
|
|
t += step
|
|
return tiles, t
|
|
|
|
|
|
def level(tile):
|
|
for name, ((b, f), _) in sp.COLORS.items():
|
|
if (tile["b"], tile["f"]) == (b, f):
|
|
return name
|
|
return "?"
|
|
|
|
|
|
class TestChecks(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = SUNDAY
|
|
self.snap = healthy_snapshot(self.now)
|
|
self.ctx = {"now": self.now, "service": True, "any_recording": False}
|
|
|
|
def test_all_green_when_healthy(self):
|
|
tiles = sp.Engine(CONFIG).step(self.snap, self.now)
|
|
bad = {k: v["x"] for k, v in tiles.items() if level(v) not in ("ok", "service", "off")}
|
|
self.assertEqual(bad, {})
|
|
self.assertEqual(tiles["all"]["x"], "ALL OK")
|
|
self.assertEqual(tiles["hd1"]["x"], "HD1\n28h")
|
|
self.assertEqual(tiles["dante"]["x"], "DANTE\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.OFF)
|
|
self.assertEqual(sp.conn_result(self.snap, "nope", self.now).msg, "NOT IN COMPANION")
|
|
self.assertEqual(sp.conn_result(self.snap, "x32", self.now + 60).msg, "NO DATA")
|
|
|
|
def test_hyperdeck_disk_and_record_sync(self):
|
|
s = self.snap
|
|
s.vars[("HyperDeck_2", "slotId")] = "-"
|
|
s.vars[("HyperDeck_2", "recordingTime")] = "--:--:--"
|
|
out = dict(sp.check_hyperdeck({"conn": "HyperDeck_2"}, s, self.ctx))
|
|
self.assertEqual(out["disk"].msg, "NO DISK")
|
|
self.assertTrue(out["disk"].service_only)
|
|
s.vars[("HyperDeck_3", "recordingTime")] = "0:20:00"
|
|
self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].msg, "LOW 20m")
|
|
s.vars[("HyperDeck_3", "recordingTime")] = "0:04:00"
|
|
self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].level, sp.ERR)
|
|
ctx = dict(self.ctx, any_recording=True)
|
|
self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))["rec"].msg, "NOT REC")
|
|
s.vars[("HyperDeck_1", "status")] = "Record"
|
|
out = dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))
|
|
self.assertEqual(out["rec"].level, sp.OK)
|
|
self.assertEqual(out["conn"].info, "REC 28h")
|
|
|
|
def test_optional_deck(self):
|
|
for i in (2, 4):
|
|
self.snap.vars[("HyperDeck_%d" % i, "slotId")] = "-"
|
|
self.snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "--:--:--"
|
|
for i in (1, 3, 5, 6):
|
|
self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual((level(tiles["hd2"]), tiles["hd2"]["x"]), ("off", "HD2\nNO DISK"))
|
|
self.assertEqual(level(tiles["rec"]), "ok")
|
|
self.assertEqual(tiles["all"]["x"], "ALL OK")
|
|
self.snap.vars[("HyperDeck_2", "slotId")] = "1" # disk goes in: now it is watched
|
|
self.snap.vars[("HyperDeck_2", "recordingTime")] = "20:00:00"
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual(tiles["hd2"]["x"], "HD2\nNOT REC")
|
|
self.assertEqual(level(tiles["hd2"]), "err")
|
|
|
|
def test_one_deck_not_recording_blinks(self):
|
|
engine = sp.Engine(CONFIG)
|
|
for i in (1, 2, 3, 5, 6):
|
|
self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
|
|
tiles, _ = run_steps(engine, self.snap, self.now, 4)
|
|
self.assertEqual(tiles["hd4"]["x"], "HD4\nNOT REC")
|
|
self.assertEqual(level(tiles["hd4"]), "err")
|
|
self.assertEqual(tiles["all"]["x"], "1 DOWN\nHD4 NOT REC")
|
|
self.assertEqual(tiles["rec"]["x"], "REC\nHD4 NOT REC")
|
|
|
|
def test_dante(self):
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["NA-Tracks"])
|
|
out = dict(sp.check_dante({"conn": "audinate-dantecontroller",
|
|
"devices": {"NA-Tracks": "DANTE", "BOH-WING": "WING"}}, self.snap, self.ctx))
|
|
self.assertEqual(out["BOH-WING"].msg, "WING GONE")
|
|
self.assertEqual(out["NA-Tracks"].level, sp.OK)
|
|
self.snap.vars[("audinate-dantecontroller", "devices")] = None
|
|
out = dict(sp.check_dante({"conn": "audinate-dantecontroller", "devices": ["BOH-WING"]}, self.snap, self.ctx))
|
|
self.assertEqual(out["conn"].msg, "NO DATA")
|
|
|
|
def test_dante_count_includes_other_buttons(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\n18/19")
|
|
self.assertEqual(level(tiles["dante"]), "ok") # the alarm is on TRACKS, not twice
|
|
self.assertEqual(tiles["tracks"]["x"], "TRACKS\nDANTE 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_midi_peer_joining_is_not_an_error(self):
|
|
for p in self.snap.midi["NA Ingest"]["peers"]:
|
|
p["state"] = "connected" # joined, clock sync not finished yet
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 8)
|
|
self.assertEqual(tiles["lightpc"]["x"], "LIGHT PC\nMIDI OK")
|
|
|
|
def test_midi_peer_and_bonjour(self):
|
|
self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"]
|
|
if p["name"] != "Lighting PC"]
|
|
self.snap.mdns.discard("NA Tracks")
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual(tiles["lightpc"]["x"], "LIGHT PC\nMIDI LOST")
|
|
self.assertEqual(tiles["tracks"]["x"], "TRACKS\nMIDI SESS GONE")
|
|
|
|
def test_bible_tv_app_down(self):
|
|
self.snap.http["http://10.6.7.13:8080/"] = False
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual(tiles["bibletv"]["x"], "BIBLE TV\nAPP DOWN")
|
|
|
|
def test_http_intervals(self):
|
|
checks = dict(sp.http_checks(CONFIG))
|
|
self.assertEqual(checks["http://10.6.7.13:8080/"], 5)
|
|
self.assertEqual(checks["https://central.resi.io/"], 30)
|
|
|
|
def test_ingest_services(self):
|
|
del self.snap.launchd["org.waterschurch.resi-propresenter"]
|
|
self.snap.launchd["org.waterschurch.lobby-captions"] = None
|
|
self.snap.disk["/"] = 6
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
|
|
self.assertEqual(tiles["ingest"]["x"], "INGEST\nBRIDGE NOT LOADED\n+2")
|
|
|
|
|
|
class TestTextSize(unittest.TestCase):
|
|
def test_no_word_is_split(self):
|
|
for text in ("TASCAM\nOK", "SERVICE\nON\nmanual", "OBS\nOFFLINE", "INGEST\n97GB free",
|
|
"4 DOWN\nHD2 NO DISK", "REC\nHD4 NO DISK\n+1", "TRACKS\nMIDI SESS GONE"):
|
|
size = sp.fit_size(text)
|
|
longest = max(len(w) for w in text.split())
|
|
self.assertLessEqual(longest, int(sp.LINE_CAPACITY / size), text)
|
|
|
|
def test_short_text_stays_big(self):
|
|
self.assertGreaterEqual(sp.fit_size("HD1\n28h"), 29)
|
|
self.assertGreater(sp.fit_size("ATEM\nOK"), sp.fit_size("TASCAM\nOK"))
|
|
|
|
def test_every_tile_has_a_size(self):
|
|
tiles = sp.Engine(CONFIG).step(healthy_snapshot(SUNDAY), SUNDAY)
|
|
for t in tiles.values():
|
|
self.assertTrue(12 <= t["s"] <= 30, t)
|
|
|
|
|
|
class TestBehaviour(unittest.TestCase):
|
|
def setUp(self):
|
|
self.snap = healthy_snapshot(SUNDAY)
|
|
|
|
def test_debounce(self):
|
|
engine = sp.Engine(CONFIG)
|
|
self.snap.conns["x32"]["category"] = "error"
|
|
tiles = engine.step(self.snap, SUNDAY)
|
|
self.assertEqual(level(tiles["x32"]), "ok") # one bad reading is not enough
|
|
tiles, t = run_steps(engine, self.snap, SUNDAY + 1, 3)
|
|
self.assertEqual(level(tiles["x32"]), "err")
|
|
self.snap.conns["x32"]["category"] = "good"
|
|
self.snap.conns_time = t
|
|
self.assertEqual(level(engine.step(self.snap, t)["x32"]), "ok") # recovery is immediate
|
|
|
|
def test_blip_shorter_than_debounce_is_ignored(self):
|
|
engine = sp.Engine(CONFIG)
|
|
self.snap.conns["x32"]["category"] = "error"
|
|
run_steps(engine, self.snap, SUNDAY, 1)
|
|
self.snap.conns["x32"]["category"] = "good"
|
|
tiles, _ = run_steps(engine, self.snap, SUNDAY + 2, 5)
|
|
self.assertEqual(level(tiles["x32"]), "ok")
|
|
|
|
def test_flapping_shows_amber(self):
|
|
engine = sp.Engine(CONFIG)
|
|
t = SUNDAY
|
|
for _ in range(3):
|
|
self.snap.conns["x32"]["category"] = "error"
|
|
_, t = run_steps(engine, self.snap, t, 4)
|
|
self.snap.conns["x32"]["category"] = "good"
|
|
tiles, t = run_steps(engine, self.snap, t, 2)
|
|
self.assertEqual(level(tiles["x32"]), "warn")
|
|
self.assertEqual(tiles["x32"]["x"], "X32\nOFFLINE x3")
|
|
tiles, _ = run_steps(engine, self.snap, t + 700, 1) # quiet for 10 minutes -> green again
|
|
self.assertEqual(level(tiles["x32"]), "ok")
|
|
|
|
def test_ack_then_recover_then_fail_again(self):
|
|
engine = sp.Engine(CONFIG)
|
|
self.snap.conns["ATEM"]["category"] = "error"
|
|
tiles, t = run_steps(engine, self.snap, SUNDAY, 4)
|
|
self.assertEqual(level(tiles["atem"]), "err")
|
|
self.assertEqual(engine.ack("atem"), 1)
|
|
tiles, t = run_steps(engine, self.snap, t, 1)
|
|
self.assertEqual(level(tiles["atem"]), "ack")
|
|
self.assertEqual(level(tiles["all"]), "ack")
|
|
self.snap.conns["ATEM"]["category"] = "good"
|
|
tiles, t = run_steps(engine, self.snap, t, 1)
|
|
self.assertEqual(level(tiles["atem"]), "ok")
|
|
self.snap.conns["ATEM"]["category"] = "error"
|
|
tiles, t = run_steps(engine, self.snap, t, 4)
|
|
self.assertEqual(level(tiles["atem"]), "err") # a new failure blinks again
|
|
|
|
def test_group_and_master_ack(self):
|
|
engine = sp.Engine(CONFIG)
|
|
self.snap.conns["VX1000"]["category"] = "error"
|
|
self.snap.conns["x32"]["category"] = "error"
|
|
tiles, t = run_steps(engine, self.snap, SUNDAY, 4)
|
|
self.assertEqual(tiles["all"]["x"].split("\n")[0], "2 DOWN")
|
|
engine.ack("video")
|
|
tiles, t = run_steps(engine, self.snap, t, 1)
|
|
self.assertEqual(level(tiles["video"]), "ack")
|
|
self.assertEqual(level(tiles["x32"]), "err")
|
|
engine.ack("all")
|
|
tiles, t = run_steps(engine, self.snap, t, 1)
|
|
self.assertEqual(level(tiles["all"]), "ack")
|
|
|
|
def test_outside_service_nothing_blinks(self):
|
|
snap = healthy_snapshot(WEDNESDAY)
|
|
snap.conns["OBS-Youtube"]["category"] = "error" # OBS PC is off midweek
|
|
snap.conns["x32"]["category"] = "error"
|
|
snap.vars[("HyperDeck_3", "slotId")] = "-" # a deck that is always used
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), snap, WEDNESDAY, 4)
|
|
self.assertEqual(level(tiles["obs"]), "idle")
|
|
self.assertEqual(level(tiles["hd3"]), "idle")
|
|
self.assertEqual(tiles["hd3"]["x"], "HD3\nNO DISK")
|
|
self.assertEqual(level(tiles["x32"]), "quiet")
|
|
self.assertEqual(tiles["all"]["x"], "1 DOWN\nX32 OFFLINE")
|
|
self.assertEqual(level(tiles["all"]), "quiet")
|
|
self.assertEqual(tiles["service"]["x"], "SERVICE\nOFF")
|
|
for t in tiles.values():
|
|
self.assertEqual((t["b"], t["f"]), (t["b2"], t["f2"]), t["x"]) # no blinking anywhere
|
|
|
|
def test_service_schedule_and_override(self):
|
|
path = os.path.join(tempfile.mkdtemp(), "service.json")
|
|
mode = sp.ServiceMode(CONFIG["service_windows"], path)
|
|
self.assertTrue(mode.on(SUNDAY))
|
|
self.assertFalse(mode.on(WEDNESDAY))
|
|
mode.set(True, WEDNESDAY) # button: on for a Wednesday event
|
|
self.assertTrue(mode.on(WEDNESDAY + 3600))
|
|
self.assertTrue(sp.ServiceMode(CONFIG["service_windows"], path).on(WEDNESDAY + 3600)) # survives restart
|
|
sunday_start = datetime(2026, 10, 11, 7, 0).timestamp()
|
|
self.assertTrue(mode.on(sunday_start)) # window edge clears the override
|
|
self.assertIsNone(mode.override)
|
|
mode.set(False, sunday_start + 60) # turned off early on Sunday
|
|
self.assertFalse(mode.on(sunday_start + 120))
|
|
self.assertFalse(mode.on(datetime(2026, 10, 11, 12, 31).timestamp()))
|
|
self.assertIsNone(mode.override) # end of window clears it too
|
|
|
|
def test_network_burst(self):
|
|
engine = sp.Engine(CONFIG)
|
|
t = SUNDAY
|
|
run_steps(engine, self.snap, t, 2)
|
|
for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"): # the Oct 5 10:21 drop
|
|
self.snap.conns[label]["category"] = "error"
|
|
tiles, t = run_steps(engine, self.snap, t + 3, 4)
|
|
self.assertEqual(tiles["net"]["x"], "NET\n5 DROPPED")
|
|
self.assertEqual(level(tiles["net"]), "err")
|
|
for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"):
|
|
self.snap.conns[label]["category"] = "good"
|
|
tiles, t = run_steps(engine, self.snap, t, 1)
|
|
self.assertTrue(tiles["net"]["x"].startswith("NET\nBLIP "))
|
|
self.assertEqual(level(tiles["net"]), "warn")
|
|
tiles, _ = run_steps(engine, self.snap, t + 200, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nOK")
|
|
|
|
def test_racks_powering_off_is_not_a_network_event(self):
|
|
engine = sp.Engine(CONFIG)
|
|
run_steps(engine, self.snap, SUNDAY, 2)
|
|
for label in ("Axient_1", "Axient_2", "ULXD-4", "PSM_1", "PSM_2", "PSM_3"):
|
|
self.snap.conns[label]["category"] = "error"
|
|
tiles, _ = run_steps(engine, self.snap, SUNDAY + 3, 4)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nOK")
|
|
self.assertEqual(tiles["rf"]["x"].split("\n")[-1], "+5")
|
|
|
|
def test_gateway_and_link(self):
|
|
self.snap.ping["10.1.0.1"] = 2
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nNO GATEWAY")
|
|
self.snap.link = False
|
|
tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
|
|
self.assertEqual(tiles["net"]["x"], "NET\nLINK DOWN\n+1")
|
|
|
|
def test_restart_targets(self):
|
|
engine = sp.Engine(CONFIG)
|
|
self.assertEqual(engine.restart_targets("hd3"), ["HyperDeck_3"])
|
|
self.assertEqual(engine.restart_targets("led"), ["VX1000", "UHDJr"])
|
|
self.assertEqual(engine.restart_targets("tracks"), ["ableton"]) # not the shared Dante one
|
|
self.assertEqual(engine.restart_targets("ingest"), [])
|
|
|
|
|
|
class FakeCompanion(BaseHTTPRequestHandler):
|
|
"""Just enough of Companion 5's HTTP API."""
|
|
store = {}
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def _send(self, status, body):
|
|
data = body.encode() if isinstance(body, str) else body
|
|
self.send_response(status)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
s = self.store
|
|
path = urllib.parse.urlparse(self.path).path
|
|
if path == "/api/connections":
|
|
return self._send(200, json.dumps([{"id": c["id"], "label": l, "enabled": c["enabled"],
|
|
"status": {"category": c["category"], "message": c["message"]}}
|
|
for l, c in s["conns"].items()]))
|
|
m = re.match(r"^/api/variable/([^/]+)/([^/]+)/value$", path)
|
|
if m:
|
|
key = (urllib.parse.unquote(m.group(1)), urllib.parse.unquote(m.group(2)))
|
|
v = s["vars"].get(key)
|
|
return self._send(404, "Not found") if v is None else self._send(200, str(v))
|
|
m = re.match(r"^/api/custom-variable/([^/]+)/value$", path)
|
|
if m:
|
|
v = s["custom"].get(m.group(1))
|
|
return self._send(404, "Not found") if v is None else self._send(200, v)
|
|
self._send(404, "")
|
|
|
|
def do_POST(self):
|
|
s = self.store
|
|
u = urllib.parse.urlparse(self.path)
|
|
m = re.match(r"^/api/custom-variable/([^/]+)/value$", u.path)
|
|
if m:
|
|
if m.group(1) not in s["custom"]:
|
|
return self._send(404, "Not found")
|
|
s["custom"][m.group(1)] = urllib.parse.parse_qs(u.query, keep_blank_values=True)["value"][0]
|
|
return self._send(200, "ok")
|
|
m = re.match(r"^/api/connections/([^/]+)/restart$", u.path)
|
|
if m:
|
|
s["restarts"].append(m.group(1))
|
|
return self._send(200, "{}")
|
|
self._send(404, "")
|
|
|
|
|
|
class TestAgainstFakeCompanion(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
snap = healthy_snapshot(time.time())
|
|
FakeCompanion.store = {
|
|
"conns": snap.conns,
|
|
"vars": dict(snap.vars),
|
|
"custom": {"sp_state": "", "sp_hb": "", "sp_cmd": "", "flash_tick": "0"},
|
|
"restarts": [],
|
|
}
|
|
cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion)
|
|
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
|
|
cls.port = cls.server.server_address[1]
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
|
|
def test_client_round_trip_and_long_state(self):
|
|
comp = sp.Companion("127.0.0.1", self.port)
|
|
self.assertEqual(len(comp.connections()), len(FakeCompanion.store["conns"]))
|
|
self.assertEqual(comp.variable("HyperDeck_1", "recordingTime"), "28:00:00")
|
|
self.assertIsNone(comp.variable("HyperDeck_1", "nope"))
|
|
snap = healthy_snapshot(time.time())
|
|
tiles = sp.Engine(CONFIG).step(snap, time.time())
|
|
js = json.dumps(tiles, separators=(",", ":"), sort_keys=True)
|
|
self.assertEqual(comp.set_custom("sp_state", js), (True, 200))
|
|
self.assertEqual(json.loads(comp.get_custom("sp_state"))["all"]["x"], "ALL OK")
|
|
self.assertEqual(comp.set_custom("missing", "x"), (False, 404))
|
|
print("\n sp_state is %d bytes, %d in the URL" % (len(js), len(urllib.parse.quote(js, safe=""))))
|
|
|
|
def test_commands(self):
|
|
comp = sp.Companion("127.0.0.1", self.port)
|
|
snap = sp.Snapshot()
|
|
sp.poll_connections(comp, snap)
|
|
engine = sp.Engine(CONFIG)
|
|
now = SUNDAY
|
|
sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now)
|
|
self.assertEqual(FakeCompanion.store["restarts"][-3:], ["id-HyperDeck_3", "id-VX1000", "id-UHDJr"])
|
|
self.assertFalse(engine.service.on(now))
|
|
sp.handle_commands("service toggle", engine, comp, snap, now)
|
|
self.assertTrue(engine.service.on(now))
|
|
|
|
def test_end_to_end_loop(self):
|
|
"""The real run() loop: it writes state and heartbeat, and consumes button commands."""
|
|
cfg = json.loads(json.dumps(CONFIG))
|
|
cfg["companion"]["port"] = self.port
|
|
# keep it offline: no pings, no web, no launchd, no bonjour lookups
|
|
cfg["network"]["gateway"] = None
|
|
keep = ("conn", "hyperdeck", "dante", "tascam", "shure", "psm")
|
|
for t in cfg["tiles"]:
|
|
if "checks" in t:
|
|
t["checks"] = [c for c in t["checks"] if c["type"] in keep] or [{"type": "conn", "conn": "x32"}]
|
|
store = FakeCompanion.store
|
|
store["conns"]["ATEM"]["category"] = "error"
|
|
store["custom"]["sp_cmd"] = ";ack atem"
|
|
data_dir = tempfile.mkdtemp()
|
|
originals = sp.poll_mdns, sp.poll_system, sp.poll_midi
|
|
sp.poll_mdns = lambda snap, seconds=3: None
|
|
sp.poll_system = lambda snap, cfg: None
|
|
sp.poll_midi = lambda snap, cfg: None # CoreMIDI has its own test
|
|
stop = threading.Event()
|
|
loop = threading.Thread(target=sp.run, args=(cfg, data_dir), kwargs={"stop": stop}, daemon=True)
|
|
try:
|
|
loop.start()
|
|
deadline = time.time() + 12
|
|
state = {}
|
|
while time.time() < deadline:
|
|
raw = store["custom"]["sp_state"]
|
|
state = json.loads(raw) if raw else {}
|
|
if state.get("atem", {}).get("x") == "ATEM\nOFFLINE":
|
|
break
|
|
time.sleep(0.25)
|
|
finally:
|
|
stop.set()
|
|
loop.join(timeout=15)
|
|
sp.poll_mdns, sp.poll_system, sp.poll_midi = originals
|
|
store["conns"]["ATEM"]["category"] = "good"
|
|
self.assertFalse(loop.is_alive()) # it shuts down cleanly
|
|
self.assertTrue(store["custom"]["sp_hb"].isdigit())
|
|
self.assertLess(abs(int(store["custom"]["sp_hb"]) - time.time()), 5)
|
|
self.assertEqual(store["custom"]["sp_cmd"], "") # the command was consumed
|
|
self.assertIn("atem", state)
|
|
self.assertIn("ATEM", state["atem"]["x"])
|
|
|
|
|
|
class TestPage(unittest.TestCase):
|
|
def setUp(self):
|
|
self.export = make_page.build(CONFIG)
|
|
self.controls = self.export["page"]["controls"]
|
|
|
|
def test_layout(self):
|
|
self.assertEqual(self.export["type"], "page")
|
|
self.assertEqual(self.export["oldPageNumber"], 19)
|
|
self.assertEqual(self.controls["0"]["0"], {"type": "pageup"})
|
|
self.assertEqual(self.controls["2"]["0"], {"type": "pagedown"})
|
|
positioned = [t for t in CONFIG["tiles"] if "pos" in t]
|
|
n = sum(1 for r in self.controls.values() for c in r.values() if c["type"] == "button-layered")
|
|
self.assertEqual(n, len(positioned))
|
|
# The master is where the smallest Stream Deck can see it.
|
|
self.assertEqual(CONFIG["tiles"][0]["pos"], [0, 1])
|
|
|
|
def test_each_button_reads_its_own_tile(self):
|
|
for t in CONFIG["tiles"]:
|
|
if "pos" not in t:
|
|
continue
|
|
r, c = (str(x) for x in t["pos"])
|
|
layers = {l["id"]: l for l in self.controls[r][c]["style"]["layers"]}
|
|
for lid, key in (("box0", "color"), ("text0", "color"), ("text0", "text"), ("text0", "fontsize")):
|
|
expr = layers[lid][key]
|
|
self.assertTrue(expr["isExpression"])
|
|
self.assertIn("'$.%s." % t["id"], expr["value"])
|
|
self.assertIn("$(custom:sp_hb) > 15", expr["value"])
|
|
self.assertIn("$(custom:flash_tick) == 1", layers["box0"]["color"]["value"])
|
|
|
|
def test_actions(self):
|
|
hd1 = self.controls["1"]["1"]["steps"]["0"]
|
|
up = hd1["action_sets"]["up"][0]["options"]["value"]
|
|
self.assertEqual(up, {"value": "$(custom:sp_cmd);ack hd1", "isExpression": False})
|
|
self.assertEqual(hd1["action_sets"]["2000"][0]["options"]["value"]["value"], "$(custom:sp_cmd);restart hd1")
|
|
self.assertEqual(hd1["options"]["runWhileHeld"], [2000])
|
|
master = self.controls["0"]["1"]["steps"]["0"]
|
|
self.assertIn("ackall", master["action_sets"]["up"][0]["options"]["value"]["value"])
|
|
self.assertNotIn("2000", master["action_sets"])
|
|
ingest = self.controls["3"]["5"]["steps"]["0"]
|
|
self.assertNotIn("2000", ingest["action_sets"]) # nothing to restart there
|
|
ids = [a["id"] for r in self.controls.values() for c in r.values()
|
|
for s in c.get("steps", {}).values() for al in s["action_sets"].values() for a in al]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
|
|
def test_stale_text_and_file(self):
|
|
master = {l["id"]: l for l in self.controls["0"]["1"]["style"]["layers"]}
|
|
self.assertIn("`${'WATCHDOG\\nDOWN\\n'}${$(internal:instance_errors)} ERR`", master["text0"]["text"]["value"])
|
|
# No string '+' anywhere: in Companion 5 expressions it adds numbers.
|
|
for r in self.controls.values():
|
|
for c in r.values():
|
|
for layer in c.get("style", {}).get("layers", []):
|
|
for p in layer.values():
|
|
if isinstance(p, dict) and p.get("isExpression"):
|
|
self.assertNotRegex(p["value"], r"'\s*\+|\+\s*'")
|
|
path = os.path.join(tempfile.mkdtemp(), "p.companionconfig")
|
|
sys.argv = ["make_page.py", path]
|
|
make_page.main()
|
|
on_disk = make_page.build(sp.load_config(os.path.join(ROOT, "config.json")))
|
|
with gzip.open(path) as f:
|
|
self.assertEqual(json.load(f), json.loads(json.dumps(on_disk)))
|
|
|
|
|
|
class TestCoreMIDI(unittest.TestCase):
|
|
def test_listener_loopback(self):
|
|
import coremidi
|
|
src = coremidi.VirtualSource("status-panel-test")
|
|
try:
|
|
time.sleep(0.3)
|
|
lis = coremidi.Listener("status-panel-test")
|
|
lis.poll()
|
|
self.assertTrue(lis.connected)
|
|
src.send([0x90, 36, 100])
|
|
src.send([0xF8]) # clock is ignored
|
|
time.sleep(0.3)
|
|
self.assertEqual((lis.last_msg, lis.count), ("NOTE 36", 1))
|
|
snap = sp.Snapshot()
|
|
snap.listeners["status-panel-test"] = lis
|
|
res = sp.check_midi_rx({"source": "status-panel-test", "name": "IN"}, snap,
|
|
{"now": time.time(), "service": True})[0][1]
|
|
self.assertEqual(res.info, "IN 0s")
|
|
finally:
|
|
src.close()
|
|
time.sleep(0.3)
|
|
lis.poll()
|
|
self.assertFalse(lis.connected)
|
|
|
|
def test_describe(self):
|
|
import coremidi
|
|
self.assertEqual(coremidi.describe([0xB0, 7, 64]), "CC 7=64")
|
|
self.assertEqual(coremidi.describe([0xC2, 5]), "PC 5")
|
|
self.assertEqual(coremidi.describe([0x90, 60, 0]), "OFF 60")
|
|
self.assertEqual(coremidi.describe([0xF1, 0x20]), "MTC")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|