status-indicators/tests/test_status_panel.py

"""Offline tests: no Companion, no gear. Run:  python3 -m unittest discover -s tests"""
import gzip
import json
import logging
import os
import re
import sys
import tempfile
import threading
import time
import unittest
import urllib.parse
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)

import make_page  # noqa: E402
import status_panel as sp  # noqa: E402

sp.log.addHandler(logging.NullHandler())
sp.log.propagate = False

CONFIG = sp.load_config(os.path.join(ROOT, "config.json"))
# The tests pin their own schedule so they do not depend on the site's choice.
CONFIG["service_windows"] = [{"days": ["sun"], "start": "07:00", "end": "12:30"}]
CONFIG["tiles"] = [dict(t, label="SERVICE") if t["id"] == "service" else t for t in CONFIG["tiles"]]
# A Sunday 08:00 (inside the service window) and a Wednesday 15:00 (outside it).
SUNDAY = datetime(2026, 10, 11, 8, 0).timestamp()
WEDNESDAY = datetime(2026, 10, 7, 15, 0).timestamp()


def healthy_snapshot(now):
    """Everything in config.json present and fine."""
    snap = sp.Snapshot()
    labels = {c["conn"] for t in CONFIG["tiles"] for c in t.get("checks", []) if c.get("conn")}
    snap.conns = {l: {"id": "id-" + l, "enabled": True, "category": "good", "message": None} for l in labels}
    snap.conns_time = now
    for i in range(1, 7):
        snap.vars[("HyperDeck_%d" % i, "status")] = "Preview"
        snap.vars[("HyperDeck_%d" % i, "slotId")] = "1"
        snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "28:00:00"
    devices = set()
    for t in CONFIG["tiles"]:
        for c in t.get("checks", []):
            if c["type"] == "dante":
                devices |= set(c["devices"])
    snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(sorted(devices))
    snap.vars[("tascam", "error")] = "No Error"
    snap.vars[("tascam", "caution")] = "No Caution"
    for label, n in (("Axient_1", 4), ("Axient_2", 4), ("ULXD-4", 1)):
        for ch in range(1, n + 1):
            snap.vars[(label, "ch_%d_name" % ch)] = "VOX %d" % ch
            snap.vars[(label, "ch_%d_battery_bars" % ch)] = "5"
            snap.vars[(label, "ch_%d_interference_status" % ch)] = "NONE"
    for label in ("PSM_1", "PSM_2", "PSM_3"):
        for ch in (1, 2):
            snap.vars[(label, "ch_%d_name" % ch)] = "PACK"
            snap.vars[(label, "ch_%d_rf_mute" % ch)] = "UNMUTED"
    snap.midi = {"NA Ingest": {"peers": [{"name": "Lighting PC", "address": "", "state": "running"},
                                         {"name": "SuperRack-LiveBox", "address": "", "state": "running"}]}}
    snap.midi_time = now
    snap.mdns, snap.mdns_time = {"NA Tracks", "Lighting PC", "NA Ingest"}, now
    snap.ping = {"10.1.3.227": 0, "10.6.7.13": 0, "10.1.0.1": 0}
    snap.link = True
    snap.launchd = {"org.waterschurch.resi-propresenter": 884, "org.waterschurch.lobby-captions": 900}
    snap.launchd_time = now
    snap.procs = {"/Applications/ProdDash.app/Contents/MacOS/ProdDash": True}
    snap.http = {"https://central.resi.io/": True, "http://10.6.7.13:8080/": True}
    snap.disk = {"/": 97}
    return snap


def run_steps(engine, snap, start, seconds, step=1.0):
    tiles, t = None, start
    while t <= start + seconds:
        snap.conns_time = t
        tiles = engine.step(snap, t)
        t += step
    return tiles, t


def level(tile):
    for name, ((b, f), _) in sp.COLORS.items():
        if (tile["b"], tile["f"]) == (b, f):
            return name
    return "?"


