christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d5ede0c91ea01623374075c10d10e60c892bcbd7
status-indicators/tests/test_readiness.py
T

508 lines
29 KiB
Python
Raw Normal View History

READY page: the weekly worship checklist, checked against Planning Center
2026-10-14 20:28:43 -04:00
"""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"))
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, "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", playing=True):
return {"name": {"value": "Crosses"},
"decks": [{"name": {"value": "Service"}, "selected": {"value": deck == "Service"}},
{"name": {"value": "Test"}, "selected": {"value": deck == "Test"}}],
"layers": [{"clips": [{"name": {"value": "SS_StainedGlass005"}, "connected": {"value": "Disconnected"}},
{"name": {"value": "Stained Glass 11"}, "connected": {"value": "Connected" if playing else "Disconnected"}}]}]}
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_led_cross(self):
memory = {}
self.assertEqual(msgs(rd.eval_resolume(RC, composition(), memory)), [])
self.assertEqual(memory["led_clip"], "Stained Glass 11")
self.assertEqual(msgs(rd.eval_resolume(RC, composition("Test", False), {})), ["WRONG DECK", "NOTHING ON"])
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]])
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"], []) # Resolume needs no plan
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_mics"]["x"], "MICS\n2 VOX MC TIM\nTAP WHEN SET")
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", False)))
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_comes_back_with_the_clip_last_seen(self):
calls = []
real = rd.http_json
comp = composition("Test", False)
def fake(url, headers=None, method="GET", data=None, timeout=10):
calls.append((method, url.split("/api/v1/composition")[-1]))
return comp
rd.http_json = fake
rd.MEMORY["led_clip"] = "Stained Glass 11"
try:
self.assertTrue(rd.run_action(CONFIG, "led_cross", sleep=lambda s: None))
finally:
rd.http_json = real
rd.MEMORY.clear()
self.assertIn(("POST", "/decks/1/select"), calls)
self.assertIn(("POST", "/layers/1/clips/2/connect"), calls)
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()
Reference in New Issue Copy Permalink