95 lines
3.4 KiB
Python
95 lines
3.4 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]
|
|
TOURNAMENT_JS = (ROOT / "hockey_data" / "static" / "tournament-menu.js").read_text(encoding="utf-8")
|
|
ROUTER = (ROOT / "hockey_data" / "router.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_visible_match_card_opens_from_schedule_without_details_roundtrip() -> None:
|
|
assert "} else if (scheduleGame) {" in TOURNAMENT_JS
|
|
assert "state.selectedGame = scheduleGame;" in TOURNAMENT_JS
|
|
assert "Opening the workspace must be immediate" in TOURNAMENT_JS
|
|
# Full details remain optional enrichment after the match is selected.
|
|
assert "void enrichOpenedGame(String(id), state.selectedId, generation);" in TOURNAMENT_JS
|
|
|
|
|
|
def test_browser_timeout_no_longer_blames_stat2tv_for_every_endpoint() -> None:
|
|
assert 'throw new Error("Превышено время ожидания ответа сервера")' in TOURNAMENT_JS
|
|
assert 'throw new Error("Превышено время ожидания ответа Stat2TV")' not in TOURNAMENT_JS
|
|
|
|
|
|
def test_session_assignment_defers_full_mapping() -> None:
|
|
assert "background_tasks: BackgroundTasks" in ROUTER
|
|
assert "wait_for_mapping=False" in ROUTER
|
|
assert "background_tasks.add_task(" in ROUTER
|
|
assert "agent_hub.apply_mapping_to_device" in ROUTER
|
|
|
|
|
|
def test_assign_match_can_return_without_running_mapping(tmp_path: Path) -> None:
|
|
database = LocalTestDatabase(tmp_path / "build109.sqlite3")
|
|
database.create_all()
|
|
hub = VmixAgentHub(database) # type: ignore[arg-type]
|
|
ws = FakeWebSocket()
|
|
user = HockeyUser(id="109", login="operator109", display_name="Operator 109")
|
|
|
|
called = False
|
|
|
|
async def slow_mapping(*_args, **_kwargs):
|
|
nonlocal called
|
|
called = True
|
|
await asyncio.sleep(30)
|
|
return {"ok": True}
|
|
|
|
async def scenario() -> None:
|
|
await hub.register(
|
|
ws, # type: ignore[arg-type]
|
|
{
|
|
"device_id": "GFX-BUILD109",
|
|
"device_secret": "x" * 40,
|
|
"device_name": "GFX BUILD109",
|
|
"hostname": "GFX-BUILD109",
|
|
"agent_version": "1.4.0",
|
|
"vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"},
|
|
},
|
|
)
|
|
await hub.pair_device("GFX-BUILD109", user)
|
|
hub.apply_mapping_to_device = slow_mapping # type: ignore[method-assign]
|
|
|
|
result = await asyncio.wait_for(
|
|
hub.assign_match(
|
|
wfl_user_id=user.id,
|
|
tournament_external_id="1437",
|
|
game_external_id="902918",
|
|
device_id="GFX-BUILD109",
|
|
wait_for_mapping=False,
|
|
),
|
|
timeout=0.5,
|
|
)
|
|
assert result is not None
|
|
assert result["delivered"] is True
|
|
assert result["mapping_apply"]["deferred"] is True
|
|
assert called is False
|
|
assert any(item.get("type") == "match.assign" for item in ws.sent)
|
|
|
|
asyncio.run(scenario())
|