class TestChecks(unittest.TestCase):
    def setUp(self):
        self.now = SUNDAY
        self.snap = healthy_snapshot(self.now)
        self.ctx = {"now": self.now, "service": True, "any_recording": False}

    def test_all_green_when_healthy(self):
        tiles = sp.Engine(CONFIG).step(self.snap, self.now)
        bad = {k: v["x"] for k, v in tiles.items() if level(v) not in ("ok", "service", "off")}
        self.assertEqual(bad, {})
        self.assertEqual(tiles["all"]["x"], "ALL OK")
        self.assertEqual(tiles["hd1"]["x"], "HD1\n28h")
        self.assertEqual(tiles["dante"]["x"], "DANTE\n19/19")

    def test_connection_states(self):
        c = self.snap.conns["ATEM"]
        c["category"], c["message"] = "error", "Network error: connect ETIMEDOUT 10.1.10.2:9910"
        self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "NO ANSWER")
        c["message"] = "connect EHOSTUNREACH"
        self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "UNREACHABLE")
        c["enabled"] = False
        self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).level, sp.OFF)
        self.assertEqual(sp.conn_result(self.snap, "nope", self.now).msg, "NOT IN COMPANION")
        self.assertEqual(sp.conn_result(self.snap, "x32", self.now + 60).msg, "NO DATA")

    def test_hyperdeck_disk_and_record_sync(self):
        s = self.snap
        s.vars[("HyperDeck_2", "slotId")] = "-"
        s.vars[("HyperDeck_2", "recordingTime")] = "--:--:--"
        out = dict(sp.check_hyperdeck({"conn": "HyperDeck_2"}, s, self.ctx))
        self.assertEqual(out["disk"].msg, "NO DISK")
        self.assertTrue(out["disk"].service_only)
        s.vars[("HyperDeck_3", "recordingTime")] = "0:20:00"
        self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].msg, "LOW 20m")
        s.vars[("HyperDeck_3", "recordingTime")] = "0:04:00"
        self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].level, sp.ERR)
        ctx = dict(self.ctx, any_recording=True)
        self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))["rec"].msg, "NOT REC")
        s.vars[("HyperDeck_1", "status")] = "Record"
        out = dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))
        self.assertEqual(out["rec"].level, sp.OK)
        self.assertEqual(out["conn"].info, "REC 28h")

    def test_optional_deck(self):
        for i in (2, 4):
            self.snap.vars[("HyperDeck_%d" % i, "slotId")] = "-"
            self.snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "--:--:--"
        for i in (1, 3, 5, 6):
            self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual((level(tiles["hd2"]), tiles["hd2"]["x"]), ("off", "HD2\nNO DISK"))
        self.assertEqual(level(tiles["rec"]), "ok")
        self.assertEqual(tiles["all"]["x"], "ALL OK")
        self.snap.vars[("HyperDeck_2", "slotId")] = "1"                 # disk goes in: now it is watched
        self.snap.vars[("HyperDeck_2", "recordingTime")] = "20:00:00"
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["hd2"]["x"], "HD2\nNOT REC")
        self.assertEqual(level(tiles["hd2"]), "err")

    def test_one_deck_not_recording_blinks(self):
        engine = sp.Engine(CONFIG)
        for i in (1, 2, 3, 5, 6):
            self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
        tiles, _ = run_steps(engine, self.snap, self.now, 4)
        self.assertEqual(tiles["hd4"]["x"], "HD4\nNOT REC")
        self.assertEqual(level(tiles["hd4"]), "err")
        self.assertEqual(tiles["all"]["x"], "1 DOWN\nHD4 NOT REC")
        self.assertEqual(tiles["rec"]["x"], "REC\nHD4 NOT REC")

    def test_dante(self):
        self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["NA-Tracks"])
        out = dict(sp.check_dante({"conn": "audinate-dantecontroller",
                                   "devices": {"NA-Tracks": "DANTE", "BOH-WING": "WING"}}, self.snap, self.ctx))
        self.assertEqual(out["BOH-WING"].msg, "WING GONE")
        self.assertEqual(out["NA-Tracks"].level, sp.OK)
        self.snap.vars[("audinate-dantecontroller", "devices")] = None
        out = dict(sp.check_dante({"conn": "audinate-dantecontroller", "devices": ["BOH-WING"]}, self.snap, self.ctx))
        self.assertEqual(out["conn"].msg, "NO DATA")

    def test_dante_button_names_the_missing_device(self):
        devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
        devices.remove("NA-Tracks")
        self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["dante"]["x"], "DANTE\nTRACKS GONE")
        self.assertEqual(level(tiles["dante"]), "err")
        self.assertEqual(tiles["all"]["x"], "1 DOWN\nDANTE TRACKS GONE")

    def test_shure_psm_tascam(self):
        s = self.snap
        s.vars[("Axient_1", "ch_1_battery_bars")] = "1"
        s.vars[("Axient_1", "ch_2_battery_bars")] = "2"
        s.vars[("Axient_1", "ch_3_battery_bars")] = "Unknown"
        s.vars[("Axient_1", "ch_4_interference_status")] = "DETECTED"
        s.vars[("Axient_1", "ch_3_interference_status")] = "DETECTED"     # transmitter off: ignored
        out = dict(sp.check_shure({"conn": "Axient_1", "channels": 4}, s, self.ctx))
        self.assertEqual((out["ch1_batt"].level, out["ch1_batt"].msg), (sp.ERR, "VOX 1 BATT"))
        self.assertEqual(out["ch2_batt"].level, sp.WARN)
        self.assertEqual(out["ch3_batt"].level, sp.OK)
        self.assertEqual(out["ch3_rf"].level, sp.OK)
        self.assertEqual(out["ch4_rf"].msg, "VOX 4 INTERF")
        s.vars[("PSM_2", "ch_1_rf_mute")] = "MUTED"
        self.assertEqual(dict(sp.check_psm({"conn": "PSM_2", "channels": 2}, s, self.ctx))["ch1_mute"].msg,
                         "PACK RF MUTE")
        s.vars[("tascam", "error")] = "Media Error"
        self.assertEqual(dict(sp.check_tascam({"conn": "tascam"}, s, self.ctx))["error"].msg, "MEDIA ERROR")

    def test_rf_group_names_the_mic(self):
        self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "1"
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["rf"]["x"], "RF\nVOX 3 BATT")
        self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "5"
        self.snap.conns["PSM_1"]["category"] = "error"
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertIn("PSM 1 OFFLINE", tiles["rf"]["x"])

    def test_bonjour_parsing_morning_and_afternoon(self):
        out = ("Browsing for _apple-midi._udp.local.\n"
               " 9:20:51.321  Add        3  13 local.               _apple-midi._udp.    NA Tracks\n"
               "19:20:51.321  Add        3  27 local.               _apple-midi._udp.    Lighting PC\n"
               "19:21:02.001  Add        3  27 local.               _apple-midi._udp.    Gone Soon\n"
               "19:21:09.001  Rmv        1  27 local.               _apple-midi._udp.    Gone Soon\n")
        self.assertEqual(sp.parse_dns_sd(out), {"NA Tracks", "Lighting PC"})

    def test_midi_peer_joining_is_not_an_error(self):
        for p in self.snap.midi["NA Ingest"]["peers"]:
            p["state"] = "connected"                      # joined, clock sync not finished yet
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 8)
        self.assertEqual(tiles["midi"]["x"], "MIDI\nOK")

    def test_midi_peer_and_bonjour(self):
        self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"]
                                               if p["name"] != "Lighting PC"]
        self.snap.mdns.discard("NA Tracks")
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["midi"]["x"], "MIDI\nLIGHT PC MIDI LOST\n+1")
        self.assertEqual(tiles["show"]["x"], "SHOW\nLIGHT PC MIDI LOST\n+1")      # named, no "MIDI" prefix
        self.assertNotIn("MIDI TRACKS", tiles["show"]["x"])
        self.snap.midi["NA Ingest"]["peers"].append({"name": "Lighting PC", "address": "", "state": "running"})
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["midi"]["x"], "MIDI\nTRACKS MIDI GONE")

    def test_bible_tv_app_down(self):
        self.snap.http["http://10.6.7.13:8080/"] = False
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["apps"]["x"], "APPS\nBIBLE APP DOWN")

    def test_apps_names_the_app(self):
        self.snap.conns["Propresenter-API"]["category"] = "error"
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["apps"]["x"], "APPS\nPROPRES OFFLINE")
        self.assertEqual(tiles["all"]["x"], "1 DOWN\nPROPRES OFFLINE")

    def test_http_intervals(self):
        checks = dict(sp.http_checks(CONFIG))
        self.assertEqual(checks["http://10.6.7.13:8080/"], 5)
        self.assertEqual(checks["https://central.resi.io/"], 30)

    def test_ingest_services(self):
        del self.snap.launchd["org.waterschurch.resi-propresenter"]
        self.snap.launchd["org.waterschurch.lobby-captions"] = None
        self.snap.disk["/"] = 6
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
        self.assertEqual(tiles["ingest"]["x"], "INGEST\nBRIDGE NOT LOADED\n+2")


