test 4

tests/test_build106_runtime_authoritative_timer.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+
+from hockey_data.agent_bridge import VmixAgentHub
+from hockey_data.auth_bridge import HockeyUser
+from tests.support import LocalTestDatabase
+
+ROOT = Path(__file__).resolve().parents[1]
+APP_JS = (ROOT / "ui_builder/static/app.js").read_text(encoding="utf-8")
+APP = (ROOT / "app.py").read_text(encoding="utf-8")
+
+
+class FakeWebSocket:
+    def __init__(self) -> None:
+        self.client = SimpleNamespace(host="127.0.0.1")
+        self.sent: list[dict] = []
+
+    async def send_json(self, payload: dict) -> None:
+        self.sent.append(payload)
+
+    async def close(self, **_kwargs) -> None:
+        return None
+
+
+def _block(start: str, end: str) -> str:
+    return APP_JS.split(start, 1)[1].split(end, 1)[0]
+
+
+def test_countdown_start_hard_reseeds_before_start() -> None:
+    helper = _block("function vmixCountdownSyncCommands", "function buildShortcutRuntimeContext")
+    assert 'Function: "PauseRender"' in helper
+    assert 'Function: "StopCountdown"' in helper
+    assert 'Function: "SetCountdown"' in helper
+    assert 'Function: "ResumeRender"' in helper
+    assert 'Function: "StartCountdown"' in helper
+    assert helper.index('Function: "StopCountdown"') < helper.index('Function: "SetCountdown"')
+    assert helper.rindex('Function: "SetCountdown"') < helper.rindex('Function: "StartCountdown"')
+
+
+def test_pause_is_reseeded_from_runtime_not_native_clock() -> None:
+    helper = _block("function vmixCountdownSyncCommands", "function buildShortcutRuntimeContext")
+    branch = helper.split('if (action === "stop" || action === "pause" || action === "set")', 1)[1].split('return reseedStopped;', 1)[0]
+    assert "Runtime value" in branch
+    reseed = helper.split("const reseedStopped = [", 1)[1].split("];", 1)[0]
+    assert 'Function: "StopCountdown"' in reseed
+    assert 'Function: "SetCountdown"' in reseed
+    assert 'Value: currentValue' in reseed
+
+
+def test_scoreboard_show_syncs_timer_before_overlay_steps() -> None:
+    runner = _block("async function runShortcutSequence", "function handleConfiguredShortcutCombo")
+    assert "if (sequence.is_scoreboard_sequence)" in runner
+    assert runner.index("await syncConfiguredScoreboardCountdownsToRuntime()") < runner.index("const stepErrors = []")
+    helper = _block("async function syncConfiguredScoreboardCountdownsToRuntime", "function penaltyMirrorKey")
+    assert 'timerState.running ? "start" : "set"' in helper
+    assert "await sendRuntimeVmixTimerSequence(commands)" in helper
+
+
+def test_ordered_timer_transport_sends_individual_frames_without_ack(tmp_path: Path) -> None:
+    database = LocalTestDatabase(tmp_path / "build106-ordered.sqlite3")
+    database.create_all()
+    hub = VmixAgentHub(database)  # type: ignore[arg-type]
+    ws = FakeWebSocket()
+    user = HockeyUser(id="106", login="operator106", display_name="operator106")
+
+    async def scenario() -> None:
+        await hub.register(
+            ws,  # type: ignore[arg-type]
+            {
+                "device_id": "GFX-TIMER-106",
+                "device_secret": "u" * 40,
+                "device_name": "Build106 Timer GFX",
+                "hostname": "BUILD106-PC",
+                "agent_version": "1.4.0",
+                "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"},
+            },
+        )
+        await hub.pair_device("GFX-TIMER-106", user)
+        await hub.assign_match(
+            wfl_user_id=user.id,
+            tournament_external_id="1437",
+            game_external_id="106106",
+        )
+
+        result = await asyncio.wait_for(
+            hub.run_vmix_sequence_for_user(
+                user,
+                [
+                    {"Function": "PauseRender", "Input": "SCORE"},
+                    {"Function": "StopCountdown", "Input": "SCORE", "SelectedName": "TIME.Text"},
+                    {"Function": "SetCountdown", "Input": "SCORE", "SelectedName": "TIME.Text", "Value": "00:16:01"},
+                    {"Function": "ResumeRender", "Input": "SCORE"},
+                    {"Function": "StartCountdown", "Input": "SCORE", "SelectedName": "TIME.Text"},
+                ],
+                delivery_mode="timer-fast-ordered",
+            ),
+            timeout=0.5,
+        )
+
+        singles = [item for item in ws.sent if item.get("type") == "vmix.command"]
+        batches = [item for item in ws.sent if item.get("type") == "vmix.batch"]
+        assert not batches
+        assert [item["command"]["Function"] for item in singles] == [
+            "PauseRender", "StopCountdown", "SetCountdown", "ResumeRender", "StartCountdown"
+        ]
+        assert result["ok"] is True
+        assert result["transport"] == "timer-ordered-no-ack"
+        assert result["confirmation"] == "not_waited"
+        assert all(row["ack_waited"] is False for row in result["results"])
+
+    asyncio.run(scenario())
+
+
+def test_build106_version() -> None:
+    assert 'BUILD_VERSION = "2026.08.24.9"' in APP
+    assert 'delivery_mode: "timer-fast-ordered"' in APP_JS