Nextcue/test_end_to_end.py

#!/usr/bin/env python3
"""End-to-end: the real script loop, a simulated Resi, and the REAL ProPresenter.

Everything except the Resi network calls is the shipping code path — event handling,
cue refresh, the estimator, and the ProPresenter pushes. Runs on the wall clock for
~30 seconds, then puts ProPresenter back how it found it.

    python3 test_end_to_end.py http://10.1.14.225:1025 --touch-live-propresenter

It TAKES OVER that ProPresenter's NextCue timer and stage message for ~30s (showing
"TEST Cue A/B"), so only run it when nothing is on stage, and stop the installed bridge
first -- two writers on one timer make this test fail and the stage flicker.
"""
import json
import sys
import threading
import time
import urllib.request

import resi_to_propresenter as R

if len(sys.argv) < 3 or sys.argv[2] != "--touch-live-propresenter":
    raise SystemExit(__doc__)
PP = sys.argv[1]
TIMER = "NextCue"

t0 = time.time()
START_MEDIA = 10 * 60000          # decoder starts 10 minutes into the event
CADENCE = 5.0                     # Resi republishes every 5s
LAG = 0.6                         # and each reading is 600ms stale
_published = [START_MEDIA - LAG * 1000, t0]

CUES = [
    {"name": "TEST Cue A", "ms": START_MEDIA + 10_000},
    {"name": "TEST Cue B", "ms": START_MEDIA + 22_000},
]


def true_media():
    return START_MEDIA + (time.time() - t0) * 1000


def published():
    if time.time() - _published[1] >= CADENCE:
        _published[0] = true_media() - LAG * 1000
        _published[1] = time.time()
    return _published[0]


def fmt(ms):
    ms = int(max(0, ms))
    return "%02d:%02d:%02d.%03d" % (ms // 3600000, (ms % 3600000) // 60000,
                                    (ms % 60000) // 1000, ms % 1000)


class FakeResi:
    def __init__(self, *a, **k):
        self.customer_id = "CUST"

    def login(self):
        R.log("FakeResi: signed in")

    def venues(self):
        return [{"uuid": "V1", "userName": "Test Venue"}]

    def players(self, venue_uuid):
        return [{"playerHardwareId": "DEC1", "userName": "Test Decoder",
                 "position": fmt(published()), "eventId": "EV1"}]

    def event(self, event_id):
        return {"name": "SIMULATED EVENT", "eventProfileId": "P1"}

    def cues(self, profile_id, event_id):
        return [dict(c) for c in CUES]


def pp_state():
    with urllib.request.urlopen(PP + "/v1/timers/current", timeout=8) as r:
        timers = json.loads(r.read())
    mine = next((t for t in timers if t.get("id", {}).get("name") == TIMER), None)
    try:
        with urllib.request.urlopen(PP + "/v1/stage/message", timeout=8) as r:
            msg = json.loads(r.read() or b'""')
    except Exception:
        msg = None
    return mine, msg


def main():
    R.Resi = FakeResi
    R.keychain_password = lambda account: "not-used"
    cfg = dict(R.DEFAULTS)
    cfg.update({"resi_user": "test@example.com", "player": "DEC1", "propresenter": PP,
                "timer_name": TIMER, "poll_seconds": 1.0, "resync_seconds": 30})

    before, _ = pp_state()
    print("ProPresenter before: %s\n" % (before,))

    t = threading.Thread(target=R.run, args=(cfg, False), daemon=True)
    t.start()

    failures = []
    seen_a = seen_b = False
    deadline = time.time() + 30
    while time.time() < deadline:
        time.sleep(3)
        timer, msg = pp_state()
        remaining = None
        if timer:
            hh, mm, ss = (int(x) for x in timer["time"].split(":"))
            remaining = hh * 3600 + mm * 60 + ss
        print("  t+%-4.0fs  decoder at %s   PP timer %s (%s)   message %r"
              % (time.time() - t0, fmt(true_media()),
                 timer["time"] if timer else "?", timer["state"] if timer else "?", msg))
        if msg == "TEST Cue A":
            seen_a = True
            expected = (CUES[0]["ms"] - true_media()) / 1000.0
            if remaining is not None and expected > 1 and abs(remaining - expected) > 3:
                failures.append("Cue A countdown off by %.1fs (PP said %ss, real %.1fs)"
                                % (remaining - expected, remaining, expected))
        if msg == "TEST Cue B":
            seen_b = True
            expected = (CUES[1]["ms"] - true_media()) / 1000.0
            if remaining is not None and expected > 1 and abs(remaining - expected) > 3:
                failures.append("Cue B countdown off by %.1fs" % (remaining - expected))

    if not seen_a:
        failures.append("never armed for Cue A")
    if not seen_b:
        failures.append("never rolled over to Cue B")

    print("\ncleaning up...")
    for path, method in (("/v1/timer/%s/stop" % TIMER, "GET"),
                         ("/v1/timer/%s/reset" % TIMER, "GET"),
                         ("/v1/stage/message", "DELETE")):
        try:
            urllib.request.urlopen(
                urllib.request.Request(PP + path, method=method), timeout=8).close()
        except Exception as e:
            print("  cleanup %s failed: %s" % (path, e))

    if failures:
        print("\nFAILED:")
        for f in failures:
            print("  - %s" % f)
        raise SystemExit(1)
    print("\nend-to-end OK: armed for both cues, countdown matched the decoder")


if __name__ == "__main__":
    main()