christianandClaude Opus 5.5
93805cb723
On a Mac without the Command Line Tools, /usr/bin/python3 is a stub that opens Apple's installer on that Mac's screen: Woonsocket's Tracks and FOH Macs showed it after joining (and would every ten minutes). Python now runs on another Mac only when xcode-select says it is there (MIDI reads, MIDI reconnects, the Computers probe); without it those Macs' own MIDI sessions are simply not read. The app list used a glob that zsh (the Macs' login shell) turns into an error when one pattern matches nothing, so no app was ever found there: find instead, and a running program's app path is taken from the program itself. New recipes: SuperRack SoundGrid and X32-Edit (AUDIO). Woonsocket now has ABLETON, SUPERRACK and X32 EDIT keys, all OK, and NA's page 99 shows them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
345 lines
21 KiB
Python
345 lines
21 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.assertIn("SUPERRACK", [k["label"] for k in cfg["restart_page"]["keys"]])
|
|
|
|
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
|