christianandClaude Opus 5.5
6d1cfea2f7
- Auto-setup: programs on the Companion Mac itself (ProPresenter at Woonsocket) get a local process check and local start/restart fixes; every campus gets a Restart page (keys arrive as computers join); NA's auto-fix hours (Tue, Wed, Sun); a program found after its Companion connection joins that key instead of making a second one. settings.campus rebuild remakes a campus from what is found now, keeping its name, PIN and pages; upgrade-companion brings Companion to 5.0.7 from NA's copy (/campus/companion.zip), as Woonsocket was; setup stops with that advice on an older one. - NA's CAMPUSES page has a RESTART key per campus; the CAMPUS page shows that campus's Restart page and a hold there restarts it at that campus. - Backups: backup/backup-campuses.sh, nightly 04:30 on NA (installed everywhere, acts on the hub), into Drive > Waters Creative > Apps > Status Panel Backups/<campus>, 30 kept, alert email via the checklist's alert.js. Campuses report the account to back up with. sites/ws/config.json in the repo. - Woonsocket rebuilt: 9 keys (PROP with ProPresenter on its Mac), Restart page 20 (SUPERRACK, VISTA, ABLETON, PROP), auto-fix on NA's hours; all OK. Both campuses backed up to Drive by the job. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
437 lines
29 KiB
Python
437 lines
29 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_programs_app_comes_from_its_running_path(self):
|
|
"""Woonsocket's FOH Mac: SuperRack lives three folders down, and zsh had hidden the app list."""
|
|
inv = small()
|
|
inv["computers"]["foh"] = {"name": "WS FOH", "os": "mac", "user": "watersmedia", "ip": "10.4.1.12", "ips": ["10.4.1.12"], "reachable": True,
|
|
"procs": ["SuperRack SoundGrid", "X32-Edit"], "apps": [],
|
|
"paths": {"SuperRack SoundGrid": "/Applications/Waves/SuperRack SoundGrid/SuperRack SoundGrid.app",
|
|
"X32-Edit": "/Applications/X32-Edit.app"}}
|
|
cfg, _ = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
|
|
self.assertEqual(schema.validate(cfg)[0], [])
|
|
sr = next(t for t in cfg["tiles"] if t["label"] == "SUPERRACK")
|
|
self.assertEqual(sr["checks"][0]["cmd"], "pgrep -xq 'SuperRack SoundGrid'")
|
|
self.assertIn("/Applications/Waves/SuperRack SoundGrid/SuperRack SoundGrid.app", cfg["fixes"][sr["checks"][0]["fix"]]["run"])
|
|
self.assertIn("X32 EDIT", [t["label"] for t in cfg["tiles"]])
|
|
self.assertEqual([c.get("msg") for c in sr["checks"]], ["SUPERRACK NOT RUNNING", "SOUNDGRID LOST"])
|
|
self.assertIn(":19000 (ESTABLISHED)", sr["checks"][1]["cmd"])
|
|
inv["computers"]["lights"] = {"name": "WS Lights", "os": "mac", "user": "wslighting", "ip": "10.4.2.2", "ips": ["10.4.2.2"], "reachable": True,
|
|
"procs": ["Vista", "VistaPlaybackNode"], "apps": ["/Applications/Vista.app"], "paths": {"Vista": "/Applications/Vista.app"}}
|
|
cfg, _ = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
|
|
vista = [t for t in cfg["tiles"] if t["label"] == "VISTA" and t["checks"][0].get("host") == "lights"]
|
|
self.assertEqual(vista[0]["checks"][0]["cmd"], "pgrep -xq Vista") # Vista on a Mac, too
|
|
self.assertIn("SUPERRACK", [k["label"] for k in cfg["restart_page"]["keys"]])
|
|
|
|
def test_links_by_name_or_ipv6_get_the_wired_address_of_the_mac_whose_session_it_is(self):
|
|
"""Woonsocket: Tracks lists "Lighting" by name and the FOH Mac by an IPv6 address."""
|
|
inv = small()
|
|
inv["computers"]["tracks"] = {"name": "WS Tracks", "os": "mac", "user": "t", "ip": "10.4.1.219", "ips": ["169.254.45.114", "10.4.1.219"], "reachable": True, "procs": []}
|
|
inv["computers"]["lights"] = {"name": "WS Lights", "os": "mac", "user": "l", "ip": "10.4.2.2", "ips": ["10.4.2.2"], "reachable": True, "procs": []}
|
|
inv["computers"]["foh"] = {"name": "WS FOH", "os": "mac", "user": "f", "ip": "10.4.1.12", "ips": ["10.4.1.12"], "reachable": True, "procs": []}
|
|
inv["midi"] = {"": {}, "tracks": {"WS Tracks": {"port": 5004, "peers": [{"name": "Lighting", "address": "\tLighting\tlocal."},
|
|
{"name": "WS FOH Mac mini", "address": "[fe80::4c3:64d1:835f:f61f]:5004"}]}},
|
|
"lights": {"Lighting": {"port": 5006, "peers": [{"name": "WS Tracks", "address": "[fe80::188d]:5004", "state": "running"}]}},
|
|
"foh": {"WS FOH Mac mini": {"port": 5004, "peers": [{"name": "WS Tracks", "address": "\tWS Tracks\tlocal."}]}}}
|
|
cfg, _ = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
|
|
midi = next(t for t in cfg["tiles"] if t["label"] == "MIDI")
|
|
got = sorted((c.get("host"), c["peer"], c["address"]) for c in midi["checks"] if c["type"] == "midi_peer")
|
|
# One check per link: Tracks-Lights from Lights (it reports the state), Tracks-FOH from one end only.
|
|
self.assertEqual(got, [("lights", "WS Tracks", "10.4.1.219:5004"), ("tracks", "WS FOH Mac mini", "10.4.1.12:5004")])
|
|
self.assertEqual(schema.validate(cfg)[0], [])
|
|
import status_panel as sp
|
|
snap = sp.Snapshot()
|
|
snap.remote_midi["tracks"], snap.remote_midi_time["tracks"] = inv["midi"]["tracks"], time.time()
|
|
chk = {"type": "midi_peer", "host": "tracks", "session": "WS Tracks", "peer": "Lighting"}
|
|
self.assertEqual(sp.check_midi_peer(chk, snap, {"now": time.time()})[0][1].level, sp.OK) # no state: older macOS, configured
|
|
|
|
def test_a_program_on_the_companion_mac_gets_a_local_check_and_fixes(self):
|
|
"""Woonsocket: ProPresenter runs on the Mac that runs Companion."""
|
|
inv = small()
|
|
inv["computers"] = {}
|
|
exe = "/Applications/ProPresenter.app/Contents/MacOS/ProPresenter"
|
|
inv["ingest"].update(procs=["ProPresenter"], paths={"ProPresenter": "/Applications/ProPresenter.app"}, exes={"ProPresenter": exe})
|
|
inv["connections"] = [dict(c, host="10.4.1.1") if c["label"] == "ProP" else c for c in inv["connections"]]
|
|
cfg, _ = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
|
|
self.assertEqual(schema.validate(cfg)[0], [])
|
|
prop = next(t for t in cfg["tiles"] if t["label"] == "PROP")
|
|
self.assertEqual([(c["type"], c.get("match") or c.get("conn")) for c in prop["checks"]], [("conn", "ProP"), ("process", exe)])
|
|
fix = cfg["fixes"][prop["checks"][1]["fix"]]
|
|
self.assertEqual((fix["kind"], fix["run"]), ("local", "pgrep -xq ProPresenter || open /Applications/ProPresenter.app"))
|
|
self.assertIn(prop["checks"][1]["fix"], cfg["auto_fix"]["fixes"])
|
|
self.assertEqual([k["label"] for k in cfg["restart_page"]["keys"]], ["PROP"])
|
|
self.assertEqual(cfg["fixes"][cfg["restart_page"]["keys"][0]["fix"]]["kind"], "local")
|
|
|
|
def test_every_campus_gets_a_restart_page_and_north_attleboros_auto_fix_hours(self):
|
|
inv = small()
|
|
inv["computers"] = {}
|
|
cfg, _ = ac.build(BASE, inv, {"id": "sc", "name": "Southcoast"}, "1234")
|
|
self.assertEqual(cfg["restart_page"]["keys"], []) # computers join later: their keys come then
|
|
na = json.load(open(os.path.join(ROOT, "config.json")))
|
|
self.assertEqual(cfg["auto_fix"]["windows"], na["auto_fix"]["windows"])
|
|
self.assertEqual(cfg["service_windows"], na["service_windows"])
|
|
self.assertEqual(cfg["nav_keys"], na["nav_keys"])
|
|
make_page.build_restart(cfg)
|
|
|
|
def test_an_m32_is_called_m32(self):
|
|
inv = small()
|
|
inv["connections"] = [c for c in inv["connections"] if c["label"] != "x32"] + [
|
|
{"label": "m32", "module": "behringer-x32", "host": "10.4.1.146", "enabled": True}]
|
|
cfg, _ = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
|
|
self.assertIn("M32", [t["label"] for t in cfg["tiles"]])
|
|
|
|
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_program_found_later_joins_its_connections_key(self):
|
|
"""Woonsocket got a second PROP key when ProPresenter on the Companion Mac was found after the connection."""
|
|
inv = small()
|
|
inv["computers"] = {}
|
|
cfg, seen = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
|
|
exe = "/Applications/ProPresenter.app/Contents/MacOS/ProPresenter"
|
|
inv["ingest"].update(procs=["ProPresenter"], paths={"ProPresenter": "/Applications/ProPresenter.app"}, exes={"ProPresenter": exe})
|
|
inv["connections"] = [dict(c, host="127.0.0.1") if c["label"] == "ProP" else c for c in inv["connections"]]
|
|
new, added, _ = ac.extend(cfg, inv, seen)
|
|
self.assertEqual([t["label"] for t in new["tiles"]].count("PROP"), 1)
|
|
prop = next(t for t in new["tiles"] if t["label"] == "PROP")
|
|
self.assertEqual([c["type"] for c in prop["checks"]], ["conn", "process"])
|
|
self.assertIn("PROP checks on the PROP key", added)
|
|
self.assertEqual(schema.validate(new)[0], [])
|
|
|
|
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)
|
|
acts = cells[0]["steps"]["0"]["action_sets"]
|
|
self.assertEqual(sorted(acts), ["2000", "down", "up"]) # tap = acknowledge, hold = fix, there
|
|
self.assertIn("sp_view_cmd", json.dumps(acts))
|
|
self.assertEqual(cv.campus_command([{"id": "hd1", "pos": [0, 2]}, {"id": "midi", "pos": [1, 3]}],
|
|
"tap 0 2 1000;hold 1 3 1000;hold 3 3 1000;hold 1 3 900", 1005),
|
|
["ack hd1", "restart midi 1000"]) # an empty cell and a stale hold are dropped
|
|
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")])
|
|
rs = export["page"]["controls"]["1"]["3"] # WS's RESTART key, under its ALL key
|
|
acts = rs["steps"]["0"]["action_sets"]["up"]
|
|
self.assertEqual([a["options"].get("value", {}).get("value") for a in acts[:2]] + [acts[2]["options"]["page"]["value"]], ["WS", "RESTART", 99])
|
|
self.assertEqual(cv.campus_command([{"id": "rs_prop", "pos": [0, 1]}], "tap 0 1 1000;hold 0 1 1000", 1001, restart=True),
|
|
["force rs_prop 1000"]) # on its Restart page a hold forces, a tap does nothing
|
|
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", "custom_variable_set_value", "set_page"])
|
|
self.assertEqual((acts[0]["options"]["value"]["value"], acts[1]["options"]["value"]["value"], acts[2]["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()
|
|
listed = m.refresh_pages()["list"]
|
|
self.assertEqual({k: {f: v for f, v in e.items() if f != "user"} for k, e in listed.items()},
|
|
{"ws": {"id": "ws", "name": "Woonsocket", "url": "http://10.5.1.1:8020"}})
|
|
self.assertTrue(listed["ws"]["user"]) # the account the hub backs it up with
|
|
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
|