Nextcue/test_rate_limit.py

#!/usr/bin/env python3
"""When Resi says "too many requests" (429), the bridge must wait, not ask again next poll.

Seen at North Attleboro: two tools polling one Resi account every second hit its rate
limit, and because neither backed off, Resi kept refusing and the ProPresenter
countdown stopped. Runs the real loop on a fake clock against a fake Resi; touches
nothing real. Takes a second.

    python3 test_rate_limit.py
"""
import io
import urllib.error

import resi_to_propresenter as R

clock = [0.0]
asks = []
sleeps = []


class Done(BaseException):
    pass


def fake_sleep(s):
    sleeps.append(s)
    clock[0] += s
    if clock[0] > 600:            # ten simulated minutes is plenty
        raise Done()


def too_many():
    return urllib.error.HTTPError("https://central.resi.io/x", 429, "Too Many Requests",
                                  {"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"}, io.BytesIO())


class LimitedResi:
    limited_until = 300           # Resi refuses for the first five minutes

    def __init__(self, *a, **k):
        self.no_cache_bust = False

    def login(self):
        pass

    def venues(self):
        return [{"uuid": "V1"}]

    def players(self, venue_uuid):
        asks.append(clock[0])
        if clock[0] < self.limited_until:
            raise too_many()
        return [{"playerHardwareId": "DEC1", "userName": "Test", "position": "00:10:00",
                 "eventId": None}]


class QuietPP:
    def __init__(self, *a, **k):
        pass

    def check(self):
        pass

    def __getattr__(self, name):
        return lambda *a, **k: True


def main():
    R.time = type("T", (), {"time": staticmethod(lambda: clock[0]),
                            "sleep": staticmethod(fake_sleep),
                            "strftime": staticmethod(lambda f: "00:00:00")})
    R.now_ms = lambda: clock[0] * 1000
    R.Resi = LimitedResi
    R.ProPresenter = QuietPP
    R.keychain_password = lambda account: "x"
    R.find_player = lambda resi, wanted: ("V1", {"playerHardwareId": "DEC1", "userName": "Test"})
    R.log = lambda msg: None
    cfg = dict(R.DEFAULTS)
    cfg.update({"resi_user": "t@example.com", "player": "DEC1", "propresenter": "http://x"})
    try:
        R.run(cfg, False)
    except Done:
        pass

    failures = []
    limited = [t for t in asks if t < LimitedResi.limited_until]
    print("asked Resi %d times in the 5 rate-limited minutes (every poll would be %d)"
          % (len(limited), int(LimitedResi.limited_until / cfg["poll_seconds"])))
    if len(limited) > 12:
        failures.append("kept asking while rate-limited: %d times" % len(limited))
    after = [t for t in asks if t >= LimitedResi.limited_until]
    if not after:
        failures.append("never asked again once Resi would answer")
    elif after[0] - LimitedResi.limited_until > 125:
        failures.append("waited %.0fs too long to recover" % (after[0] - LimitedResi.limited_until))
    steady = [b - a for a, b in zip(after, after[1:])]
    if steady and max(steady) > cfg["poll_seconds"] + 0.5:
        failures.append("did not return to the normal poll once answered (gaps %s)" % steady[:5])
    if failures:
        raise SystemExit("FAILED:\n  - " + "\n  - ".join(failures))
    print("rate limit tests passed")


if __name__ == "__main__":
    main()