"""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_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_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": "Mac"}, "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")]) 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