class TestTextSize(unittest.TestCase):
    def test_no_word_is_split(self):
        for text in ("TASCAM\nOK", "SERVICE\nON\nmanual", "OBS\nOFFLINE", "INGEST\n97GB free",
                     "4 DOWN\nHD2 NO DISK", "REC\nHD4 NO DISK\n+1", "TRACKS\nMIDI SESS GONE"):
            size = sp.fit_size(text)
            longest = max(len(w) for w in text.split())
            self.assertLessEqual(longest, int(sp.LINE_CAPACITY / size), text)

    def test_short_text_stays_big(self):
        self.assertGreaterEqual(sp.fit_size("HD1\n28h"), 29)
        self.assertGreater(sp.fit_size("ATEM\nOK"), sp.fit_size("TASCAM\nOK"))

    def test_every_tile_has_a_size(self):
        tiles = sp.Engine(CONFIG).step(healthy_snapshot(SUNDAY), SUNDAY)
        for t in tiles.values():
            self.assertTrue(12 <= t["s"] <= 30, t)


class TestBehaviour(unittest.TestCase):
    def setUp(self):
        self.snap = healthy_snapshot(SUNDAY)

    def test_debounce(self):
        engine = sp.Engine(CONFIG)
        self.snap.conns["x32"]["category"] = "error"
        tiles = engine.step(self.snap, SUNDAY)
        self.assertEqual(level(tiles["x32"]), "ok")            # one bad reading is not enough
        tiles, t = run_steps(engine, self.snap, SUNDAY + 1, 3)
        self.assertEqual(level(tiles["x32"]), "err")
        self.snap.conns["x32"]["category"] = "good"
        self.snap.conns_time = t
        self.assertEqual(level(engine.step(self.snap, t)["x32"]), "ok")   # recovery is immediate

    def test_blip_shorter_than_debounce_is_ignored(self):
        engine = sp.Engine(CONFIG)
        self.snap.conns["x32"]["category"] = "error"
        run_steps(engine, self.snap, SUNDAY, 1)
        self.snap.conns["x32"]["category"] = "good"
        tiles, _ = run_steps(engine, self.snap, SUNDAY + 2, 5)
        self.assertEqual(level(tiles["x32"]), "ok")

    def test_flapping_shows_amber(self):
        engine = sp.Engine(CONFIG)
        t = SUNDAY
        for _ in range(3):
            self.snap.conns["x32"]["category"] = "error"
            _, t = run_steps(engine, self.snap, t, 4)
            self.snap.conns["x32"]["category"] = "good"
            tiles, t = run_steps(engine, self.snap, t, 2)
        self.assertEqual(level(tiles["x32"]), "warn")
        self.assertEqual(tiles["x32"]["x"], "X32\nOFFLINE x3")
        tiles, _ = run_steps(engine, self.snap, t + 700, 1)      # quiet for 10 minutes -> green again
        self.assertEqual(level(tiles["x32"]), "ok")

    def test_ack_then_recover_then_fail_again(self):
        engine = sp.Engine(CONFIG)
        self.snap.conns["ATEM"]["category"] = "error"
        tiles, t = run_steps(engine, self.snap, SUNDAY, 4)
        self.assertEqual(level(tiles["atem"]), "err")
        self.assertEqual(engine.ack("atem"), 1)
        tiles, t = run_steps(engine, self.snap, t, 1)
        self.assertEqual(level(tiles["atem"]), "ack")
        self.assertEqual(level(tiles["all"]), "ack")
        self.snap.conns["ATEM"]["category"] = "good"
        tiles, t = run_steps(engine, self.snap, t, 1)
        self.assertEqual(level(tiles["atem"]), "ok")
        self.snap.conns["ATEM"]["category"] = "error"
        tiles, t = run_steps(engine, self.snap, t, 4)
        self.assertEqual(level(tiles["atem"]), "err")             # a new failure blinks again

    def test_group_and_master_ack(self):
        engine = sp.Engine(CONFIG)
        self.snap.conns["VX1000"]["category"] = "error"
        self.snap.conns["x32"]["category"] = "error"
        tiles, t = run_steps(engine, self.snap, SUNDAY, 4)
        self.assertEqual(tiles["all"]["x"].split("\n")[0], "2 DOWN")
        engine.ack("video")
        tiles, t = run_steps(engine, self.snap, t, 1)
        self.assertEqual(level(tiles["video"]), "ack")
        self.assertEqual(level(tiles["x32"]), "err")
        engine.ack("all")
        tiles, t = run_steps(engine, self.snap, t, 1)
        self.assertEqual(level(tiles["all"]), "ack")

    def test_outside_service_nothing_blinks(self):
        snap = healthy_snapshot(WEDNESDAY)
        snap.conns["OBS-Youtube"]["category"] = "error"           # OBS PC is off midweek
        snap.conns["x32"]["category"] = "error"
        snap.vars[("HyperDeck_3", "slotId")] = "-"                # a deck that is always used
        tiles, _ = run_steps(sp.Engine(CONFIG), snap, WEDNESDAY, 4)
        self.assertEqual(level(tiles["obs"]), "idle")
        self.assertEqual(level(tiles["hd3"]), "idle")
        self.assertEqual(tiles["hd3"]["x"], "HD3\nNO DISK")
        self.assertEqual(level(tiles["x32"]), "quiet")
        self.assertEqual(tiles["all"]["x"], "1 DOWN\nX32 OFFLINE")
        self.assertEqual(level(tiles["all"]), "quiet")
        self.assertEqual(tiles["service"]["x"], "SERVICE\nOFF")
        for t in tiles.values():
            self.assertEqual((t["b"], t["f"]), (t["b2"], t["f2"]), t["x"])   # no blinking anywhere

    def test_service_schedule_and_override(self):
        path = os.path.join(tempfile.mkdtemp(), "service.json")
        mode = sp.ServiceMode(CONFIG["service_windows"], path)
        self.assertTrue(mode.on(SUNDAY))
        self.assertFalse(mode.on(WEDNESDAY))
        mode.set(True, WEDNESDAY)                                 # button: on for a Wednesday event
        self.assertTrue(mode.on(WEDNESDAY + 3600))
        self.assertTrue(sp.ServiceMode(CONFIG["service_windows"], path).on(WEDNESDAY + 3600))  # survives restart
        sunday_start = datetime(2026, 10, 11, 7, 0).timestamp()
        self.assertTrue(mode.on(sunday_start))                    # window edge clears the override
        self.assertIsNone(mode.override)
        mode.set(False, sunday_start + 60)                        # turned off early on Sunday
        self.assertFalse(mode.on(sunday_start + 120))
        self.assertFalse(mode.on(datetime(2026, 10, 11, 12, 31).timestamp()))
        self.assertIsNone(mode.override)                          # end of window clears it too

    def test_network_burst(self):
        engine = sp.Engine(CONFIG)
        t = SUNDAY
        run_steps(engine, self.snap, t, 2)
        for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"):   # the Oct 5 10:21 drop
            self.snap.conns[label]["category"] = "error"
        tiles, t = run_steps(engine, self.snap, t + 3, 4)
        self.assertEqual(tiles["net"]["x"], "NET\n5 DROPPED")
        self.assertEqual(level(tiles["net"]), "err")
        for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"):
            self.snap.conns[label]["category"] = "good"
        tiles, t = run_steps(engine, self.snap, t, 1)
        self.assertTrue(tiles["net"]["x"].startswith("NET\nBLIP "))
        self.assertEqual(level(tiles["net"]), "warn")
        tiles, _ = run_steps(engine, self.snap, t + 200, 1)
        self.assertEqual(tiles["net"]["x"], "NET\nOK")

    def test_racks_powering_off_is_not_a_network_event(self):
        engine = sp.Engine(CONFIG)
        run_steps(engine, self.snap, SUNDAY, 2)
        for label in ("Axient_1", "Axient_2", "ULXD-4", "PSM_1", "PSM_2", "PSM_3"):
            self.snap.conns[label]["category"] = "error"
        tiles, _ = run_steps(engine, self.snap, SUNDAY + 3, 4)
        self.assertEqual(tiles["net"]["x"], "NET\nOK")
        self.assertEqual(tiles["rf"]["x"].split("\n")[-1], "+5")

    def test_gateway_and_link(self):
        self.snap.ping["10.1.0.1"] = 2
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
        self.assertEqual(tiles["net"]["x"], "NET\nNO GATEWAY")
        self.snap.link = False
        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1)
        self.assertEqual(tiles["net"]["x"], "NET\nLINK DOWN\n+1")

    def test_restart_targets(self):
        engine = sp.Engine(CONFIG)
        self.assertEqual(engine.restart_targets("hd3"), ["HyperDeck_3"])
        self.assertEqual(engine.restart_targets("led"), ["VX1000", "UHDJr"])
        self.assertEqual(engine.restart_targets("apps"), ["Propresenter-API", "ableton", "BibleTV"])
        self.assertEqual(engine.restart_targets("dante"), [])                 # shared connection: never
        self.assertEqual(engine.restart_targets("ingest"), [])


