christianandClaude Opus 5.5
d7dc931bd4
Campuses set up from NA's install line tell NA where they are and learn nothing back (no campus ever shows another). NA alone reads each campus's key states every two seconds and keeps two pages in its Companion (added at the end, 98 and 99, since every other page is in use): CAMPUSES, a key per campus (its ALL key, purple when it stops answering), and CAMPUS, the tapped campus's status page drawn live in the same places, view only. Tried on NA with a test campus pointing at NA itself, then removed. tools/fleet.py: an Apple Remote Desktop command that turns on Remote Login, adds the laptop's key and reports each Mac to NA, and the laptop SSH shortcuts made from those reports (tried on NA Translation). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
326 lines
20 KiB
Python
326 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
|
|
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")
|