status-indicators/tests/test_readiness.py

"""The READY page, offline: readiness.py's verdicts on this weekend's checklist (from trimmed copies of
what PCO, ProPresenter, the Ableton set, ProdDash and Resolume really said on 2026-10-14), and how the
watchdog and the page builder show them. Run:  python3 -m unittest discover -s tests"""
import json
import logging
import os
import sys
import unittest
from datetime import datetime

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

import make_page  # noqa: E402
import readiness as rd  # 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"))
CONFIG["tiles"] = [t for t in CONFIG["tiles"] if t.get("kind") != "autofix"]   # ALARMS doing both, as these tests expect
# The tests pin their own READY hours (the location sets its own on the settings page).
CONFIG["ready"]["due"] = [{"days": ["sat"], "start": "12:00", "end": "23:59"}, {"days": ["sun"], "start": "00:00", "end": "13:30"}]
RC = CONFIG["ready"]
WEDNESDAY = datetime(2026, 10, 14, 15, 0).timestamp()       # the week before the plan: gaps are amber
SATURDAY = datetime(2026, 10, 17, 16, 0).timestamp()        # inside "due": gaps are red
SUNDAY_PLAN = datetime(2026, 10, 18, 9, 30).timestamp()


def item(title, type_="item", position="during", description="", notes=None):
    return {"id": "i-" + title, "note_ids": {k: "n-" + title for k in (notes or {})}, "title": title, "type": type_,
            "position": position, "description": description, "notes": notes or {}}


def plan():
    return {"id": "92013595", "title": "Part 8", "dates": "October 18, 2026", "time": SUNDAY_PLAN,
            "items": [
                item("Doors Open", "header"),
                item("Preservice", position="pre"),
                item("MC Announcements", "media", "pre", "- Celebrate recovery\n- WK Team night\n- Peru missions trip\n- Family skate night"),
                item("Worship", "header"),
                item("Rejoice Again", "song", notes={"LTC": "00:01:17:26"}),
                item("Can't Help But Praise (Live)", "song", notes={"LTC": "00:06:43:21"}),
                item("Holy Forever", "song", notes={"LTC": "00:13:33:07"}),
                item("Baptisms", "media", description="730 930 None\n1130 Samantha Lopez"),
                item("Message", "header"),
                item("Message", description="Pastor Tim Live"),
                item("Covered By The Blood TAG", "song", notes={"LTC": "00:24:13:07"}),
                item("Postservice", "song", "post"),
            ],
            "team": [
                {"name": "Jonah Perdue", "position": "[Leader]", "team": "Band", "status": "C"},
                {"name": "Krista Perdue", "position": "Vocalist", "team": "Band", "status": "C"},
                {"name": "Joshua Tineo", "position": "Vocalist", "team": "Band", "status": "D"},
                {"name": "Kevin Michel", "position": "Piano/Pad", "team": "Band", "status": "C"},
                {"name": "Luis Decamps", "position": "Drums", "team": "Band", "status": "D"},
                {"name": "Anthony Hoyt", "position": "MC", "team": "Communication", "status": "C"},
                {"name": "Tim Hatch", "position": "Speaker", "team": "Communication", "status": "C"},
                {"name": "Chris Meireles", "position": "Audio - FOH", "team": "Sunday Production", "status": "C"},
            ]}


def slide(label="", text="", notes="", enabled=True):
    return {"label": label, "text": text, "notes": notes, "enabled": enabled}


def pres(name, slides, timeline=False, cues=None):
    return {"name": name, "path": "/Volumes/MEDIA/ProPresenter/Libraries/%s.pro" % name, "type": "multitracks" if timeline else "standard", "timeline": timeline,
            "cues": len(slides) if cues is None else cues, "slides": slides}


OLD_NOTES = "- Operation Christmas child\n- 5k\n- Movie night\n- Flourish"


def prop():
    song = lambda name, bg="Heavenly_Lights_1_4k.mp4": pres(name, [slide(bg, "Verse"), slide("", "Chorus")], timeline=True)
    return {"playlist": "Where is Your Faith? - Part 8 - October 18, 2026", "items": [
        {"name": "Doors Open", "type": "header", "pco": True, "pres": None},
        {"name": "Preservice", "type": "presentation", "pco": True, "pres": pres("Preservice", [slide("CLICK THIS TO START")])},
        {"name": "MC Announcements", "type": "presentation", "pco": True,
         "pres": pres("MC Announcements", [slide("ANNOUNCEMENTS", notes=OLD_NOTES)] + [slide(notes=OLD_NOTES)] * 4)},
        {"name": "Worship", "type": "header", "pco": True, "pres": None},
        {"name": "Rejoice Again - [ Full ]", "type": "presentation", "pco": True, "pres": song("Rejoice Again")},
        {"name": "Can't Help But Praise LM - [ Full ]", "type": "presentation", "pco": True, "pres": song("Can't Help But Praise LM")},
        {"name": "Holy Forever - [ Full ]", "type": "presentation", "pco": True, "pres": song("Holy Forever", bg="")},
        {"name": "Baptisms", "type": "presentation", "pco": True,
         "pres": pres("Baptisms", [slide("Zoey 1130"), slide("LIVE BAPTISM"), slide("Bobby 1130")])},
        {"name": "Message", "type": "header", "pco": True, "pres": None},
        {"name": "Message Bumper", "type": "presentation", "pco": True, "pres": pres("Message Bumper", [slide("Bumper.mp4")])},
        {"name": "Covered By The Blood TAG - [ TAG ]", "type": "presentation", "pco": True,
         "pres": pres("Covered By The Blood TAG", [slide("Grainyents_Dark_2_4k.mp4")], timeline=False, cues=0)},
        {"name": "Postservice", "type": "presentation", "pco": True, "pres": pres("Postservice", [slide("24_Prayer-1.png")])},
    ]}


def track(id_, name, kind="AudioTrack", group="-1", vol=1.0, ext=True, clips=((0, 100, None),)):
    return {"id": id_, "kind": kind, "name": name, "group": group, "vol": vol, "ext": ext,
            "out": "Ext. Out 5/6 Keys" if ext else "Master", "clips": [list(c) for c in clips]}


def ableton(lighting=((0, 50, 12), (100, 150, 8))):
    tracks = [
        track("1", "LIGHTING CUES", "MidiTrack", clips=lighting),
        track("10", "REJOICE AGAIN", "GroupTrack", clips=()),
        track("11", "21-Piano", group="10", vol=0.25, clips=((0, 90, None),)),
        track("12", "22-Keys", group="10", clips=((0, 90, None),)),
        track("20", "CAN'T HELP BUT PRAISE", "GroupTrack", clips=()),
        track("21", "44-Piano", group="20", clips=((100, 190, None),)),
        track("22", "45-Piano 2", group="20", clips=((100, 190, None),)),
        track("23", "Sub group", "GroupTrack", group="20", clips=()),
        track("24", "46-EG", group="23", clips=((100, 190, None),), ext=False),
        track("30", "HOLY FOREVER", "GroupTrack", clips=()),
        track("31", "72-Piano", group="30", vol=0.5, clips=((200, 290, None),)),
        track("40", "COVERED BY THE BLOOD", "GroupTrack", clips=()),
        track("41", "80-Bass", group="40", vol=0.7, clips=((300, 390, None),)),
    ]
    path = "/Projects/NA/261018 NA Project/261018 NA Project.als"
    # The real tempo map and locators of 261018 NA Project (Rejoice 151 BPM, Can't Help 143, Holy Forever 72, Covered 76).
    tempo = {"manual": 68.3, "events": [[-63072000, 143], [0, 143], [0, 151], [1016, 151], [1016, 143], [1992, 143],
                                        [1992, 72], [2760, 72], [2760, 76], [5599.5, 76], [5599.5, 71]]}
    locators = [{"name": "1- CALL TO WORSHIP ", "time": 0}, {"name": "LOOP", "time": 76},
                {"name": "2 - REJOICE AGAIN - Bb", "time": 196}, {"name": "3- CAN'T HELP BUT PRAISE - Eb", "time": 1016},
                {"name": "4- HOLY FOREVER - B", "time": 1992}, {"name": "5- COVERED BY THE BLOOD-  B", "time": 2760}]
    ltc = {"start": 0.0, "name": "LTC_00000000_60mins_30fps_48000x24", "warped": False, "offset": 0.0}
    return {"set": path, "live_set": path, "mtime": WEDNESDAY,
            "summary": {"tracks": tracks, "locators": locators, "tempo": tempo, "ltc": ltc},
            "bundles": {"AB": {"folder": "/Projects/AB/261018 AB Project", "als": True, "copying": False},
                        "SC": {"folder": "/Projects/SC/261018 SC Project", "als": False, "copying": False}}}