class FakeCompanion(BaseHTTPRequestHandler):
    """Just enough of Companion 5's HTTP API."""
    store = {}

    def log_message(self, *a):
        pass

    def _send(self, status, body):
        data = body.encode() if isinstance(body, str) else body
        self.send_response(status)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        s = self.store
        path = urllib.parse.urlparse(self.path).path
        if path == "/api/connections":
            return self._send(200, json.dumps([{"id": c["id"], "label": l, "enabled": c["enabled"],
                                                "status": {"category": c["category"], "message": c["message"]}}
                                               for l, c in s["conns"].items()]))
        m = re.match(r"^/api/variable/([^/]+)/([^/]+)/value$", path)
        if m:
            key = (urllib.parse.unquote(m.group(1)), urllib.parse.unquote(m.group(2)))
            v = s["vars"].get(key)
            return self._send(404, "Not found") if v is None else self._send(200, str(v))
        m = re.match(r"^/api/custom-variable/([^/]+)/value$", path)
        if m:
            v = s["custom"].get(m.group(1))
            return self._send(404, "Not found") if v is None else self._send(200, v)
        self._send(404, "")

    def do_POST(self):
        s = self.store
        u = urllib.parse.urlparse(self.path)
        m = re.match(r"^/api/custom-variable/([^/]+)/value$", u.path)
        if m:
            if m.group(1) not in s["custom"]:
                return self._send(404, "Not found")
            s["custom"][m.group(1)] = urllib.parse.parse_qs(u.query, keep_blank_values=True)["value"][0]
            return self._send(200, "ok")
        m = re.match(r"^/api/connections/([^/]+)/restart$", u.path)
        if m:
            s["restarts"].append(m.group(1))
            return self._send(200, "{}")
        self._send(404, "")


