108 lines
4.3 KiB
Python
108 lines
4.3 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")
|
|
APP = (ROOT / "app.py").read_text(encoding="utf-8")
|
|
BRIDGE = (ROOT / "hockey_data" / "agent_bridge.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_build100_frontend_shortcut_edges_and_pending_press_are_guarded() -> None:
|
|
assert 'pressedShortcutKeys: new Set()' in APP_JS
|
|
assert 'state.pressedShortcutKeys.has(physicalKey)' in APP_JS
|
|
assert 'state.pressedShortcutKeys.delete(physicalKey)' in APP_JS
|
|
assert 'pendingShortcutSequenceRuns: new Map()' in APP_JS
|
|
assert 'if (!state.pendingShortcutSequenceRuns.has(sequence.id))' in APP_JS
|
|
assert 'one stale/broken title must not prevent the remaining title' in APP_JS
|
|
assert 'requestTimeoutMs = Math.min(60000' in APP_JS
|
|
assert 'BUILD_VERSION = "2026.08.24.4"' in APP
|
|
|
|
|
|
def test_build102_interactive_shortcut_is_dispatched_without_ack_or_mapping_lock(tmp_path: Path) -> None:
|
|
database = LocalTestDatabase(tmp_path / "build102-shortcut.sqlite3")
|
|
database.create_all()
|
|
hub = VmixAgentHub(database) # type: ignore[arg-type]
|
|
ws = FakeWebSocket()
|
|
user = HockeyUser(id="100", login="operator100", display_name="operator100")
|
|
|
|
async def scenario() -> None:
|
|
await hub.register(
|
|
ws, # type: ignore[arg-type]
|
|
{
|
|
"device_id": "GFX-BUILD100",
|
|
"device_secret": "z" * 40,
|
|
"device_name": "Build102 GFX",
|
|
"hostname": "BUILD102-PC",
|
|
"agent_version": "1.4.0",
|
|
"vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"},
|
|
},
|
|
)
|
|
await hub.pair_device("GFX-BUILD100", user)
|
|
assigned = await hub.assign_match(
|
|
wfl_user_id=user.id,
|
|
tournament_external_id="1437",
|
|
game_external_id="100100",
|
|
)
|
|
assert assigned is not None
|
|
|
|
# Simulate Mapping currently waiting for its own ACK lock. Operator shortcuts
|
|
# must bypass that lock completely in BUILD102.
|
|
mapping_lock = hub._device_vmix_send_lock("GFX-BUILD100")
|
|
await mapping_lock.acquire()
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
hub.run_vmix_sequence_for_user(
|
|
user,
|
|
[
|
|
{"Function": "OverlayInput1In", "Input": "SCORE"},
|
|
{"Function": "OverlayInput2In", "Input": "PENALTY"},
|
|
{"Function": "OverlayInput3In", "Input": "LINEUP"},
|
|
],
|
|
sequence_id="shortcut-build102",
|
|
sequence_name="Titles",
|
|
),
|
|
timeout=0.5,
|
|
)
|
|
finally:
|
|
mapping_lock.release()
|
|
|
|
commands = [item for item in ws.sent if item.get("type") == "vmix.command"]
|
|
assert [item["command"]["Input"] for item in commands] == ["SCORE", "PENALTY", "LINEUP"]
|
|
assert result["transport"] == "shortcut-no-ack"
|
|
assert result["confirmation"] == "not_waited"
|
|
assert result["requested"] == 3
|
|
assert result["attempted"] == 3
|
|
assert result["applied"] == 3
|
|
assert result["failed"] == 0
|
|
assert result["ok"] is True
|
|
assert all(row["ack_waited"] is False for row in result["results"])
|
|
assert not hub._pending_commands
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_build102_backend_marks_interactive_shortcuts_no_ack() -> None:
|
|
assert 'transport = "shortcut-no-ack" if interactive_shortcut' in BRIDGE
|
|
assert 'async with self._device_vmix_shortcut_send_lock(target_device_id)' in BRIDGE
|
|
assert '"ack_waited": False' in BRIDGE
|
|
assert '"confirmation": "not_waited" if interactive_shortcut else "ack"' in BRIDGE
|