def lineup(names, plan_id="92013595"):
    return {"state": {"plan": {"id": plan_id, "dates": "October 18, 2026"}, "lastError": "",
                      "slots": [{"index": i, "label": "S%d" % i, "person": {"name": n} if n else None} for i, n in enumerate(names)]}}


def composition(deck="Service", empty=(), ignored=(), columns=12):
    """Resolume's "Crosses": a clip in every column of layer 1 but the `empty` ones; layer 2 all empty."""
    clip = lambda n, full, ign: {"name": {"value": "Clip %d" % n if full else ""},
                                 "connected": {"value": "Disconnected" if full else "Empty"},
                                 "ignorecolumntrigger": {"value": "On" if ign else "Layer Determined"}}
    return {"name": {"value": "Crosses"},
            "decks": [{"name": {"value": "Service"}, "selected": {"value": deck == "Service"}},
                      {"name": {"value": "Test"}, "selected": {"value": deck == "Test"}}],
            "columns": [{"name": {"value": "Theme 10 White" if n == 11 else "Column %d" % n}} for n in range(1, columns + 1)],
            "layers": [{"clips": [clip(n, n not in empty, n in ignored) for n in range(1, columns + 1)]},
                       {"clips": [clip(n, False, False) for n in range(1, columns + 1)]}]}


# Companion as the chain really is: page 17 "White" presses 6/1/4, page 6's 1/4 connects column 11;
# page 17 "Salvation" presses 6/1/5 (column 12) and 17/2/4 (White again); 17/0/1 presses itself.
COMPANION = {
    17: {"controls": {"2": {"4": {"steps": {"0": {"action_sets": {"down": [
            {"connectionId": "internal", "definitionId": "button_pressrelease", "options": {"location": {"value": "6/1/4"}}}]}}}}},
                      "4": {"3": {"steps": {"0": {"action_sets": {"down": [
            {"connectionId": "internal", "definitionId": "button_pressrelease", "options": {"location": {"value": "6/1/5"}}},
            {"connectionId": "internal", "definitionId": "button_pressrelease", "options": {"location": {"value": "17/2/4"}}}]}}}}},
                      "0": {"1": {"steps": {"0": {"action_sets": {"down": [
            {"connectionId": "internal", "definitionId": "button_pressrelease", "options": {"location": {"value": "17/0/1"}}}]}}}}}}},
    6: {"controls": {"1": {"4": {"steps": {"0": {"action_sets": {"down": [
            {"connectionId": "arena-id", "definitionId": "custom", "options": {"customPath": {"value": "/composition/columns/11/connect"}}}]}}}},
                           "5": {"steps": {"0": {"action_sets": {"down": [
            {"connectionId": "arena-id", "definitionId": "custom", "options": {"customPath": {"value": "/composition/columns/12/connect"}}}]}}}}}}},
}
MACROS = {"U-WHITE": {"name": "White", "cc": [["Stream Deck", "CC x", "17/2/4"]], "calls": []},
          "U-SALV": {"name": "Salvation Moment", "cc": [["Stream Deck", "CC x", "17/4/3"]], "calls": []},
          "U-LOOP": {"name": "Loop", "cc": [["Stream Deck", "CC x", "17/0/1"]], "calls": ["U-WHITE"]}}


def macros_found(paths):
    """Holy Forever and Rejoice Again fire White; Baptisms fires the macro that runs White too."""
    use = {"Holy Forever": ["U-WHITE"], "Rejoice Again": ["U-WHITE"], "Baptisms": ["U-LOOP"]}
    return {"macros": MACROS, "used": {p: use.get(p.rsplit("/", 1)[-1][:-4], []) for p in paths}}


def msgs(line):
    return [m for _, m, _ in line.faults]


class TestNames(unittest.TestCase):
    def test_titles_match_across_pco_propresenter_and_ableton(self):
        self.assertTrue(rd.same("Can't Help But Praise (Live)", "Can't Help But Praise LM - [ Full ]"))
        self.assertTrue(rd.same("Can't Help But Praise (Live)", "3- CAN’T HELP BUT PRAISE - Eb"))
        self.assertTrue(rd.same("Covered By The Blood TAG", "COVERED BY THE BLOOD"))
        self.assertTrue(rd.same("Message", "Message Bumper"))
        self.assertFalse(rd.same("Holy Forever", "Holy Spirit"))
        self.assertFalse(rd.same("", "Anything"))

    def test_lists_in_descriptions(self):
        self.assertEqual(rd.bullets("- Celebrate recovery\n- WK Team night\nnot a bullet"), ["Celebrate recovery", "WK Team night"])
        self.assertEqual(rd.listed_names("730 930 None\n1130 Samantha Lopez, Bob Ray & Ann Lee"),
                         ["Samantha Lopez", "Bob Ray", "Ann Lee"])

    def test_the_plan_is_todays_or_the_next(self):
        p = lambda pid, iso: {"id": pid, "attributes": {"sort_date": iso}}
        plans = [p("old", "2026-10-11T13:30:00Z"), p("next", "2026-10-18T13:30:00Z"), p("later", "2026-10-25T13:30:00Z")]
        self.assertEqual(rd.pick_plan(plans, WEDNESDAY)[0]["id"], "next")
        self.assertEqual(rd.pick_plan(plans, datetime(2026, 10, 11, 12, 0).timestamp())[0]["id"], "old")   # Sunday itself
        self.assertEqual(rd.pick_plan([], WEDNESDAY), (None, None))

    def test_week_starts_monday_midnight(self):
        self.assertEqual(rd.week_start(WEDNESDAY), datetime(2026, 10, 12).timestamp())
        self.assertEqual(rd.week_start(datetime(2026, 10, 12, 0, 0).timestamp()), datetime(2026, 10, 12).timestamp())


