christianandClaude Opus 5.5
3976667bba
settings/discover.py reads Companion's connections (and their addresses and channel counts), the Dante devices, the network MIDI sessions here and on joined Macs, this Mac's wired network, and what runs on each joined computer. settings/autoconf.py turns that into keys, checks, start and restart fixes and a page layout the way North Attleboro's were made (group keys only when not everything fits; receivers fold first), all through the existing builders and rules. settings/campus.py prepares Companion through its internal API (the panel's custom variables and the blink trigger), picks free pages, applies the first config, serves the program kit, the campus install line and the Mac/PC join lines, and looks every ten minutes to add new gear (discover.auto; never during a service or a recording, never removes, never re-adds what someone took off). Settings > Auto-setup shows what was found and the lines to paste. install.sh keeps the kit and makes the panel's login key. Checked on North Attleboro read-only: discovery finds all 43 connections, 19 Dante devices, four Macs' MIDI sessions and seven computers, and a from-scratch build reproduces NA's layout. The Companion calls were tried with a throwaway variable and a disabled trigger, then removed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
248 lines
15 KiB
Python
248 lines
15 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()
|