christianandClaude Opus 5.5
9a818f63ca
WS and SC updated to the current build (code only, their settings kept). NA's Resi cues polling slowed to 10 s / 20 s (Companion setting). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
209 lines
8.4 KiB
Python
209 lines
8.4 KiB
Python
"""prop_guard.py (2026-10-25, Southcoast): a song's timeline started again when the timecode drops, and the stage
|
|
display put on the worship layout when the first slide's macro did not, against a fake ProPresenter API."""
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
import prop_guard as pg # noqa: E402
|
|
|
|
|
|
class FakeProP:
|
|
def __init__(self):
|
|
self.t, self.running, self.slide, self.layout, self.calls, self.song = 0.0, False, 0, "MC", [], "Jehovah"
|
|
|
|
def __call__(self, path, timeout=2.0):
|
|
self.calls.append(path)
|
|
if path == "/v1/presentation/active":
|
|
return {"presentation": {"id": {"uuid": "U-" + self.song, "name": self.song}, "has_timeline": True,
|
|
"groups": [{"slides": [{}] * 10}]}} if self.song else {"presentation": None}
|
|
if path == "/v1/presentation/active/timeline":
|
|
return {"is_running": self.running, "current_time": self.t}
|
|
if path == "/v1/presentation/active/timeline/play":
|
|
self.running = True
|
|
return {}
|
|
if path == "/v1/presentation/slide_index":
|
|
return {"presentation_index": {"index": self.slide}}
|
|
if path == "/v1/stage/layout_map":
|
|
return [{"screen": {"uuid": "S0", "name": "Stage Display"}, "layout": {"name": self.layout}}]
|
|
if path == "/v1/stage/layouts":
|
|
return [{"id": {"uuid": "L1", "name": "Worship"}}, {"id": {"uuid": "L2", "name": "MC"}}]
|
|
if path.startswith("/v1/stage/screen/S0/layout/"):
|
|
self.layout = "Worship" if path.endswith("L1") else self.layout
|
|
return {}
|
|
return None
|
|
|
|
|
|
class TestGuard(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fake, self.clock = FakeProP(), [1000.0]
|
|
self.orig, self.cfg = pg.api, dict(pg.CFG)
|
|
pg.api = self.fake
|
|
pg.CFG["resume"] = True # off by default; these tests turn it on
|
|
self.g = pg.Guard(now=lambda: self.clock[0])
|
|
|
|
def tearDown(self):
|
|
pg.api = self.orig
|
|
pg.CFG.clear()
|
|
pg.CFG.update(self.cfg)
|
|
|
|
def at(self, t):
|
|
self.clock[0] = 1000.0 + t
|
|
self.g.tick()
|
|
|
|
def test_stage_put_on_worship_when_the_macro_missed(self):
|
|
song, self.fake.song = self.fake.song, None # the guard watching before the song starts
|
|
self.at(-1)
|
|
self.fake.song = song
|
|
self.at(0)
|
|
self.assertEqual(self.fake.layout, "MC") # judged only after a few seconds
|
|
self.at(3)
|
|
self.assertEqual(self.fake.layout, "Worship")
|
|
self.fake.layout = "Message" # someone picks another layout later: left alone
|
|
self.at(20)
|
|
self.assertEqual(self.fake.layout, "Message")
|
|
|
|
def test_a_timecode_drop_starts_the_timeline_again(self):
|
|
self.fake.layout = "Worship"
|
|
self.fake.running = True
|
|
self.at(0)
|
|
self.at(10)
|
|
self.fake.running, self.fake.slide = False, 4 # LTC gone mid-song
|
|
self.at(10.5)
|
|
self.assertNotIn("/v1/presentation/active/timeline/play", self.fake.calls) # a blink is not a drop
|
|
self.at(11.6)
|
|
self.assertTrue(self.fake.running)
|
|
self.assertEqual(len(self.g.state["timecode_drops"]), 1)
|
|
self.assertTrue(self.g.state["timecode_drops"][0]["resumed"])
|
|
|
|
def test_stopped_on_purpose_or_at_the_end_is_left_alone(self):
|
|
self.fake.layout, self.fake.running = "Worship", True
|
|
self.at(0)
|
|
self.fake.running, self.fake.slide = False, 4
|
|
self.at(1)
|
|
self.at(2.1) # restarted
|
|
self.fake.running = False # and stopped again at once: an operator
|
|
self.at(3)
|
|
self.at(10)
|
|
self.assertEqual(self.fake.calls.count("/v1/presentation/active/timeline/play"), 1)
|
|
self.fake.song, self.fake.running, self.fake.slide = "Owe You Praise", True, 0
|
|
self.at(20)
|
|
self.fake.running, self.fake.slide = False, 9 # its last slide: the song is over
|
|
self.at(21)
|
|
self.at(25)
|
|
self.assertEqual(self.fake.calls.count("/v1/presentation/active/timeline/play"), 1)
|
|
|
|
|
|
class TestTimecode(unittest.TestCase):
|
|
def test_timecode_lost_in_a_song_and_back(self):
|
|
fake, clock = FakeProP(), [1000.0]
|
|
fake.layout = "Worship"
|
|
orig_api, orig_ltc = pg.api, pg.Guard.ltc
|
|
ltc = {"running": True, "tc": "00:01:00:00"}
|
|
pg.api = fake
|
|
pg.Guard.ltc = lambda self: dict(ltc, time=clock[0])
|
|
try:
|
|
g = pg.Guard(now=lambda: clock[0])
|
|
def at(t):
|
|
clock[0] = 1000.0 + t
|
|
g.tick()
|
|
at(0)
|
|
at(5)
|
|
ltc["running"] = False # gone mid-song
|
|
at(6)
|
|
self.assertFalse(g.lost) # not yet 2 s
|
|
at(8.5)
|
|
self.assertTrue(g.lost)
|
|
self.assertEqual(g.state["ltc_drops"][0]["song"], "Jehovah")
|
|
ltc["running"] = True
|
|
at(10)
|
|
self.assertFalse(g.lost)
|
|
self.assertIn("back", g.state["ltc_drops"][0])
|
|
fake.song = None # between songs no timecode is fine
|
|
ltc["running"] = False
|
|
at(20)
|
|
at(30)
|
|
self.assertFalse(g.lost)
|
|
finally:
|
|
pg.api, pg.Guard.ltc = orig_api, orig_ltc
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class TestEveryLocation(unittest.TestCase):
|
|
def setUp(self):
|
|
self.orig, self.cfg = pg.api, dict(pg.CFG)
|
|
|
|
def tearDown(self):
|
|
pg.api = self.orig
|
|
pg.CFG.clear()
|
|
pg.CFG.update(self.cfg)
|
|
|
|
def test_restart_is_off_unless_a_location_turns_it_on(self):
|
|
self.assertFalse(self.cfg["resume"])
|
|
|
|
def test_the_api_port_is_found_when_the_set_one_does_not_answer(self):
|
|
import http.server
|
|
import json
|
|
import threading
|
|
|
|
class H(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({"api_version": "v1"}).encode())
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
srv = http.server.HTTPServer(("127.0.0.1", 0), H)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
try:
|
|
pg.CFG["api"] = "http://127.0.0.1:1"
|
|
port = srv.server_address[1]
|
|
self.assertEqual(pg.find_api(ports=lambda: [1, port]), "http://127.0.0.1:%d" % port)
|
|
self.assertEqual(pg.CFG["api"], "http://127.0.0.1:%d" % port)
|
|
pg.CFG["api"] = "http://127.0.0.1:1"
|
|
self.assertIsNone(pg.find_api(ports=lambda: []))
|
|
finally:
|
|
srv.shutdown()
|
|
|
|
def test_without_a_screen_called_stage_display_the_first_stage_screen_is_used(self):
|
|
pg.api = lambda path, timeout=2.0: [{"screen": {"uuid": "B", "name": "Band", "index": 1}, "layout": {"name": "X"}},
|
|
{"screen": {"uuid": "A", "name": "Front", "index": 0}, "layout": {"name": "Y"}}]
|
|
screen, layout = pg.Guard().stage()
|
|
self.assertEqual((screen["uuid"], layout["name"]), ("A", "Y"))
|
|
|
|
|
|
class TestStart(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fake, self.clock, self.orig = FakeProP(), [1000.0], pg.api
|
|
pg.api = self.fake
|
|
self.g = pg.Guard(now=lambda: self.clock[0])
|
|
|
|
def tearDown(self):
|
|
pg.api = self.orig
|
|
|
|
def at(self, t):
|
|
self.clock[0] = 1000.0 + t
|
|
self.g.tick()
|
|
|
|
def test_a_song_already_up_when_the_guard_starts_is_left_alone(self):
|
|
self.fake.calls.clear()
|
|
self.at(0) # the guard (re)starts mid-song, stage on Message
|
|
self.at(3)
|
|
self.assertEqual(self.fake.layout, "MC")
|
|
self.fake.song = "Goodness" # the next song is judged as usual
|
|
self.at(10)
|
|
self.at(13)
|
|
self.assertEqual(self.fake.layout, "Worship")
|
|
|
|
def test_prop_not_answering_at_the_start_does_not_count_as_seen(self):
|
|
pg.api = lambda path, timeout=2.0: None
|
|
self.at(0)
|
|
pg.api = self.fake
|
|
self.at(1) # first answer: this song was already up
|
|
self.at(4)
|
|
self.assertEqual(self.fake.layout, "MC")
|