class TestAgainstFakeCompanion(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        snap = healthy_snapshot(time.time())
        FakeCompanion.store = {
            "conns": snap.conns,
            "vars": dict(snap.vars),
            "custom": {"sp_state": "", "sp_hb": "", "sp_cmd": "", "flash_tick": "0"},
            "restarts": [],
        }
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion)
        threading.Thread(target=cls.server.serve_forever, daemon=True).start()
        cls.port = cls.server.server_address[1]

    @classmethod
    def tearDownClass(cls):
        cls.server.shutdown()

    def test_client_round_trip_and_long_state(self):
        comp = sp.Companion("127.0.0.1", self.port)
        self.assertEqual(len(comp.connections()), len(FakeCompanion.store["conns"]))
        self.assertEqual(comp.variable("HyperDeck_1", "recordingTime"), "28:00:00")
        self.assertIsNone(comp.variable("HyperDeck_1", "nope"))
        snap = healthy_snapshot(time.time())
        tiles = sp.Engine(CONFIG).step(snap, time.time())
        js = json.dumps(tiles, separators=(",", ":"), sort_keys=True)
        self.assertEqual(comp.set_custom("sp_state", js), (True, 200))
        self.assertEqual(json.loads(comp.get_custom("sp_state"))["all"]["x"], "ALL OK")
        self.assertEqual(comp.set_custom("missing", "x"), (False, 404))
        print("\n  sp_state is %d bytes, %d in the URL" % (len(js), len(urllib.parse.quote(js, safe=""))))

    def test_commands(self):
        comp = sp.Companion("127.0.0.1", self.port)
        snap = sp.Snapshot()
        sp.poll_connections(comp, snap)
        engine = sp.Engine(CONFIG)
        now = SUNDAY
        sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now)
        self.assertEqual(FakeCompanion.store["restarts"][-3:], ["id-HyperDeck_3", "id-VX1000", "id-UHDJr"])
        self.assertFalse(engine.service.on(now))
        sp.handle_commands("service toggle", engine, comp, snap, now)
        self.assertTrue(engine.service.on(now))

    def test_end_to_end_loop(self):
        """The real run() loop: it writes state and heartbeat, and consumes button commands."""
        cfg = json.loads(json.dumps(CONFIG))
        cfg["companion"]["port"] = self.port
        # keep it offline: no pings, no web, no launchd, no bonjour lookups
        cfg["network"]["gateway"] = None
        keep = ("conn", "hyperdeck", "dante", "tascam", "shure", "psm")
        for t in cfg["tiles"]:
            if "checks" in t:
                t["checks"] = [c for c in t["checks"] if c["type"] in keep] or [{"type": "conn", "conn": "x32"}]
        store = FakeCompanion.store
        store["conns"]["ATEM"]["category"] = "error"
        store["custom"]["sp_cmd"] = ";ack atem"
        data_dir = tempfile.mkdtemp()
        originals = sp.poll_mdns, sp.poll_system, sp.poll_midi
        sp.poll_mdns = lambda snap, seconds=3: None
        sp.poll_system = lambda snap, cfg: None
        sp.poll_midi = lambda snap, cfg: None            # CoreMIDI has its own test
        stop = threading.Event()
        loop = threading.Thread(target=sp.run, args=(cfg, data_dir), kwargs={"stop": stop}, daemon=True)
        try:
            loop.start()
            deadline = time.time() + 12
            state = {}
            while time.time() < deadline:
                raw = store["custom"]["sp_state"]
                state = json.loads(raw) if raw else {}
                if state.get("atem", {}).get("x") == "ATEM\nOFFLINE":
                    break
                time.sleep(0.25)
        finally:
            stop.set()
            loop.join(timeout=15)
            sp.poll_mdns, sp.poll_system, sp.poll_midi = originals
            store["conns"]["ATEM"]["category"] = "good"
        self.assertFalse(loop.is_alive())                 # it shuts down cleanly
        self.assertTrue(store["custom"]["sp_hb"].isdigit())
        self.assertLess(abs(int(store["custom"]["sp_hb"]) - time.time()), 5)
        self.assertEqual(store["custom"]["sp_cmd"], "")          # the command was consumed
        self.assertIn("atem", state)
        self.assertIn("ATEM", state["atem"]["x"])