class TestPropresenter(unittest.TestCase):
    def test_pco_against_the_playlist(self):
        out = rd.eval_prop(RC, plan(), prop())
        self.assertEqual(msgs(out["prop"]), ["MC SLIDES OLD?", "BAPTISM: NO SAMANTHA"])
        self.assertEqual(msgs(out["timecode"]), ["NO TC COVERED BY THE."])
        self.assertEqual(msgs(out["backgrounds"]), ["NO BG HOLY FOREVER"])

    def test_too_few_announcement_slides_and_a_missing_item(self):
        p = plan()
        p["items"][2]["description"] += "\n- Fifth thing\n- Sixth thing"
        p["items"].insert(4, item("Welcome Video", "media"))
        pp = prop()
        pp["items"][2]["pres"]["slides"] = [slide(notes=p["items"][2]["description"])] * 3
        out = rd.eval_prop(RC, p, pp)
        self.assertIn("MC 3/6 SLIDES", msgs(out["prop"]))
        self.assertIn("MISSING WELCOME VIDEO", msgs(out["prop"]))
        self.assertNotIn("MC SLIDES OLD?", msgs(out["prop"]))          # the notes are this week's

    def test_no_playlist_for_the_weekend(self):
        out = rd.eval_prop(RC, plan(), {"playlist": None, "items": []})
        self.assertTrue(all(msgs(line)[0].startswith("NO PLAYLIST") for line in out.values()))


class TestAbleton(unittest.TestCase):
    def test_mixed_routed_and_lit(self):
        out = rd.eval_ableton(RC, plan(), ableton())
        self.assertEqual(msgs(out["tracks"]), [])
        # every Can't Help track is at 0 dB, and one of them goes to Master; Holy Forever has no lighting cue
        self.assertEqual(msgs(out["mix_foh"]), ["NOT MIXED CANT HELP BUT P.", "TO MASTER CANT HELP BUT P."])
        self.assertEqual(out["mix_foh"].faults[0][0], rd.WARN)
        self.assertEqual(msgs(out["lights"]), ["NO LIGHTS HOLY FOREVER", "NO LIGHTS COVERED BY THE."])

    def test_a_song_with_no_tracks_and_an_old_set_open(self):
        al = ableton()
        al["summary"]["tracks"] = [t for t in al["summary"]["tracks"] if t["id"] not in ("30", "31")]
        al["live_set"] = "/Projects/NA/261011 NA Project/261011 NA Project.als"
        out = rd.eval_ableton(RC, plan(), al)
        self.assertEqual(msgs(out["tracks"]), ["NO TRACKS HOLY FOREVER", "LIVE HAS OLD SET"])

    def test_no_set_for_the_weekend(self):
        out = rd.eval_ableton(RC, plan(), {"set": None, "bundles": {}})
        self.assertEqual(msgs(out["tracks"]), ["NO SET 261018"])

    def test_tracks_sent_to_the_locations(self):
        self.assertEqual(msgs(rd.eval_track_bundle(RC, ableton())), ["NOT SENT SC"])

    def test_the_remote_scripts_are_valid_python(self):
        params = json.dumps(rd.ableton_params(RC, plan()))
        for script in (rd.ALS_SCRIPT % params, rd.COPY_SCRIPT % params, rd.BUNDLE_SCRIPT % json.dumps({"folder": "x", "pattern": "*"})):
            compile(script, "remote", "exec")
        self.assertEqual(json.loads(params)["date"], "261018")


FINISHED = "/Drive/Ableton/Finished Songs"


def keyed(keys):
    """plan() with PCO keys on its songs ({title: key})."""
    p = plan()
    for it in p["items"]:
        it["key"] = keys.get(it["title"], "")
    return p


def with_pads_and_finished(al=None):
    """The set with the real pads of 261025 (named by key) under its songs, and a slice of Finished Songs."""
    al = al or ableton()
    al["summary"]["pads"] = [[0, 95, "A# - WARM - CHURCHFRONT PADS"], [95, 196, "D# - WARM - CHURCHFRONT PADS"],
                             [196, 300, "B - WARM - CHURCHFRONT PADS"], [300, 400, "B - WARM - CHURCHFRONT PADS"]]
    al["finished"] = {FINISHED: ["CAN'T HELP BUT PRAISE LM [Eb]", "CAN'T HELP BUT PRAISE [D]", "CAN'T HELP BUT PRAISE [Eb]",
                                 "COVERED BY THE BLOOD [B]", "HERE AGAIN UNFINISHED [C]", "HOLY FOREVER [B]",
                                 "REJOICE AGAIN [B]", "zHOLY FOREVER [F]"]}
    return al


GOOD = {"Rejoice Again": "Bb", "Can't Help But Praise (Live)": "Eb", "Holy Forever": "B", "Covered By The Blood TAG": "B"}


class TestKeys(unittest.TestCase):
    def test_key_names(self):
        self.assertTrue(rd.same_key("Eb", "D#"))
        self.assertTrue(rd.same_key("Bb", "A#"))
        self.assertFalse(rd.same_key("B", "Bb"))
        self.assertFalse(rd.same_key("Em", "E"))
        self.assertTrue(rd.same_key("Em", "E", mode=False))
        self.assertIsNone(rd.key_of("LOOP"))

    def test_every_song_in_its_key(self):
        al = with_pads_and_finished()
        al["finished"][FINISHED].append("REJOICE AGAIN [Bb]")
        line = rd.eval_keys(RC, keyed(GOOD), al)
        self.assertEqual(msgs(line), [])
        self.assertEqual(line.info, "4 KEYS OK")

    def test_the_set_in_another_key(self):
        keys = dict(GOOD, **{"Holy Forever": "C"})
        line = rd.eval_keys(RC, keyed(keys), with_pads_and_finished())
        # Rejoice Again: Finished Songs only has it in B (PCO and the set say Bb)
        self.assertEqual(msgs(line), ["REJOICE AGAIN ONLY B", "HOLY FOREVER B NOT C", "HOLY FOREVER PAD B NOT C",
                                      "HOLY FOREVER ONLY B"])
        self.assertEqual([lv for lv, _, _ in line.faults], [rd.WARN, rd.ERR, rd.ERR, rd.WARN])

    def test_a_song_pco_has_no_key_for(self):
        line = rd.eval_keys(RC, keyed(dict(GOOD, **{"Holy Forever": ""})), with_pads_and_finished())
        self.assertIn("NO KEY IN PCO HOLY FOREVER", msgs(line))

    def test_nothing_to_check_a_key_by(self):
        al = ableton()
        al["summary"]["locators"] = [{"name": "HOLY FOREVER", "time": 200}]
        al["summary"]["tracks"] = [t for t in al["summary"]["tracks"] if t["id"] in ("30", "31")]
        line = rd.eval_keys(RC, keyed(GOOD), al)
        self.assertEqual(msgs(line), ["KEY UNCHECKED REJOICE AGAIN", "KEY UNCHECKED CANT HELP BUT P.",
                                      "KEY UNCHECKED HOLY FOREVER", "KEY UNCHECKED COVERED BY THE."])

    def test_a_song_missing_from_finished_songs_is_only_noted(self):
        al = with_pads_and_finished()
        al["finished"][FINISHED].remove("HOLY FOREVER [B]")
        line = rd.eval_keys(RC, keyed(GOOD), al)
        self.assertNotIn("HOLY FOREVER", " ".join(msgs(line)))
        self.assertIn("Holy Forever is not in Finished Songs", line.as_dict()["details"])

    def test_no_set(self):
        self.assertEqual(msgs(rd.eval_keys(RC, keyed(GOOD), {"set": None})), ["NO SET 261018"])


def live(name, index, grouped=True, type_="Ext. Out", channel="5/6 Keys", send=1.0, group=False, audio=True):
    return {"name": name, "index": index, "group": group, "grouped": grouped, "audio": audio, "type": type_,
            "channel": channel, "send": send}


