christianandClaude Opus 5.5
f179df59f1
- SONG KEYS: each song's PCO key against the set's locator, the pad under it and Finished Songs - W TUNE: the Waves Tune clip starting each song is the PCO key (and sends that key's program) - ROUTING / IN EARS: read live over AbletonOSC (then handed back to Companion); the hold, and auto-fix in its hours, put stems back on the routing template (from 261025 NA Project) and raise in-ear sends at -inf to 0 dB Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
805 lines
46 KiB
Python
805 lines
46 KiB
Python
"""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"])
|
||
|
||
|
||
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", "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})
|