class TestPage(unittest.TestCase):
    def setUp(self):
        self.export = make_page.build(CONFIG)
        self.controls = self.export["page"]["controls"]

    def test_layout(self):
        self.assertEqual(self.export["type"], "page")
        self.assertEqual(self.export["oldPageNumber"], 19)
        self.assertEqual(self.controls["0"]["0"], {"type": "pageup"})       # only the page number follows ALL
        self.assertEqual(self.controls["2"]["0"], {"type": "pagedown"})
        for (r, kind, action) in (("1", "pagenum", "set_page"),):
            key = self.controls[r]["0"]
            self.assertEqual(key["type"], "button-layered")
            self.assertTrue(key["options"]["notes"].startswith("STATUS nav key (%s)" % kind))
            step = key["steps"]["0"]
            self.assertEqual(step["action_sets"]["down"], [])
            tap, hold = step["action_sets"]["up"][0], step["action_sets"]["1000"][0]
            self.assertEqual((tap["definitionId"], tap["options"]["page"]["value"]), (action, "startup"))
            self.assertEqual((hold["definitionId"], hold["options"]["page"]["value"]), ("set_page", 19))
            self.assertEqual(step["options"]["runWhileHeld"], [1000])
            box = [l for l in key["style"]["layers"] if l["type"] == "box"][0]
            self.assertTrue(box["color"]["isExpression"])
            self.assertIn("'$.all.b'", box["color"]["value"])          # follows the ALL button
        positioned = [t for t in CONFIG["tiles"] if "pos" in t]
        n = sum(1 for r in self.controls.values() for c in r.values() if c["type"] == "button-layered")
        self.assertEqual(n, len(positioned) + 1)                       # + the page number key
        # The master is where the smallest Stream Deck can see it.
        self.assertEqual(CONFIG["tiles"][0]["pos"], [0, 1])

    def test_each_button_reads_its_own_tile(self):
        for t in CONFIG["tiles"]:
            if "pos" not in t:
                continue
            r, c = (str(x) for x in t["pos"])
            layers = {l["id"]: l for l in self.controls[r][c]["style"]["layers"]}
            for lid, key in (("box0", "color"), ("text0", "color"), ("text0", "text"), ("text0", "fontsize")):
                expr = layers[lid][key]
                self.assertTrue(expr["isExpression"])
                self.assertIn("'$.%s." % t["id"], expr["value"])
                self.assertIn("$(custom:sp_hb) > 15", expr["value"])
            self.assertIn("$(custom:flash_tick) == 1", layers["box0"]["color"]["value"])

    def test_actions(self):
        hd1 = self.controls["1"]["1"]["steps"]["0"]
        up = hd1["action_sets"]["up"][0]["options"]["value"]
        self.assertEqual(up, {"value": "$(custom:sp_cmd);ack hd1", "isExpression": False})
        self.assertEqual(hd1["action_sets"]["2000"][0]["options"]["value"]["value"], "$(custom:sp_cmd);restart hd1")
        self.assertEqual(hd1["options"]["runWhileHeld"], [2000])
        master = self.controls["0"]["1"]["steps"]["0"]
        self.assertIn("ackall", master["action_sets"]["up"][0]["options"]["value"]["value"])
        self.assertNotIn("2000", master["action_sets"])
        ingest = self.controls["3"]["5"]["steps"]["0"]
        self.assertNotIn("2000", ingest["action_sets"])              # nothing to restart there
        ids = [a["id"] for r in self.controls.values() for c in r.values()
               for s in c.get("steps", {}).values() for al in s["action_sets"].values() for a in al]
        self.assertEqual(len(ids), len(set(ids)))

    def test_stale_text_and_file(self):
        master = {l["id"]: l for l in self.controls["0"]["1"]["style"]["layers"]}
        self.assertIn("`${'WATCHDOG\\nDOWN\\n'}${$(internal:instance_errors)} ERR`", master["text0"]["text"]["value"])
        # No string '+' anywhere: in Companion 5 expressions it adds numbers.
        for r in self.controls.values():
            for c in r.values():
                for layer in c.get("style", {}).get("layers", []):
                    for p in layer.values():
                        if isinstance(p, dict) and p.get("isExpression"):
                            self.assertNotRegex(p["value"], r"'\s*\+|\+\s*'")
        path = os.path.join(tempfile.mkdtemp(), "p.companionconfig")
        sys.argv = ["make_page.py", path]
        make_page.main()
        on_disk = make_page.build(sp.load_config(os.path.join(ROOT, "config.json")))
        with gzip.open(path) as f:
            self.assertEqual(json.load(f), json.loads(json.dumps(on_disk)))