def live_set(**over):
    """What AbletonOSC said about 261025 NA Project on 2026-10-22, trimmed: a song dragged in last (its stems'
    in-ear sends at -inf, one routed to Master) and a new stem name."""
    tracks = [
        live("TEMPO", 0, False, "Master", "", 0.0),
        live("LTC", 1, False, "Ext. Out", "16 LTC", 0.0),
        live("MIDI GROUP", 2, False, group=True),
        live("WAVES TUNE", 3, audio=False),
        live("RESI TRACKS", 4, False, group=True),
        live("RESI TRACKS", 5, True, "Ext. Out", "3/4 IN-EARS", 0.0),
        live("RESI KEYS", 6, True, "RESI TRACKS", "", 0.0),
        live("AMBIENT PADS", 7, False, "Ext. Out", "5/6 Keys", 0.36),
        live("CAN'T HELP BUT PRAISE", 8, False, group=True),
        live("21-Piano", 9),
        live("22-Piano 2", 10, send=0.75),
        live("29-EG 1", 11, channel="7/8 GTRS", send=0.44),
        live("JEHOVAH", 12, False, group=True),
        live("50-Keys 1 [2026-10-21 132128]", 13, send=0.0),
        live("56-EG 13 [2026-10-21 132128]", 14, channel="7/8 GTRS", send=0.0),
        live("62-Bass [2026-10-21 132128]", 15, True, "Master", "", 0.0),
        live("70-Banjo [2026-10-21 132128]", 16, channel="7/8 GTRS", send=0.5),
    ]
    out = {"playing": False, "tracks": tracks}
    out.update(over)
    return out


class TestAbletonLive(unittest.TestCase):
    def test_routing_and_in_ears(self):
        routing, ears, changes = rd.mix_plan(RC, live_set())
        self.assertEqual(msgs(routing), ["BASS TO MASTER", "NEW STEM BANJO"])
        self.assertEqual(msgs(ears), ["EARS OFF JEHOVAH"])
        self.assertIn("50-Keys 1 [2026-10-21 132128] in JEHOVAH: send A is at -inf", ears.as_dict()["details"])
        # EG 13 is not in the template, but every EG goes to 7/8 GTRS; RESI and top-level tracks keep their -inf sends
        self.assertEqual(changes, [
            {"index": 13, "name": "50-Keys 1 [2026-10-21 132128]", "send": 1.0},
            {"index": 14, "name": "56-EG 13 [2026-10-21 132128]", "send": 1.0},
            {"index": 15, "name": "62-Bass [2026-10-21 132128]", "type": "Ext. Out", "channel": "7/8 GTRS", "send": 1.0}])
        self.assertEqual(ears.info, "4 SENDS ON")

    def test_stem_names(self):
        self.assertEqual(rd.stem_name("50-Keys 1 [2026-10-21 132128]"), "KEYS 1")
        self.assertEqual(rd.stem_name("105-Drums (Live)"), "DRUMS (LIVE)")
        self.assertEqual(rd.template_for(RC["routing"], "88-Drums (Live) [2026-10-19 132517]"), ["Ext. Out", "9/10 PERC / FX"])
        self.assertIsNone(rd.template_for(RC["routing"], "Banjo"))

    def test_ableton_not_answering(self):
        routing, ears, changes = rd.mix_plan(RC, {"error": "ABLETON NOT ANSWERING"})
        self.assertEqual((msgs(routing), msgs(ears), changes), (["ABLETON NOT ANSWERING"], ["ABLETON NOT ANSWERING"], []))

    def test_the_live_script_runs(self):
        for apply in (None, [{"index": 1, "name": "x", "send": 1.0}]):
            compile(rd.LIVE_SCRIPT % json.dumps({"send_index": 0, "apply": apply}), "remote", "exec")

    def test_the_hold_reads_then_sets(self):
        calls = []

        def fake(cfg, rc, apply=None):
            calls.append(apply)
            return live_set() if apply is None else {"done": [c["name"] for c in apply], "skipped": []}
        real, rd.live_look = rd.live_look, fake
        try:
            self.assertTrue(rd.run_action(CONFIG, "ableton_mix"))
        finally:
            rd.live_look = real
        self.assertEqual([None if c is None else len(c) for c in calls], [None, 3])

    def test_osc_hand_back_message(self):
        self.assertEqual(rd.osc_message("/live/song/get/tempo"), b"/live/song/get/tempo\0\0\0\0,\0\0\0")


def with_tune(al=None):
    al = al or ableton()
    al["summary"]["tune"] = [[0, 1, "Bbmaj", 2], [100, 101, "Ebmaj", 7], [200, 201, "Bmaj", 3], [300, 301, "Emaj", 8]]
    al["tune_library"] = {"Bbmaj": 2, "Bmaj": 3, "Ebmaj": 7, "Emaj": 8, "OFF | Chromatic": 0, "LOOP CUE": None}
    return al


class TestWavesTune(unittest.TestCase):
    def test_clip_names(self):
        self.assertEqual([rd.tune_key(n) for n in ("Ebmaj", "C#min", "Bmaj", "OFF | Chromatic", "LOOP CUE")],
                         ["Eb", "C#m", "B", None, None])
        self.assertTrue(rd.same_tune("G", "Em"))
        self.assertFalse(rd.same_tune("E", "Em"))

    def test_each_song_on_its_key(self):
        line = rd.eval_tune(RC, keyed(GOOD), with_tune())
        # Covered By The Blood is in B, but its song starts on the Emaj clip
        self.assertEqual(msgs(line), ["COVERED BY TH. TUNE E"])
        self.assertIn("put the Bmaj clip there instead", line.as_dict()["details"][0])
        self.assertEqual(line.info, "3 SONGS IN KEY")

    def test_a_renamed_clip_sends_the_wrong_program(self):
        al = with_tune()
        al["summary"]["tune"][2][3] = 8                       # named Bmaj, sends Emaj's program
        self.assertIn("HOLY FOREVER TUNE PGM", msgs(rd.eval_tune(RC, keyed(GOOD), al)))

    def test_no_clips(self):
        self.assertEqual(msgs(rd.eval_tune(RC, keyed(GOOD), ableton())), ["NO TUNE CLIPS"])


def with_cues(al=None):
    """Flashes over Can't Help But Praise's drums (beats 100-190 of the fixture, at 151 BPM there)."""
    al = al or ableton()
    al["summary"]["tracks"].append(track("25", "42-Drums (Live)", group="20", clips=((100, 190, None),)))
    al["summary"]["cues"] = [[104, 30], [104, 36], [108, 31], [112, 35], [116, 30], [120, 99]]
    al["summary"]["hit_src"] = [["25", 100, 190, "/Projects/NA/261018 NA Project/MultiTracks/Drums (Live).wav", False, 0.0]]
    al["light_names"] = {"30": "Blinder Flash 1", "31": "Blinder Flash 2", "35": "Blinder FX 1", "36": "Blinder FX 2"}
    return al


