"""Auto-setup for the other locations: an inventory becomes keys that pass every rule, laid out like
North Attleboro's; a running location 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")
        inv["midi"][""] = {"ProPresenter": {"port": 5004, "peers": [{"name": "Lighting", "address": "10.4.2.2:5006", "state": "running"}]}}
        inv["midi"]["lights"]["Lighting"]["peers"].append({"name": "ProPresenter", "address": "10.4.1.1:5004", "state": "running"})
        cfg, _ = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
        midi = next(t for t in cfg["tiles"] if t["label"] == "MIDI")
        pairs = [(c.get("host"), c["peer"]) for c in midi["checks"] if c["type"] == "midi_peer"]
        self.assertEqual(len([p for p in pairs if "ProPresenter" in p or ("lights" not in p and "Lighting" in p and None in p)]), 1)   # the Companion Mac's link counted once
        inv["midi"][""] = {}
        inv["midi"]["lights"]["Lighting"]["peers"].pop()
        cfg, _ = ac.build(BASE, inv, {"id": "ws", "name": "Woonsocket"}, "1234")
        midi = next(t for t in cfg["tiles"] if t["label"] == "MIDI")
        # 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")
        import re
        self.assertEqual([c["type"] for c in prop["checks"]], ["conn", "process"])
        match = prop["checks"][1]["match"]                                   # any installed copy counts (an update beside the old)
        self.assertTrue(re.search(match, exe) and re.search(match, "/Applications/ProPresenter 21.4.app/Contents/MacOS/ProPresenter"))
        fix = cfg["fixes"][prop["checks"][1]["fix"]]
        self.assertEqual(fix["kind"], "local")
        self.assertTrue(fix["run"].startswith("pgrep -xq ProPresenter || { APP=$(for a in /Applications/ProPresenter*.app"), fix["run"])
        self.assertTrue(fix["run"].endswith('open "$APP"; }'), fix["run"])                # the newest version opens
        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_rebuild_keeps_what_is_down_right_now(self):
        """Southcoast's rebuild dropped its MIDI key: the ProPresenter-to-Lighting link was down at that moment."""
        old, _ = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "1234")
        now = small()
        now["midi"] = {"": {"SC Ingest": {"port": 5004, "peers": []}}}                  # the link dropped
        now["computers"]["lighting_pc"]["reachable"] = False                             # and the Lighting PC is off
        new, _ = ac.build(BASE, now, {"id": "sc", "name": "Southcoast"}, "1234")
        self.assertNotIn("MIDI", [t["label"] for t in new["tiles"]])
        kept, words = ac.keep_from(new, old)
        self.assertIn("MIDI", [t["label"] for t in kept["tiles"]])
        self.assertIn("VISTA", [t["label"] for t in kept["tiles"]])
        self.assertIn("lighting_pc", kept["hosts"])
        self.assertIn("MIDI link Lighting PC", words)
        self.assertEqual(schema.validate(kept)[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 location can help the new one
        status, body = get(self.port, "/location/install?id=ws")                  # the address people see now
        self.assertEqual(status, 200)
        self.assertIn("/location/program.tgz", body.decode())
        self.assertIn("settings.location setup", body.decode())

    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 location: a copy of each status page (view only) and a LOCATIONS 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\\nLocation")
        self.assertIn("LOCATIONS", 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 location is shown
        cells = [ctl for row in ctrls.values() for ctl in row.values() if "location 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 "'location " 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("'location na'", keys[0][2]["options"]["notes"])
        self.assertIn("page 19", keys[0][2]["options"]["notes"])
        ws_key = next(k for _, _, k in keys if "'location 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


class TestStandalone(unittest.TestCase):
    """Apollo Beach runs its own panel (2026-10-21): it tells nobody, keeps no location list, backs itself up."""

    def test_a_standalone_campus_tells_nobody_and_is_no_hub(self):
        from settings import campuses as cv
        data_dir = tempfile.mkdtemp(prefix="sp-sa-")
        cv.save(data_dir, {"standalone": True})
        self.assertFalse(cv.Mirror.is_hub(cv.load(data_dir)))
        self.assertTrue(cv.Mirror.is_hub({}))                                         # NA: no home
        self.assertFalse(cv.Mirror.is_hub({"home": "http://10.1.1.1:8020"}))

        class FakeSite:
            def __init__(self):
                self.data_dir = data_dir

            def load(self):
                return {"site": {"id": "ab", "name": "Apollo Beach"}}
        sent = []
        real, real_url = cv.http_json, cv.my_url
        cv.http_json = lambda url, body=None, timeout=3: sent.append(url) or {"ok": True}
        cv.my_url = lambda cfg: "http://10.4.1.1:8020"
        try:
            m = cv.Mirror(FakeSite())
            data = m.sync_registry()
            m.poll()
            m.refresh_pages()
        finally:
            cv.http_json, cv.my_url = real, real_url
        self.assertEqual(sent, [])
        self.assertEqual(list(data["list"]), ["ab"])

    def test_setup_marks_it_standalone_and_drops_any_home(self):
        from settings import campuses as cv
        data_dir = tempfile.mkdtemp(prefix="sp-sa-")
        cv.save(data_dir, {"home": "http://10.1.1.1:8020"})
        real = campus.is_fresh
        campus.is_fresh = lambda cfg: False                                            # set up already: only the look runs
        real_look = campus.look_and_add
        campus.look_and_add = lambda site, say=print, force=False: None
        real_site = server.Site
        server.Site = lambda d: type("S", (), {"load": lambda self: {}})()
        try:
            campus.setup(data_dir, "ab", "Apollo Beach", say=lambda *a: None, standalone=True)
        finally:
            campus.is_fresh, campus.look_and_add, server.Site = real, real_look, real_site
        data = cv.load(data_dir)
        self.assertTrue(data.get("standalone"))
        self.assertNotIn("home", data)

    def test_the_standalone_zip_has_the_program_an_installer_and_no_settings(self):
        import zipfile
        sys.path.insert(0, os.path.join(ROOT, "tools"))
        import standalone
        out = tempfile.mkdtemp(prefix="sp-zip-")
        path = standalone.build("ab", "Apollo Beach", out)
        with zipfile.ZipFile(path) as z:
            names = z.namelist()
            inst = z.read("Apollo Beach Status Panel/install.command").decode()
            mode = z.getinfo("Apollo Beach Status Panel/install.command").external_attr >> 16
        top = "Apollo Beach Status Panel/program/"
        for f in ("status_panel.py", "install.sh", "settings/campus.py", "sites/_template/config.json", "backup/backup-campuses.sh"):
            self.assertIn(top + f, names)
        for never in ("config.json", "outlets.json", "campuses.json", "sites/ws/config.json", "sites/sc/config.json"):
            self.assertNotIn(top + never, names)
        self.assertFalse([n for n in names if "__pycache__" in n or "/._" in n])
        self.assertIn("Apollo Beach Status Panel/READ ME FIRST.txt", names)
        self.assertTrue(mode & 0o100)
        self.assertIn("--standalone", inst)
        self.assertNotIn("10.1.1.1", inst)
        self.assertNotIn("authorized_keys", inst)                                     # no one else's key is let in


class TestSwitchKeys(unittest.TestCase):
    def test_a_new_campus_gets_auto_fix_by_alarms_and_a_rebuild_keeps_its_own_switches(self):
        new, _ = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "1234")
        af = [t for t in new["tiles"] if t.get("kind") == "autofix"]
        self.assertEqual([t["pos"] for t in af], [[3, 1]])                            # beside ALARMS
        self.assertEqual(schema.validate(new)[0], [])
        old = json.loads(json.dumps(new))
        old["tiles"].append({"id": "lights", "label": "LIGHTS", "kind": "bypass", "pos": [2, 7], "bypass": {"peers": ["Lighting"]}})
        again, _ = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "1234")
        kept_cfg, kept = ac.keep_from(again, old)
        lights = [t for t in kept_cfg["tiles"] if t["id"] == "lights"]
        self.assertEqual((len(lights), lights[0]["pos"], "LIGHTS" in kept), (1, [2, 7], True))
        self.assertEqual(len([t for t in kept_cfg["tiles"] if t.get("kind") == "autofix"]), 1)


class TestJoinWindowsAdmin(unittest.TestCase):
    def test_the_pc_line_stops_with_plain_words_when_not_administrator(self):
        cfg = json.load(open(os.path.join(ROOT, "sites", "sc", "config.json")))
        text = campus.join_windows(cfg, "http://10.5.2.60:8020")
        check = text.index("IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator)")
        self.assertLess(check, text.index("Get-WindowsCapability"))                # before anything needs it
        self.assertIn("Run as administrator", text)


class TestHelperJoinsLater(unittest.TestCase):
    def test_rtpmidi_on_a_pc_that_joins_later_lands_on_the_midi_key(self):
        inv = small()
        inv["computers"].pop("lighting_pc")
        cfg, seen = ac.build(BASE, inv, {"id": "sc", "name": "Southcoast"}, "1234")
        midi = next((t for t in cfg["tiles"] if any(c.get("type") == "midi_wired" for c in t.get("checks", []))), None)
        self.assertIsNotNone(midi)                                                   # the MIDI key is there first
        new, added, now_seen = ac.extend(cfg, small(), seen)
        midi = next(t for t in new["tiles"] if any(c.get("type") == "midi_wired" for c in t.get("checks", [])))
        self.assertTrue(any((c.get("params") or {}).get("exe") == "rtpMIDI.exe" for c in midi["checks"]), added)
        self.assertTrue(any("RTPMIDI" in a for a in added), added)
        self.assertEqual(schema.validate(new)[0], [])


class TestLocationsBand(unittest.TestCase):
    """NA's page keys name the other locations with a problem in blue; red stays NA's own (2026-10-21)."""

    def test_the_summary_names_locations_down_and_flashes_until_acknowledged(self):
        from settings import campuses as cv
        now = 1000.0
        st = lambda x, b, b2: {"hb": now - 1, "state": json.dumps({"all": {"x": x, "b": b, "b2": b2}})}
        remote = {"ws": st("ALL OK", 1, 1), "sc": st("2 DOWN\nPROP LIGHTS", 0xFF0000, 0)}
        self.assertEqual(cv.remote_summary(["sc", "ws", "ab"], remote, now), {"t": 1000, "n": 2, "x": "AB? SC", "f": 1})
        remote["sc"] = st("1 DOWN\nX", 0x8B0000, 0x8B0000)                                        # acknowledged at SC: steady
        self.assertEqual(cv.remote_summary(["sc", "ws"], remote, now), {"t": 1000, "n": 1, "x": "SC", "f": 0})
        remote["sc"] = dict(remote["sc"], hb=now - 60)                                            # stopped answering
        self.assertEqual(cv.remote_summary(["sc"], remote, now)["x"], "SC?")

    def test_only_a_page_key_set_for_it_carries_the_band(self):
        na = json.load(open(os.path.join(ROOT, "config.json")))
        names = lambda cfg: [l.get("name") for l in make_page.nav_key("pagenum", cfg["variables"], 19, cfg=cfg)["style"]["layers"]]
        self.assertIn("Locations", names(dict(na, nav_locations_band=True)))
        self.assertNotIn("Locations", names(dict(na, nav_locations_band=False)))
        band = [l for l in make_page.nav_key("pagenum", na["variables"], 19, cfg=dict(na, nav_locations_band=True))["style"]["layers"]
                if l.get("name") == "Locations"][0]
        self.assertNotIn(str(0xFF0000), json.dumps(band))                                         # never red


class TestAllAutoFix(unittest.TestCase):
    def test_the_key_names_the_paused_locations_and_sits_where_auto_fix_does(self):
        from settings import campuses as cv
        st = lambda x: json.dumps({"autofix": {"x": x}})
        states = {"na": st("AUTO-FIX\nON\nTIL 9PM"), "sc": st("AUTO-FIX\nOFF\nTIL SUN 5AM"), "ws": st("AUTO-FIX\nOFF"), "ab": None}
        self.assertEqual(cv.all_autofix_summary(states, 100.0), {"t": 100, "paused": 1, "x": "SC\nPAUSED"})   # WS: outside its hours
        self.assertEqual(cv.all_autofix_summary({"na": st("AUTO-FIX\nON\nTIL 9PM")}, 100.0)["x"], "ALL ON")
        na = json.load(open(os.path.join(ROOT, "config.json")))
        page = cv.build_overview(na, {"na": {"name": "North Attleboro"}, "sc": {"name": "Southcoast"}}, 98, 99)
        key = page["page"]["controls"]["3"]["1"]
        self.assertIn("all autofix", key["options"]["notes"])
        self.assertIn("sp_all_cmd", json.dumps(key["steps"]))