поправили механику работы шорткатов
This commit is contained in:
118
tests/test_build100_shortcut_delivery_key_edge.py
Normal file
118
tests/test_build100_shortcut_delivery_key_edge.py
Normal file
@@ -0,0 +1,118 @@
|
||||
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.3"' in APP
|
||||
|
||||
|
||||
def test_build100_interactive_shortcut_is_single_command_acked_and_continues_after_failure(tmp_path: Path) -> None:
|
||||
database = LocalTestDatabase(tmp_path / "build100-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": "Build100 GFX",
|
||||
"hostname": "BUILD100-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
|
||||
|
||||
task = asyncio.create_task(
|
||||
hub.run_vmix_sequence_for_user(
|
||||
user,
|
||||
[
|
||||
{"Function": "OverlayInput1In", "Input": "SCORE"},
|
||||
{"Function": "OverlayInput2In", "Input": "BROKEN"},
|
||||
{"Function": "OverlayInput3In", "Input": "LINEUP"},
|
||||
],
|
||||
sequence_id="shortcut-build100",
|
||||
sequence_name="Titles",
|
||||
)
|
||||
)
|
||||
|
||||
seen: list[dict] = []
|
||||
for index, ok in enumerate([True, False, True]):
|
||||
command = None
|
||||
for _ in range(100):
|
||||
await asyncio.sleep(0)
|
||||
commands = [item for item in ws.sent if item.get("type") == "vmix.command"]
|
||||
if len(commands) > index:
|
||||
command = commands[index]
|
||||
break
|
||||
assert command is not None
|
||||
seen.append(command)
|
||||
await hub.receive_command_ack(
|
||||
"GFX-BUILD100",
|
||||
{
|
||||
"type": "command.ack",
|
||||
"request_id": command["request_id"],
|
||||
"ok": ok,
|
||||
"reason": "bad input" if not ok else "",
|
||||
},
|
||||
)
|
||||
|
||||
assert not [item for item in ws.sent if item.get("type") == "vmix.batch"]
|
||||
assert [item["command"]["Input"] for item in seen] == ["SCORE", "BROKEN", "LINEUP"]
|
||||
result = await task
|
||||
assert result["transport"] == "shortcut-sequential"
|
||||
assert result["requested"] == 3
|
||||
assert result["attempted"] == 3
|
||||
assert result["applied"] == 2
|
||||
assert result["failed"] == 1
|
||||
assert result["ok"] is False
|
||||
assert result["results"][2]["ok"] is True
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_build100_backend_marks_interactive_shortcuts_sequential() -> None:
|
||||
assert 'interactive_shortcut = bool' in BRIDGE
|
||||
assert 'transport = "batch" if use_batch else ("shortcut-sequential"' in BRIDGE
|
||||
assert 'if not item["ok"] and not interactive_shortcut' in BRIDGE
|
||||
Reference in New Issue
Block a user