class TestLightTiming(unittest.TestCase):
    def tearDown(self):
        rd.MOVED.clear()
        rd.LAST.pop("light", None)

    def test_cue_times_in_the_stem(self):
        cues = rd.cue_times(RC, with_cues()["summary"])
        self.assertEqual([(c[0], c[1], c[2]) for c in cues], [(104, 30, "CAN'T HELP BUT PRAISE"), (108, 31, "CAN'T HELP BUT PRAISE"),
                                                             (112, 35, "CAN'T HELP BUT PRAISE"), (116, 30, "CAN'T HELP BUT PRAISE")])
        tempo = with_cues()["summary"]["tempo"]
        self.assertAlmostEqual(cues[0][4], rd.seconds_at(tempo, 104) - rd.seconds_at(tempo, 100))
        self.assertAlmostEqual(cues[0][6], 60 / 151 / 4, places=4)                # a 16th at 151 BPM
        self.assertAlmostEqual(cues[0][5], 0.95 * 60 / 151 / 4, places=4)         # searched out to almost a full 16th

    def test_off_cues_are_named(self):
        # [where the hit starts against the cue (s), the hit's strength, the strength right under the cue]
        hits = [[0.010, 5000, 5000],          # on a solid hit: this song's cues sit 10-12 ms before their hits
                [0.047, 5200, 1000],          # on a quiet spot, a strong hit 35 ms (against the others) later
                [-0.053, 900, 300],           # off, but a weak hit for this song: left alone
                [0.050, 5100, 4800]]          # already on a solid hit (a double hit follows): left alone
        hits[3][0] = 0.012
        line = rd.eval_light_timing(RC, with_cues(), hits)
        self.assertEqual(msgs(line), ["CUE OFF 28.1.1"])
        self.assertEqual(line.as_dict()["details"], ["CAN'T HELP BUT PRAISE bar 28.1.1: Blinder Flash 2 is 35 ms before the hit"])
        self.assertEqual(line.info, "3 CUES ON TIME")

    def test_a_double_hit_is_not_named(self):
        hits = [[0.0, 5000, 5000], [0.060, 6000, 3100], [0.0, 5000, 5000], [0.0, 5000, 5000]]
        self.assertEqual(msgs(rd.eval_light_timing(RC, with_cues(), hits)), [])

    def test_a_16th_late(self):
        hits = [[-0.09, 5000, 300], [0.037, 5200, 1000], [0.0, 5000, 5000], [0.0, 5000, 5000]]
        line = rd.eval_light_timing(RC, with_cues(), hits)
        self.assertEqual(msgs(line), ["2 CUES OFF"])
        self.assertIn("Blinder Flash 1 is 90 ms after the hit (about a 16th)", line.as_dict()["details"][0])

    def test_a_stem_it_cannot_read(self):
        line = rd.eval_light_timing(RC, with_cues(), ["unknown format: 3"] * 4)
        self.assertEqual(msgs(line), [])
        self.assertEqual(line.as_dict()["details"], ["could not read Drums (Live).wav"])

    def test_the_hold_moves_the_notes_onto_the_hit(self):
        al = with_cues()
        hits = [[-0.09, 5000, 300], [0.047, 5200, 1000], [0.0, 5000, 5000], [0.0, 5000, 5000]]
        cues = rd.cue_times(RC, al["summary"])
        off = rd.off_cues(RC, cues, hits)
        # 104 (Flash 1 and FX 2 on the same 16th) goes back a 16th onto the hit; 108 a little later, off the grid
        moves = rd.moves(RC, al["summary"], off)
        self.assertEqual([(b, n) for b, n, _ in moves], [(104, 30), (104, 36), (108, 31)])
        self.assertEqual(moves[0][2], 103.75)
        self.assertAlmostEqual(moves[2][2], 108 + 0.047 * 151 / 60, places=3)
        rd.LAST["light"] = {"key": (al["set"], al["mtime"]), "summary": al["summary"], "off": off, "live_set": "/elsewhere/261011.als"}
        sent = []

        def fake_remote(cfg, host, script, timeout=60):
            sent.append(json.loads(script.split("P = json.loads(", 1)[1].split(")\n", 1)[0].strip()[1:-1]))
            if "Save Live Set" in script:
                return {"saved": True}
            return {"done": [[b, n, 1] for b, n, _ in sent[-1]["moves"]]}
        real = (rd.run_remote, rd.hand_back)
        rd.run_remote, rd.hand_back = fake_remote, lambda cfg, host: None
        try:
            self.assertTrue(rd.run_action(CONFIG, "light_timing"))
        finally:
            rd.run_remote, rd.hand_back = real
        self.assertEqual(sent[0]["track"], "LIGHTING CUES")
        # another set is open in Live: nothing is saved, and until the set is saved the key asks for a save
        self.assertEqual(len(sent), 1)
        line = rd.eval_light_timing(RC, al, hits)
        self.assertEqual(msgs(line), ["SAVE SET (2 MOVED)"])
        # this weekend's set open in Live: moved, then saved straight away
        rd.MOVED.clear()
        rd.LAST["light"]["live_set"] = al["set"]
        rd.run_remote, rd.hand_back = fake_remote, lambda cfg, host: None
        try:
            self.assertTrue(rd.run_action(CONFIG, "light_timing"))
        finally:
            rd.run_remote, rd.hand_back = real
        self.assertEqual(sent[-1], {"set": al["set"]})
        compile(rd.SAVE_SCRIPT % json.dumps({"set": "/x.als"}), "remote", "exec")

    def test_bars_as_live_shows_them(self):
        self.assertEqual(rd.bar_of(1315.75), "329.4.4")
        self.assertEqual(rd.bar_of(0), "1.1.1")

    def test_the_scripts_run(self):
        compile(rd.HIT_SCRIPT % json.dumps({"cues": [["/x.wav", 1.5, 0.05]]}), "remote", "exec")
        compile(rd.MOVE_SCRIPT % json.dumps({"track": "LIGHTING CUES", "moves": [[104, 30, 103.75]]}), "remote", "exec")


class TestTimecode(unittest.TestCase):
    def test_seconds_through_the_tempo_map(self):
        tempo = ableton()["summary"]["tempo"]
        self.assertAlmostEqual(rd.seconds_at(tempo, 1016), 403.7086, places=3)       # Live's own warp marker
        self.assertAlmostEqual(rd.seconds_at(tempo, 2760), 1453.2191, places=3)
        self.assertAlmostEqual(rd.seconds_at({"manual": 120, "events": []}, 8), 4.0)
        self.assertAlmostEqual(rd.seconds_at({"manual": 0, "events": [[0, 60], [4, 120]]}, 4), 60 * 4 / 60 * __import__("math").log(2))
        self.assertEqual(rd.ltc_clip_info("LTC_01000000_60mins_25fps"), (3600.0, 25.0))
        self.assertEqual(rd.timecode(813.2191, 30), "00:13:33:07")

    def test_tracks_timecode_is_what_pco_says(self):
        out = [(s["title"], s["tracks"]) for s in rd.song_ltc(RC, plan(), ableton())]
        self.assertEqual(out, [("Rejoice Again", "00:01:17:26"), ("Can't Help But Praise (Live)", "00:06:43:21"),
                               ("Holy Forever", "00:13:33:07"), ("Covered By The Blood TAG", "00:24:13:07")])
        self.assertEqual(msgs(rd.eval_ltc(RC, plan(), ableton(), rd.Line())), [])

    def test_wrong_or_missing_ltc_in_pco(self):
        p = plan()
        p["items"][4]["notes"] = {}                                   # Rejoice: no LTC note
        p["items"][6]["notes"]["LTC"] = "00:13:30:00"                 # Holy Forever: moved in the tracks
        self.assertEqual(msgs(rd.eval_ltc(RC, p, ableton(), rd.Line())), ["NO LTC REJOICE AGAIN", "LTC WRONG HOLY FOREVER"])

    def test_hold_writes_the_tracks_timecode_into_pco(self):
        p = plan()
        p["items"][4]["notes"], p["items"][4]["note_ids"] = {}, {}
        p["items"][6]["notes"]["LTC"] = "00:13:30:00"
        songs = rd.song_ltc(RC, p, ableton())
        calls, real_get, real_login, real_http = [], rd.pco_get, rd.pco_login, rd.http_json
        rd.pco_login = lambda rc: ("Basic x", "275974")
        rd.pco_get = lambda auth, path, pages=10: ([{"id": "77", "attributes": {"name": "LTC"}}], [])
        rd.http_json = lambda url, headers=None, method="GET", data=None, timeout=10: calls.append(
            (method, url.split("/plans/")[1], json.loads(data)["data"]["attributes"])) or {}
        try:
            self.assertTrue(rd.write_ltc(RC, songs, "275974", "92013595"))
        finally:
            rd.pco_get, rd.pco_login, rd.http_json = real_get, real_login, real_http
        self.assertEqual(calls, [
            ("POST", "92013595/items/i-Rejoice Again/item_notes", {"content": "00:01:17:26", "item_note_category_id": "77"}),
            ("PATCH", "92013595/items/i-Holy Forever/item_notes/n-Holy Forever", {"content": "00:13:33:07"})])
        self.assertTrue(rd.write_ltc(RC, rd.song_ltc(RC, plan(), ableton()), "275974", "92013595"))   # all right: no writes


