Files
status-indicators/tests/test_campus.py
T
christianandClaude Opus 5.5 777afac0b1 NA's CAMPUS page always shows a campus, and says which
Opened directly (nothing tapped), page 99 was all purple: no campus chosen. It now shows the first
other campus until one is tapped, and the back key reads WS / ALL SITES.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-20 16:28:21 -04:00

330 lines
20 KiB
Python

"""Auto-setup for the other campuses: an inventory becomes keys that pass every rule, laid out like
North Attleboro's; a running campus only gains what is new (never loses anything, never gets back
what someone took off); a computer's join report is checked; the lines people paste are right."""
import io
import json
import os
import socket
import sys
import tarfile
import tempfile
import threading
import time
import unittest
from http.server import ThreadingHTTPServer
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
sys.path.insert(0, os.path.join(ROOT, "tests"))
import schema # noqa: E402
import make_page # noqa: E402
from settings import autoconf as ac, campus, discover, server # noqa: E402
from test_settings import get # noqa: E402
from test_status_panel import FakeCompanion, healthy_snapshot # noqa: E402
BASE = json.load(open(os.path.join(ROOT, "sites", "_template", "config.json")))
def small():
return {
"ingest": {"name": "SC Ingest", "hostname": "SC-Ingest.local", "ips": ["10.4.1.1", "10.6.20.5"],
"wired": {"interface": "en0", "ip": "10.4.1.1", "net": "10.4.0.0/16", "gateway": "10.4.0.1"}},
"connections": [
{"label": "HyperDeck_1", "module": "bmd-hyperdeck", "host": "10.4.10.11", "enabled": True},
{"label": "HyperDeck_2", "module": "bmd-hyperdeck", "host": "10.4.10.12", "enabled": True},
{"label": "x32", "module": "behringer-x32", "host": "10.4.1.112", "enabled": True},
{"label": "dante", "module": "audinate-dantecontroller", "host": "", "enabled": True},
{"label": "ProP", "module": "renewedvision-propresenter-api", "host": "10.4.14.20", "enabled": True},
{"label": "Govee", "module": "govee-lights", "host": "", "enabled": True},
{"label": "old_obs", "module": "obs-studio", "host": "10.4.9.9", "enabled": False}],
"dante": {"conn": "dante", "devices": ["SC-X32", "SC-ProP-Mac"]},
"midi": {"": {"SC Ingest": {"port": 5004, "peers": [
{"name": "Lighting PC", "address": "10.4.10.206:5004", "state": "running"},
{"name": "Phone", "address": "10.6.20.77:5004", "state": "running"}]}}}, # Wi-Fi: not made a key
"computers": {
"prop": {"name": "ProPresenter Mac", "os": "mac", "user": "media", "ip": "10.4.14.20", "ips": ["10.4.14.20"], "reachable": True,
"procs": ["ProPresenter", "Finder"], "apps": ["/Applications/ProPresenter.app", "/Applications/Safari.app"]},
"lighting_pc": {"name": "Lighting PC", "os": "windows", "user": "media", "ip": "10.4.10.206", "ips": ["10.4.10.206"],
"reachable": True, "procs": ["Vista.exe", "rtpMIDI.exe", "Arena.exe"], "tasks": ["Start Vista", "Start rtpMIDI"],
"macs": ["aa:bb:cc:dd:ee:ff"]},
"off_pc": {"name": "Office", "os": "windows", "user": "x", "ip": "10.4.3.3", "ips": ["10.4.3.3"], "reachable": False}}}
def big():
inv = small()
inv["connections"] += [{"label": "HyperDeck_%d" % n, "module": "bmd-hyperdeck", "host": "10.4.10.%d" % (10 + n), "enabled": True} for n in range(3, 7)]
inv["connections"] += [{"label": "Axient_%d" % n, "module": "shure-wireless", "host": "10.4.1.%d" % (100 + n), "enabled": True, "channels": 4} for n in range(1, 4)]
inv["connections"] += [{"label": "PSM_%d" % n, "module": "shure-psm1000", "host": "10.4.1.%d" % (110 + n), "enabled": True, "channels": 2} for n in range(1, 9)]
inv["connections"] += [{"label": m, "module": mod, "host": "10.4.5.%d" % i, "enabled": True}
for i, (m, mod) in enumerate([("ATEM", "bmd-atem"), ("VX1000", "novastar-controller"), ("OBS", "obs-studio"),
("RESI", "resi-decoders"), ("tascam", "tascam-da-6400"), ("Arena", "resolume-arena")])]
return inv
def layout_of(cfg):
return {tuple(t["pos"]): t["label"] for t in cfg["tiles"] if t.get("pos")}
class TestBuild(unittest.TestCase):
def test_a_small_campus_gets_one_key_per_thing_in_order_and_passes_the_rules(self):
cfg, seen = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "4821", (21, 22))
self.assertEqual(schema.validate(cfg)[0], [])
labels = [t["label"] for t in cfg["tiles"] if not t.get("kind")]
self.assertEqual(labels, ["HD1", "HD2", "DANTE", "X32", "PROP", "VISTA", "MIDI"]) # REC, AUDIO, SHOW; no group keys
self.assertNotIn("RTPMIDI", labels) # rides in the MIDI key
midi = next(t for t in cfg["tiles"] if t["label"] == "MIDI")
self.assertEqual([c["peer"] for c in midi["checks"] if c["type"] == "midi_peer"], ["Lighting PC"]) # the Wi-Fi phone is not
self.assertEqual(midi["checks"][-2]["type"], "midi_wired")
self.assertEqual(midi["checks"][-1]["params"]["exe"], "rtpMIDI.exe")
prop = next(t for t in cfg["tiles"] if t["label"] == "PROP")
self.assertEqual([c["type"] for c in prop["checks"]], ["conn", "remote"]) # Companion connection + the program
self.assertEqual(prop["checks"][1]["cmd"], "pgrep -xq ProPresenter")
self.assertNotIn("ARENA", labels) # running, but no start task on that PC
self.assertNotIn("off_pc", cfg["hosts"]) # not reachable: not added
self.assertEqual(cfg["hosts"]["lighting_pc"]["ssh"], "media@10.4.10.206")
self.assertEqual((cfg["page"], cfg["restart_page"]["page"]), (21, 22))
self.assertEqual([k["label"] for k in cfg["restart_page"]["keys"]], ["PROP", "VISTA"])
self.assertEqual(cfg["site"]["editing_code"], "4821")
self.assertEqual(cfg["site"]["ingest_ips"][0], "10.4.1.1")
self.assertTrue(cfg["discover"]["auto"])
self.assertIn("hyperdeck", cfg["auto_fix"]["fixes"])
self.assertFalse(any(f.startswith("restart_") for f in cfg["auto_fix"]["fixes"])) # restarts stay a hold
for t in cfg["tiles"]: # every key's hold does something
if not t.get("kind") and not t.get("members"):
fixable = any(c.get("fix") or c["type"] in ("conn", "hyperdeck", "tascam", "shure", "psm") for c in t["checks"])
self.assertTrue(fixable, t["label"])
make_page.build(cfg)
make_page.build_restart(cfg)
self.assertIn("conn:x32", seen)
self.assertIn("dante:SC-X32", seen)
def test_a_big_campus_gets_group_keys_and_the_receivers_fold_first(self):
cfg, _ = ac.build(BASE, big(), {"id": "ab", "name": "Apollo Beach"}, "1111")
self.assertEqual(schema.validate(cfg)[0], [])
top = [layout_of(cfg).get((0, c)) for c in range(2, 8)]
self.assertEqual(top, ["REC", "AUDIO", "RF", "SHOW", "VIDEO", "STREAM"])
unplaced = [t["label"] for t in cfg["tiles"] if not t.get("pos")]
self.assertTrue(unplaced and all(u.startswith(("AXIENT", "PSM")) for u in unplaced), unplaced)
rf = next(t for t in cfg["tiles"] if t["label"] == "RF")
self.assertEqual(len(rf["members"]), 11)
self.assertEqual(layout_of(cfg)[(3, 7)], "NET")
self.assertEqual(layout_of(cfg)[(3, 0)], "ALARMS")
make_page.build(cfg)
class TestExtend(unittest.TestCase):
def setUp(self):
self.cfg, self.seen = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "4821")
def test_nothing_new_adds_nothing(self):
new, added, _ = ac.extend(self.cfg, small(), self.seen)
self.assertEqual(added, [])
self.assertEqual(new, self.cfg)
def test_new_gear_is_added_into_free_cells_and_merged_into_its_key(self):
inv = small()
inv["connections"].append({"label": "HyperDeck_3", "module": "bmd-hyperdeck", "host": "10.4.10.13", "enabled": True})
inv["dante"]["devices"].append("SC-Tracks")
inv["midi"][""]["SC Ingest"]["peers"].append({"name": "SuperRack", "address": "10.4.13.140:5004", "state": "running"})
new, added, now_seen = ac.extend(self.cfg, inv, self.seen)
self.assertEqual(sorted(added), ["Dante device SC-Tracks", "HD3 key", "MIDI link SuperRack"])
self.assertEqual(schema.validate(new)[0], [])
hd = [t for t in new["tiles"] if t["label"] == "HD3"][0]
self.assertTrue(hd.get("pos"))
self.assertIn("SC-Tracks", next(c for t in new["tiles"] for c in t.get("checks", []) if c["type"] == "dante")["devices"])
# Nothing that was there changed place.
self.assertTrue(set(layout_of(self.cfg).items()) <= set(layout_of(new).items()))
again, added2, _ = ac.extend(new, inv, set(self.seen) | set(now_seen))
self.assertEqual(added2, [])
def test_a_key_someone_took_off_is_not_added_back(self):
cfg = json.loads(json.dumps(self.cfg))
cfg["tiles"] = [t for t in cfg["tiles"] if t["label"] != "X32"]
_, added, _ = ac.extend(cfg, small(), self.seen)
self.assertNotIn("X32 key", added)
_, added, _ = ac.extend(cfg, small(), []) # never seen: it would be
self.assertIn("X32 key", added)
def test_north_attleboro_like_config_only_gets_what_it_does_not_watch(self):
cfg = json.load(open(os.path.join(ROOT, "config.json")))
inv = {"ingest": {"wired": {"net": "10.1.0.0/16"}},
"connections": [{"label": "HyperDeck_1", "module": "bmd-hyperdeck", "host": "10.1.10.11", "enabled": True},
{"label": "x32", "module": "behringer-x32", "host": "10.1.1.112", "enabled": True}],
"dante": {}, "midi": {}, "computers": {}}
_, added, _ = ac.extend(cfg, inv, [])
self.assertEqual(added, [])
class TestJoin(unittest.TestCase):
def test_a_join_report_is_checked(self):
e = discover.clean_hello({"user": "media", "name": "Lighting PC", "os": "Windows", "ips": ["10.4.10.206", "bad"],
"macs": ["AA-BB-CC-DD-EE-FF", "nope"], "tasks": ["Start Vista"]}, "10.4.10.206")
self.assertEqual((e["os"], e["ips"], e["macs"], e["tasks"]), ("windows", ["10.4.10.206"], ["aa:bb:cc:dd:ee:ff"], ["Start Vista"]))
for bad in ({"name": "x"}, {"user": "a;rm -rf /"}, {"user": ""}):
with self.assertRaises(ValueError):
discover.clean_hello(bad, "10.4.1.2")
self.assertEqual(discover.clean_hello({"user": "u", "name": "<b>Mac</b>"}, "10.4.1.2")["name"], "bMacb")
def test_the_lines_people_paste(self):
line = campus.install_script("http://10.1.1.1:8020", "ab; rm", "ssh-ed25519 AAAA na")
self.assertIn("zsh install.sh --site ab", line) # the id is cleaned
self.assertIn("--name 'Apollo Beach'", line)
self.assertIn("SUPPORT='ssh-ed25519 AAAA na'", line)
cfg = {"site": {"ingest_name": "SC Ingest", "ingest_ips": ["10.4.1.1"]}}
ps = campus.join_windows(cfg, "http://10.4.1.1:8020")
self.assertIn("'Vista.exe' = 'Vista'", ps)
self.assertIn("Invoke-RestMethod -Method Post -Uri 'http://10.4.1.1:8020/api/hello'", ps)
self.assertNotIn("{{", ps)
class TestRoutes(unittest.TestCase):
@classmethod
def setUpClass(cls):
snap = healthy_snapshot(time.time())
FakeCompanion.store = {"conns": snap.conns, "vars": dict(snap.vars), "custom": {}, "restarts": []}
cls.fake = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion)
threading.Thread(target=cls.fake.serve_forever, daemon=True).start()
cls.data = tempfile.mkdtemp(prefix="sp-campus-")
key = os.path.join(cls.data, "id_test")
open(key + ".pub", "w").write("ssh-ed25519 AAAAC3NzaTESTKEY status-panel@sc\n")
cfg, _ = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "4821")
cfg["companion"] = {"host": "127.0.0.1", "port": cls.fake.server_address[1]}
cfg["ssh_key"] = key
json.dump(cfg, open(os.path.join(cls.data, "config.json"), "w"))
json.dump({"pid": 1, "started_at": time.time(), "config_sha": "x", "version": "test"}, open(os.path.join(cls.data, "started.json"), "w"))
open(os.path.join(cls.data, "status_panel.py"), "w").write("# code\n")
os.makedirs(os.path.join(cls.data, "kit", "sites", "_template"))
open(os.path.join(cls.data, "kit", "install.sh"), "w").write("# installer\n")
open(os.path.join(cls.data, "kit", "sites", "_template", "config.json"), "w").write("{}")
open(os.path.join(cls.data, "outlets.json"), "w").write('{"tuya": {"access_secret": "never-shipped"}}')
sock = socket.socket(); sock.bind(("127.0.0.1", 0)); cls.port = sock.getsockname()[1]; sock.close()
cls.site, cls.servers = server.serve(cls.data, cls.port, ["127.0.0.1"])
@classmethod
def tearDownClass(cls):
for s in cls.servers:
s.shutdown()
cls.fake.shutdown()
def test_hello_registers_a_computer_without_the_code(self):
status, body = get(self.port, "/api/hello", {"user": "media", "name": "Tracks Mac", "os": "mac", "ips": ["10.4.3.227"]})
self.assertEqual(status, 200)
self.assertIn("Tracks Mac", discover.registered(self.data))
self.assertTrue(self.site.discover_poke.is_set())
self.assertEqual(get(self.port, "/api/hello", {"name": "no user"})[0], 400)
status, body = get(self.port, "/api/discover")
d = json.loads(body)
self.assertEqual(d["join"]["mac"], "curl -fsSL http://127.0.0.1:%d/join | sh" % self.port)
self.assertIn("Tracks Mac", d["joined"])
def test_join_lines_carry_this_macs_key_and_address(self):
status, body = get(self.port, "/join")
self.assertEqual(status, 200)
self.assertIn("KEY='ssh-ed25519 AAAAC3NzaTESTKEY status-panel@sc'", body.decode())
self.assertIn("http://127.0.0.1:%d/api/hello" % self.port, body.decode())
status, body = get(self.port, "/join.ps1")
self.assertEqual(status, 200)
self.assertIn("$IngestIPs = @('10.4.1.1', '10.6.20.5')", body.decode())
def test_the_program_kit_has_the_code_and_never_the_settings(self):
status, body = get(self.port, "/campus/program.tgz")
self.assertEqual(status, 200)
names = tarfile.open(fileobj=io.BytesIO(body), mode="r:gz").getnames()
self.assertIn("status_panel.py", names)
self.assertIn("install.sh", names)
self.assertIn("sites/_template/config.json", names)
self.assertFalse([n for n in names if n.endswith(("config.json",)) and not n.startswith("sites/_template")])
self.assertNotIn("outlets.json", names)
status, body = get(self.port, "/campus/install?id=ws")
self.assertIn("--site ws", body.decode())
self.assertIn("SUPPORT='ssh-ed25519 AAAAC3NzaTESTKEY", body.decode()) # this campus can help the new one
def test_look_now_needs_the_code(self):
self.assertEqual(get(self.port, "/api/discover/now", {})[0], 401)
if __name__ == "__main__":
unittest.main()
class TestCampusesView(unittest.TestCase):
"""North Attleboro alone shows every campus: a copy of each status page (view only) and a CAMPUSES page."""
def test_the_campus_page_is_view_only_cells_with_a_way_back(self):
from settings import campuses as cv
import push_page
na = json.load(open(os.path.join(ROOT, "config.json")))
export = cv.build_view(na, 99, 98)
ctrls = export["page"]["controls"]
self.assertEqual(export["page"]["name"], "99\\nCampus")
self.assertIn("CAMPUSES", json.dumps(ctrls["3"]["0"])) # where ALARMS sits on every status page
self.assertIn("$(custom:sp_view)", json.dumps(ctrls["3"]["0"]["style"])) # it says which campus is shown
cells = [ctl for row in ctrls.values() for ctl in row.values() if "campus view" in (ctl.get("options") or {}).get("notes", "")]
self.assertEqual(len(cells), 28)
self.assertTrue(all(c["steps"]["0"]["action_sets"] == {"down": [], "up": []} for c in cells)) # no press, no hold
self.assertIn("sp_state_view", json.dumps(cells[0]["style"]))
self.assertTrue(push_page.safe_target(export)[0]) # a rebuild may overwrite it
def test_a_campus_state_lands_on_its_cells(self):
from settings import campuses as cv
ws, _ = ac.build(BASE, small(), {"id": "ws", "name": "Woonsocket"}, "1234")
keys = [{"id": t["id"], "pos": t["pos"]} for t in ws["tiles"] if t.get("pos")]
out = cv.view_state({"hd1": {"x": "HD1\nREC", "b": 1}, "all": {"x": "ALL OK"}}, keys)
hd1 = next(k for k in keys if k["id"] == "hd1")
self.assertEqual(out[cv.cell_id(*hd1["pos"])]["x"], "HD1\nREC")
self.assertEqual(out["c_0_1"]["x"], "ALL OK")
self.assertEqual(out["c_2_7"]["x"], "") # nothing there: blank
def test_the_campuses_page_puts_this_campus_first_and_taps_to_each(self):
from settings import campuses as cv
na = json.load(open(os.path.join(ROOT, "config.json")))
roster = {"na": {"name": "North Attleboro"}, "ws": {"name": "Woonsocket"}, "ab": {"name": "Apollo Beach"}}
export = cv.build_overview(na, roster, 98, 99)
keys = [(r, c, ctl) for r, row in export["page"]["controls"].items() for c, ctl in row.items() if "'campus " in (ctl.get("options") or {}).get("notes", "")]
self.assertEqual([(r, c) for r, c, _ in keys], [("0", "1"), ("0", "2"), ("0", "3")])
self.assertIn("'campus na'", keys[0][2]["options"]["notes"])
self.assertIn("page 19", keys[0][2]["options"]["notes"])
ws_key = next(k for _, _, k in keys if "'campus ws'" in k["options"]["notes"])
acts = ws_key["steps"]["0"]["action_sets"]["up"]
self.assertEqual([a["definitionId"] for a in acts], ["custom_variable_set_value", "set_page"])
self.assertEqual((acts[0]["options"]["value"]["value"], acts[1]["options"]["page"]["value"]), ("WS", 99))
def test_a_campus_tells_the_hub_and_learns_nothing_back(self):
from settings import campuses as cv
data_dir = tempfile.mkdtemp(prefix="sp-cv-")
cv.save(data_dir, {"home": "http://10.1.1.1:8020", "list": {"na": {"url": "x"}, "ab": {"url": "y"}}})
class FakeSite:
def __init__(self):
self.data_dir = data_dir
def load(self):
return {"site": {"id": "ws", "name": "Woonsocket"}}
sent = []
real = cv.http_json
cv.http_json = lambda url, body=None, timeout=3: sent.append((url, body)) or {"ok": True}
real_url = cv.my_url
cv.my_url = lambda cfg: "http://10.5.1.1:8020"
try:
m = cv.Mirror(FakeSite())
data = m.sync_registry()
m.poll()
self.assertEqual(m.refresh_pages()["list"], {"ws": {"id": "ws", "name": "Woonsocket", "url": "http://10.5.1.1:8020"}})
finally:
cv.http_json, cv.my_url = real, real_url
self.assertEqual(list(data["list"]), ["ws"]) # the others are forgotten
self.assertTrue(all(u.endswith("/api/campus/hello") for u, _ in sent)) # it only ever says hello
class TestCampusHello(TestRoutes):
def test_campus_hello_is_checked(self):
self.assertEqual(get(self.port, "/api/campus/hello", {"id": "ws", "name": "Woonsocket", "url": "http://10.5.1.1:8020"})[0], 200)
for bad in ({"id": "WS; rm", "url": "http://10.5.1.1:8020"}, {"id": "ws", "url": "http://evil.example.com:80"},
{"id": "ws", "url": "http://8.8.8.8:8020"}):
self.assertEqual(get(self.port, "/api/campus/hello", bad)[0], 400, bad)
from settings import campuses as cv
self.assertEqual(cv.load(self.data)["list"]["ws"]["url"], "http://10.5.1.1:8020")
get(self.port, "/api/campus/hello", {"id": "new", "url": "http://10.5.1.9:8020"})
get(self.port, "/api/campus/hello", {"id": "sc", "name": "Southcoast", "url": "http://10.5.1.9:8020"})
self.assertNotIn("new", cv.load(self.data)["list"]) # same Mac, now named: the old entry goes