class TestCoreMIDI(unittest.TestCase):
    def test_listener_loopback(self):
        import coremidi
        src = coremidi.VirtualSource("status-panel-test")
        try:
            time.sleep(0.3)
            lis = coremidi.Listener("status-panel-test")
            lis.poll()
            self.assertTrue(lis.connected)
            src.send([0x90, 36, 100])
            src.send([0xF8])                       # clock is ignored
            time.sleep(0.3)
            self.assertEqual((lis.last_msg, lis.count), ("NOTE 36", 1))
            snap = sp.Snapshot()
            snap.listeners["status-panel-test"] = lis
            res = sp.check_midi_rx({"source": "status-panel-test", "name": "IN"}, snap,
                                   {"now": time.time(), "service": True})[0][1]
            self.assertEqual(res.info, "IN 0s")
        finally:
            src.close()
        time.sleep(0.3)
        lis.poll()
        self.assertFalse(lis.connected)

    def test_describe(self):
        import coremidi
        self.assertEqual(coremidi.describe([0xB0, 7, 64]), "CC 7=64")
        self.assertEqual(coremidi.describe([0xC2, 5]), "PC 5")
        self.assertEqual(coremidi.describe([0x90, 60, 0]), "OFF 60")
        self.assertEqual(coremidi.describe([0xF1, 0x20]), "MTC")


if __name__ == "__main__":
    unittest.main()