class TestLineupCrossBundle(unittest.TestCase):
    def test_lineup_against_pco(self):
        line = rd.eval_lineup(RC, plan(), lineup(["Jonah Perdue", "Krista Perdue", "DJ McLean", "Luis Decamps",
                                                  "Anthony Hoyt", "Tim Hatch", None]))
        self.assertEqual(msgs(line), ["DJ MCLEAN NOT ON PLAN", "LUIS DECAMPS DECLINED", "KEVIN MICHEL MISSING"])
        good = rd.eval_lineup(RC, plan(), lineup(["Jonah Perdue", "Krista Perdue", "Kevin Michel", "Anthony Hoyt", "Tim Hatch"]))
        self.assertEqual(msgs(good), [])
        old = rd.eval_lineup(RC, plan(), lineup(["Jonah Perdue", "Krista Perdue", "Kevin Michel", "Anthony Hoyt", "Tim Hatch"], "91"))
        self.assertEqual(msgs(old), ["OLD PLAN"])

    def test_short_names_on_the_board(self):
        p = plan()
        p["team"] += [{"name": "Alex Gaitan", "position": "Drums", "team": "Band", "status": "C"},
                      {"name": "Alexander Zinni", "position": "Guitar [Bass]", "team": "Band", "status": "C"}]
        board = ["Jonah Perdue", "KRISTA", "Kevin M", "AlexG", "Alex Zinni", "Anthony Hoyt", "Tim Hatch", "JONAH", "Luis"]
        self.assertEqual(msgs(rd.eval_lineup(RC, p, lineup(board))), ["LUIS DECLINED"])
        self.assertIsNone(rd.best_person("Alex", p["team"]))                 # two Alexes: nobody for sure
        self.assertEqual(rd.name_score("Kevin Mitchell", "Kevin Michel"), 0)

    def test_columns_follow_macro_companion_resolume(self):
        get = lambda n: COMPANION.get(n, {})
        self.assertEqual(rd.button_columns(get, "17/2/4"), {11})
        self.assertEqual(rd.button_columns(get, "17/4/3"), {11, 12})              # through two buttons
        self.assertEqual(rd.button_columns(get, "17/0/1"), set())                 # presses itself: no loop
        self.assertEqual(rd.macro_columns(MACROS, "U-LOOP", get), {11})           # a macro that runs another
        found = macros_found([it["pres"]["path"] for it in prop()["items"] if it["pres"]])
        line = rd.eval_columns(RC, prop(), found, get, composition())
        self.assertEqual((msgs(line), line.info), ([], "1 COLUMNS OK"))
        line = rd.eval_columns(RC, prop(), found, get, composition(empty=(11,)))
        self.assertEqual(msgs(line), ["COL 11 EMPTY"])
        self.assertIn("Theme 10 White", line.faults[0][2])
        self.assertIn("Holy Forever", line.faults[0][2])
        self.assertEqual(msgs(rd.eval_columns(RC, prop(), found, get, composition(ignored=(11,)))), ["COL 11 EMPTY"])
        self.assertEqual(msgs(rd.eval_columns(RC, prop(), found, get, composition(columns=10))), ["COL 11 MISSING"])
        self.assertEqual(msgs(rd.eval_columns(RC, prop(), found, get, composition("Test"))), ["WRONG DECK"])
        self.assertEqual(rd.eval_columns(RC, prop(), {"macros": {}, "used": {}}, get, composition()).info, "NO COLUMNS USED")

    def test_the_macro_reader_runs(self):
        compile(rd.MACRO_SCRIPT % json.dumps({"paths": ["/x.pro"]}), "remote", "exec")

    def test_prop_bundle_in_this_weekends_folder(self):
        rc = dict(RC, prop_bundle_folder="~/Shared drives/Location Assets/{date}")
        plan = {"time": datetime(2026, 10, 25, 9, 0).timestamp()}
        folder = rd.prop_bundle_folder(rc, plan)
        self.assertEqual(folder, "~/Shared drives/Location Assets/261025")
        self.assertEqual(msgs(rd.eval_prop_bundle(rc, [["WORSHIP BUNDLE.probundle", WEDNESDAY - 86400 * 10]], WEDNESDAY, folder)), [])
        self.assertEqual(msgs(rd.eval_prop_bundle(rc, [], WEDNESDAY, folder)), ["NOT IN 261025"])

    def test_prop_bundle_this_week(self):
        since = rd.week_start(WEDNESDAY)
        rc = dict(RC, prop_bundle_folder="~/Shared drives/Location Assets/ProPresenter")   # one folder for every week
        self.assertEqual(msgs(rd.eval_prop_bundle(rc, [["Message Templates.probundle", WEDNESDAY - 3600]], since)), [])
        self.assertEqual(msgs(rd.eval_prop_bundle(rc, [["Old.probundle", since - 86400]], since)), ["NOT SENT (LAST 10/11)"])

    def test_mics_from_pco(self):
        self.assertEqual(rd.mics_needed(RC, plan()), "2 VOX MC TIM")


class TestLook(unittest.TestCase):
    def sources(self, **over):
        src = dict(pco=plan, prop=lambda p: prop(), ableton=lambda p: ableton(),
                   lineup=lambda: lineup(["Jonah Perdue", "Krista Perdue", "Kevin Michel", "Anthony Hoyt", "Tim Hatch"]),
                   resolume=composition, prop_bundle=lambda plan: [["Message Templates.probundle", WEDNESDAY]],
                   macros=macros_found, companion=lambda n: COMPANION.get(n, {}), live=live_set)
        src.update(over)
        return src

    def test_everything_at_once(self):
        r = rd.look(CONFIG, now=WEDNESDAY, sources=self.sources())
        self.assertEqual(r["plan"]["dates"], "October 18, 2026")
        self.assertEqual(set(r["items"]), set(sp.READY_ITEMS))
        self.assertEqual(r["items"]["lineup"]["faults"], [])
        self.assertEqual(r["items"]["mics"]["info"], "2 VOX MC TIM")
        self.assertEqual(r["items"]["track_bundle"]["faults"], [["err", "NOT SENT SC"]])

    def test_one_source_down_does_not_hide_the_others(self):
        def down(*a):
            raise rd.ReadyError("PROP UNREACHABLE")
        r = rd.look(CONFIG, now=WEDNESDAY, sources=self.sources(prop=down))
        self.assertEqual(r["items"]["prop"]["faults"], [["warn", "PROP UNREACHABLE"]])
        self.assertEqual(r["items"]["lineup"]["faults"], [])
        r = rd.look(CONFIG, now=WEDNESDAY, sources=self.sources(pco=lambda: (_ for _ in ()).throw(rd.ReadyError("NO PCO LOGIN"))))
        self.assertEqual(r["items"]["tracks"]["faults"], [["warn", "NO PCO LOGIN"]])
        self.assertEqual(r["items"]["led_cross"]["faults"], [["warn", "NO PCO LOGIN"]])   # no plan, no columns to look at

    def test_a_campus_that_only_receives(self):
        cfg = json.loads(json.dumps(CONFIG))
        for k in ("locations", "prop_bundle_host", "resolume_url", "lineup_url"):
            cfg["ready"].pop(k)
        cfg["ready"]["prop_url"] = ""
        r = rd.look(cfg, now=WEDNESDAY, sources=self.sources())
        self.assertEqual(sorted(r["items"]), ["ears", "keys", "light_timing", "lights", "mics", "mix_foh", "routing", "timecode", "tracks", "tune"])   # LTC: tracks vs PCO
        engine, snap = sp.Engine(cfg, None), sp.Snapshot()
        snap.ready = r
        tiles = engine.step(snap, WEDNESDAY)
        self.assertEqual(tiles["rd_trk_out"]["x"], "TRACKS OUT\nNOT SET UP")
        self.assertEqual(tiles["rd_prop"]["x"], "PROP\nNOT SET UP")


