102 lines
3.9 KiB
Python
102 lines
3.9 KiB
Python
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
|
|
from tests.timer_contract_support import assert_current_build_contract
|
|
from tests.test_build103_space_timer_fast_transport import run_transport_contract
|
|
|
|
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 test_countdown_start_hard_reseeds_before_start() -> None:
|
|
"""Keep the historical test ID; a fresh start now seeds once without render/reset wrappers."""
|
|
run_transport_contract("native_first_start")
|
|
|
|
|
|
def test_pause_is_reseeded_from_runtime_not_native_clock() -> None:
|
|
"""Keep the historical ID; BUILD159 replaces reseeding with native pause/resume."""
|
|
run_transport_contract("native_pause_resume")
|
|
|
|
|
|
def test_scoreboard_show_syncs_timer_before_overlay_steps() -> None:
|
|
run_transport_contract("scoreboard_before_overlay")
|
|
|
|
|
|
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": "SetCountdown", "Input": "SCORE", "SelectedName": "TIME.Text", "Value": "00:20:00"},
|
|
{"Function": "StartCountdown", "Input": "SCORE", "SelectedName": "TIME.Text"},
|
|
{"Function": "SetCountdown", "Input": "PENALTY", "SelectedName": "TIME.Text", "Value": "00:02:00"},
|
|
{"Function": "StopCountdown", "Input": "PENALTY", "SelectedName": "TIME.Text"},
|
|
{"Function": "StartCountdown", "Input": "PENALTY", "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] == [
|
|
"SetCountdown", "StartCountdown", "SetCountdown", "StopCountdown", "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_current_build_contract()
|
|
run_transport_contract("ordered_transport")
|