"""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")) # The tests pin their own READY hours (the campus 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") 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_this_week(self): since = rd.week_start(WEDNESDAY) 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: [["Message Templates.probundle", WEDNESDAY]], macros=macros_found, companion=lambda n: COMPANION.get(n, {})) 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"]), ["lights", "mics", "mix_foh", "timecode", "tracks"]) # 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_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": 24})]) 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"], "24\\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 TestSpotifyWhileMuted(unittest.TestCase): def test_spotify_still_plays_with_alarms_muted(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, ["spotify_play"]) cfg["auto_fix"]["while_muted"] = [] # without it: nothing while muted engine2 = sp.Engine(cfg, None) engine2.service.set(False, wed) engine2.executor = engine.executor for t in range(0, 40, 2): engine2.step(snap, wed + t) self.assertEqual(engine2.auto_fix(snap, None, wed + 40), []) if __name__ == "__main__": unittest.main() class TestOtherCampuses(unittest.TestCase): """The other campuses 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 campus 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 campus's songs are still in order, nothing is wrong at the campus, 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 campus song