class TestWatchdog(unittest.TestCase):
    def setUp(self):
        self.engine = sp.Engine(CONFIG, None)
        self.snap = sp.Snapshot()
        self.snap.conns_time = 0

    def step(self, now):
        self.engine.step(self.snap, now)
        return self.engine.step(self.snap, now + 5)

    def test_amber_until_due_then_red(self):
        self.snap.ready = rd.look(CONFIG, now=WEDNESDAY, sources=TestLook().sources())
        tiles = self.step(WEDNESDAY)
        self.assertEqual(tiles["rd_trk_out"]["x"], "TRACKS OUT\nNOT SENT SC")
        self.assertEqual(tiles["rd_trk_out"]["b"], sp.COLORS["warn"][0][0])
        self.assertEqual(tiles["rd_boh"]["x"], "BOH MIX\nTAP WHEN DONE")
        self.assertEqual(tiles["rd_cross"]["b"], sp.COLORS["ok"][0][0])
        self.assertTrue(tiles["ready"]["x"].startswith("READY\n"))
        self.assertIn("TO DO", tiles["ready"]["x"])
        # the checklist never reaches ALL: a weekday gap is not an alarm
        self.assertNotIn("rd_trk_out", self.engine.detail_ids)
        self.assertFalse(any(k.startswith("rd_") for k in self.engine.keys_for("all")))
        tiles = self.step(SATURDAY)
        self.assertEqual(tiles["rd_trk_out"]["b"], sp.COLORS["err"][0][0])          # due: red, blinking
        self.assertEqual(tiles["rd_boh"]["b"], sp.COLORS["err"][0][0])
        foh = {res.msg: lv for lv, res, _ in self.engine.items["rd_foh"]}
        self.assertEqual(foh, {"NOT MIXED CANT HELP BUT P.": "warn", "TO MASTER CANT HELP BUT P.": "err"})   # NOT MIXED stays a warning

    def test_tap_done_lasts_the_week(self):
        self.snap.ready = {"time": WEDNESDAY, "items": {}}
        sp.handle_commands("done rd_boh", self.engine, None, self.snap, WEDNESDAY)
        tiles = self.step(WEDNESDAY + 10)
        self.assertEqual(tiles["rd_boh"]["x"], "BOH MIX\nDONE WED 15:00")
        sp.handle_commands("done rd_boh", self.engine, None, self.snap, WEDNESDAY + 20)     # tapped again: taken back
        self.assertNotIn("mix_boh", self.engine.ready_done(WEDNESDAY + 30))
        sp.handle_commands("done rd_boh", self.engine, None, self.snap, WEDNESDAY + 40)
        next_monday = datetime(2026, 10, 19, 9, 0).timestamp()
        self.assertIn("mix_boh", self.engine.ready_done(SATURDAY))
        self.assertNotIn("mix_boh", self.engine.ready_done(next_monday))               # a new week starts clean
        self.assertIsNone(self.engine.toggle_done("rd_tracks", WEDNESDAY))              # not a tap key

    def test_one_hold_fixes_everything_that_can_be_fixed(self):
        look = rd.look(CONFIG, now=WEDNESDAY, sources=TestLook().sources(
            lineup=lambda: lineup(["DJ McLean"]), resolume=lambda: composition("Test")))
        self.snap.ready = look
        self.step(WEDNESDAY)
        plan = self.engine.fix_plan("ready", self.snap)
        fixes = sorted(what for _, kind, what, _ in plan if kind == "fix")
        self.assertEqual(fixes, ["ready_ableton_mix", "ready_copy_tracks", "ready_led_cross", "ready_lineup", "ready_ltc", "ready_recheck"])
        ran = []
        self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) or True
        self.engine.run_fixes("ready", self.snap, None, WEDNESDAY + 10)
        for t in self.engine.fix_threads:
            t.join(2)
        self.assertEqual(sorted(ran), fixes)
        self.assertEqual(self.engine.fix_plan("rd_all", self.snap), plan)              # the key on the READY page too

    def test_checking_before_the_first_look(self):
        tiles = self.step(WEDNESDAY)
        self.assertEqual(tiles["rd_tracks"]["x"], "TRACKS\nCHECKING")
        self.assertEqual(tiles["ready"]["x"].split("\n")[0], "READY")


class TestPages(unittest.TestCase):
    def actions(self, export, row, col):
        step = export["page"]["controls"][str(row)][str(col)]["steps"]["0"]["action_sets"]
        return {k: [(a["definitionId"], {o: v["value"] for o, v in a["options"].items()}) for a in v] for k, v in step.items()}

    def test_status_page_ready_key_opens_the_page(self):
        acts = self.actions(make_page.build(CONFIG), 2, 7)
        self.assertEqual(acts["up"], [("set_page", {"surfaceId": "self", "page": 21})])
        self.assertEqual(acts["2000"][0][1]["value"], "$(custom:sp_cmd);restart ready $(internal:time_unix)")

    def test_ready_page(self):
        export = make_page.build_ready(CONFIG)
        self.assertEqual(export["page"]["name"], "21\\nReady")
        self.assertEqual(self.actions(export, 0, 3)["up"][0][1]["value"], "$(custom:sp_cmd);done rd_boh")
        self.assertNotIn("2000", self.actions(export, 0, 3))                        # a tap key has no hold
        self.assertEqual(self.actions(export, 2, 1)["up"][0][1]["value"], "$(custom:sp_cmd);ack rd_lineup")
        self.assertIn("2000", self.actions(export, 2, 1))
        self.assertEqual(self.actions(export, 3, 0)["up"], [("set_page", {"surfaceId": "self", "page": 19})])
        notes = [c["options"]["notes"] for row in export["page"]["controls"].values() for c in row.values() if c and "options" in c]
        self.assertTrue(all(any(m in n for m in make_page.MARKERS) for n in notes))   # push_page may overwrite it

    def test_settings_page_applies_page_24(self):
        from settings import apply as ap
        self.assertIn("ready", ap.build_hashes(CONFIG))
        moved = json.loads(json.dumps(CONFIG))
        moved["ready_page"]["keys"][0]["pos"] = [3, 5]
        self.assertEqual(ap.what_apply_does(CONFIG, moved, ap.build_hashes(CONFIG))["pages"], ["ready"])

    def test_rules(self):
        bad = json.loads(json.dumps(CONFIG))
        bad["ready_page"]["keys"][1]["pos"] = bad["ready_page"]["keys"][0]["pos"]
        bad["ready_page"]["keys"][0]["checks"][0]["item"] = "nonsense"
        bad["ready_page"]["page"] = 19
        bad["fixes"]["ready_lineup"]["action"] = "explode"
        import schema
        errors = "\n".join(schema.validate(bad)[0])
        for expect in ("share row 0 column 1", "unknown checklist line nonsense", "cannot be the Status or Restart page",
                       "unknown READY action explode"):
            self.assertIn(expect, errors)


