125 lines
5.0 KiB
Python
125 lines
5.0 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
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_JS = (ROOT / "ui_builder" / "static" / "app.js").read_text(encoding="utf-8")
|
|
BRIDGE = (ROOT / "hockey_data" / "agent_bridge.py").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_space_edge_is_single_fire_and_local_timer_changes_before_transport() -> None:
|
|
# One physical Space press fires once; repeats are swallowed until keyup.
|
|
assert "state.pressedShortcutKeys.has(physicalKey)" in APP_JS
|
|
assert "if (event.repeat) return false;" in APP_JS
|
|
assert "state.pressedShortcutKeys.add(physicalKey)" in APP_JS
|
|
assert "state.pressedShortcutKeys.delete(physicalKey)" in APP_JS
|
|
assert 'combo: "Space"' in APP_JS
|
|
|
|
block = APP_JS.split('case "hockey_vmix_timers_start":', 1)[1].split('case "delay":', 1)[0]
|
|
assert block.index("if (step.start_web_game)") < block.index("if (commands.length) sendRuntimeVmixTimerSequence(commands);")
|
|
assert 'delivery_mode: "timer-fast"' in APP_JS
|
|
|
|
|
|
def test_timer_fast_backend_uses_one_no_ack_batch_and_bypasses_mapping_lock(tmp_path: Path) -> None:
|
|
database = LocalTestDatabase(tmp_path / "build103-timer-fast.sqlite3")
|
|
database.create_all()
|
|
hub = VmixAgentHub(database) # type: ignore[arg-type]
|
|
ws = FakeWebSocket()
|
|
user = HockeyUser(id="103", login="operator103", display_name="operator103")
|
|
|
|
async def scenario() -> None:
|
|
await hub.register(
|
|
ws, # type: ignore[arg-type]
|
|
{
|
|
"device_id": "GFX-TIMER-FAST",
|
|
"device_secret": "t" * 40,
|
|
"device_name": "Build103 Timer GFX",
|
|
"hostname": "BUILD103-PC",
|
|
"agent_version": "1.4.0",
|
|
"vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"},
|
|
},
|
|
)
|
|
await hub.pair_device("GFX-TIMER-FAST", user)
|
|
assigned = await hub.assign_match(
|
|
wfl_user_id=user.id,
|
|
tournament_external_id="1437",
|
|
game_external_id="103103",
|
|
)
|
|
assert assigned is not None
|
|
|
|
# Mapping may be stuck waiting for ACK. Space timer transport must not wait for it.
|
|
mapping_lock = hub._device_vmix_send_lock("GFX-TIMER-FAST")
|
|
await mapping_lock.acquire()
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
hub.run_vmix_sequence_for_user(
|
|
user,
|
|
[
|
|
{"Function": "SetCountdown", "Input": "SCORE", "SelectedName": "TIME.Text", "Value": "00:18:42"},
|
|
{"Function": "StartCountdown", "Input": "SCORE", "SelectedName": "TIME.Text"},
|
|
],
|
|
delivery_mode="timer-fast",
|
|
),
|
|
timeout=0.5,
|
|
)
|
|
finally:
|
|
mapping_lock.release()
|
|
|
|
stop_result = await asyncio.wait_for(
|
|
hub.run_vmix_sequence_for_user(
|
|
user,
|
|
[
|
|
{"Function": "StopCountdown", "Input": "SCORE", "SelectedName": "TIME.Text"},
|
|
],
|
|
delivery_mode="timer-fast",
|
|
),
|
|
timeout=0.5,
|
|
)
|
|
|
|
batches = [item for item in ws.sent if item.get("type") == "vmix.batch"]
|
|
singles = [item for item in ws.sent if item.get("type") == "vmix.command"]
|
|
assert len(batches) == 1
|
|
assert [item["Function"] for item in batches[0]["commands"]] == ["SetCountdown", "StartCountdown"]
|
|
assert len(singles) == 1
|
|
assert singles[0]["command"]["Function"] == "StopCountdown"
|
|
assert result["ok"] is True
|
|
assert result["transport"] == "timer-batch-no-ack"
|
|
assert result["confirmation"] == "not_waited"
|
|
assert result["applied"] == 2
|
|
assert all(row["ack_waited"] is False for row in result["results"])
|
|
assert stop_result["ok"] is True
|
|
assert stop_result["transport"] == "timer-no-ack"
|
|
assert stop_result["confirmation"] == "not_waited"
|
|
assert stop_result["applied"] == 1
|
|
assert all(row["ack_waited"] is False for row in stop_result["results"])
|
|
assert not hub._pending_commands
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_timer_fast_contract_and_build_version() -> None:
|
|
assert 'delivery_mode: str = Field(default="", max_length=32)' in BRIDGE
|
|
assert 'fast_timer = delivery_mode == "timer-fast"' in BRIDGE
|
|
assert 'transport = "timer-batch-no-ack" if supports_batch else "timer-no-ack"' in BRIDGE
|
|
assert '"type": "vmix.batch"' in BRIDGE
|
|
assert 'BUILD_VERSION = "2026.08.24.6"' in APP
|