class TestFixes(unittest.TestCase):
    def test_lineup_sync_waits_for_the_new_pull(self):
        calls, states = [], iter([{"lastFetched": 1}, {"lastFetched": 1}, {"lastFetched": 2, "lastError": ""}])
        real = rd.http_json

        def fake(url, headers=None, method="GET", data=None, timeout=10):
            calls.append((method, url.rsplit("/", 1)[-1]))
            return {"ok": True} if method == "POST" else {"state": next(states)}
        rd.http_json = fake
        try:
            self.assertTrue(rd.run_action(CONFIG, "lineup_sync", sleep=lambda s: None))
        finally:
            rd.http_json = real
        self.assertEqual(calls[:2], [("GET", "state"), ("POST", "sync")])

    def test_cross_hold_puts_resolume_back_on_its_deck(self):
        calls = []
        real = rd.http_json

        def fake(url, headers=None, method="GET", data=None, timeout=10):
            calls.append((method, url.split("/api/v1/composition")[-1]))
            return composition("Test")
        rd.http_json = fake
        try:
            self.assertTrue(rd.run_action(CONFIG, "led_cross", sleep=lambda s: None))
        finally:
            rd.http_json = real
        self.assertIn(("POST", "/decks/1/select"), calls)
        self.assertFalse(any("/clips/" in path for _, path in calls))            # never plays a clip by itself


class TestSpotifyAndMute(unittest.TestCase):
    def test_alarms_muted_stops_spotify_too(self):
        cfg = json.loads(json.dumps(CONFIG))
        cfg["auto_fix"]["log_dir"] = __import__("tempfile").mkdtemp(prefix="sp-autolog-")
        engine, snap = sp.Engine(cfg, None), sp.Snapshot()
        wed = datetime(2026, 10, 14, 20, 20).timestamp()                     # inside Wednesday's auto-fix window
        engine.service.set(False, wed)                                       # ALARMS muted
        snap.spotify = {"running": True, "state": "paused"}
        snap.remote = {"tracks\0pgrep -xq Live": False}                      # a gentle fix that must still wait
        ran = []
        engine.executor = lambda c, fix_id, chk: ran.append(fix_id) or True
        for t in range(0, 40, 2):
            engine.step(snap, wed + t)
        engine.auto_fix(snap, None, wed + 40)
        for th in engine.fix_threads:
            th.join(2)
        self.assertEqual(ran, [])                                            # muted: nothing, not even Spotify
        engine2 = sp.Engine(cfg, None)
        engine2.service.set(True, wed)                                       # alarms on: it plays
        engine2.executor = engine.executor
        snap.remote = {}
        for t in range(0, 40, 2):
            engine2.step(snap, wed + t)
        engine2.auto_fix(snap, None, wed + 40)
        for th in engine2.fix_threads:
            th.join(2)
        self.assertEqual(ran, ["spotify_play"])


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


class TestOtherCampuses(unittest.TestCase):
    """The other locations play the same songs: their plans get North Attleboro's timecode, in its order."""
    CAMPUS = {"name": "Apollo Beach", "short": "AB", "service_type": "1113969"}

    def campus_plan(self, titles, notes=None):
        p = plan()
        p["id"], p["service_type"] = "ab-1", "1113969"
        p["items"] = [item(t, "song", notes=(notes or {}).get(t)) for t in titles]
        return p

    def test_songs_match_in_home_order_and_out_of_order_ones_are_skipped(self):
        na = rd.song_ltc(RC, plan(), ableton())
        titles = [s["title"] for s in na]
        self.assertGreaterEqual(len(titles), 3)
        # the location plays song 2 first, then 1, then 3: song 1 breaks the order and is skipped; a song NA does not play is skipped too
        cp = self.campus_plan([titles[1], titles[0], titles[2], "Something Else"])
        res = rd.campus_ltc(RC, plan(), na, self.CAMPUS, cp)
        self.assertEqual([s["title"] for s in res["songs"]], [titles[1], titles[2]])
        self.assertEqual(res["skipped"], [titles[0], "Something Else"])
        self.assertEqual(res["songs"][0]["tracks"], na[1]["tracks"])
        line = rd.eval_campus_ltc(RC, res, rd.Line())
        self.assertEqual([m for _, m, _ in line.faults], ["AB NO LTC " + rd.short(titles[1], 12), "AB NO LTC " + rd.short(titles[2], 12)])
        self.assertTrue(any("left alone" in d for d in line.as_dict()["details"]))
        right = self.campus_plan(titles, notes={t: {"LTC": s["tracks"]} for t, s in zip(titles, na)})
        self.assertEqual(rd.eval_campus_ltc(RC, rd.campus_ltc(RC, plan(), na, self.CAMPUS, right), rd.Line()).faults, [])

    def test_a_song_without_home_timecode_is_matched_but_not_faulted_or_written(self):
        """The set is not on Tracks yet: the location's songs are still in order, nothing is wrong at the location, nothing is written."""
        p = plan()
        cp = self.campus_plan([it["title"] for it in rd.plan_songs(RC, p)])
        res = rd.campus_ltc(RC, p, [], self.CAMPUS, cp)
        self.assertEqual(res["skipped"], [])
        self.assertTrue(res["songs"] and all(s["tracks"] is None for s in res["songs"]))
        self.assertEqual(rd.eval_campus_ltc(RC, res, rd.Line()).faults, [])
        self.assertIs(rd.write_ltc(RC, res["songs"], "1113969", "ab-1"), True)

    def test_the_fix_writes_every_campus(self):
        na = rd.song_ltc(RC, plan(), ableton())
        cp = self.campus_plan([s["title"] for s in na])
        res = rd.campus_ltc(RC, plan(), na, self.CAMPUS, cp)
        rd.LAST["ltc"], rd.LAST["service_type"], rd.LAST["plan_id"] = na, "275974", "92013595"
        rd.LAST["ltc_other"] = [res]
        calls = []
        real_http, real_get, real_login = rd.http_json, rd.pco_get, rd.pco_login
        rd.pco_login = lambda rc: ("Basic x", "275974")
        rd.pco_get = lambda auth, path, pages=10: ([{"id": "77", "attributes": {"name": "LTC"}}], [])
        rd.http_json = lambda url, headers=None, method="GET", data=None, timeout=10: calls.append((method, url)) or {}
        try:
            cfg = dict(CONFIG, ready=dict(RC, ltc_campuses=[self.CAMPUS]))
            self.assertTrue(rd.run_action(cfg, "ltc_to_pco", sleep=lambda s: None))
        finally:
            rd.http_json, rd.pco_get, rd.pco_login = real_http, real_get, real_login
        self.assertEqual(len([c for c in calls if "/1113969/" in c[1] and c[0] == "POST"]), len(res["songs"]))   # one note per location song


class TestRunHere(unittest.TestCase):
    def test_a_script_for_this_mac_runs_here_without_ssh(self):
        cfg = {"hosts": {"ingest": {"self": True}}}
        self.assertEqual(rd.run_remote(cfg, "ingest", "import json; print(json.dumps({'ok': 1}))"), {"ok": 1})