from __future__ import annotations import asyncio from pathlib import Path from types import SimpleNamespace from sqlalchemy import select from hockey_data.agent_bridge import VmixAgentHub from hockey_data.auth_bridge import HockeyUser from hockey_data.models import OperatorSession, VmixAssignment, VmixDevice from tests.support import LocalTestDatabase class FakeWebSocket: def __init__(self) -> None: self.client = SimpleNamespace(host="127.0.0.1") self.sent: list[dict] = [] self.closed = False async def send_json(self, payload: dict) -> None: self.sent.append(payload) async def close(self, **_kwargs) -> None: self.closed = True def test_device_pairing_and_match_switch(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "agent.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() user = HockeyUser(id="42", login="operator", display_name="operator") async def scenario() -> None: registered = await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-PC-TEST", "device_secret": "x" * 40, "device_name": "Test GFX", "hostname": "TEST-PC", "agent_version": "1.0.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) assert registered["device_id"] == "GFX-PC-TEST" before = await hub.list_for_user(user) assert before["devices"][0]["pair_state"] == "free" paired = await hub.pair_device("GFX-PC-TEST", user) assert paired["paired_to_me"] is True assert paired["active_for_account"] is True with database.session() as session: session.add( OperatorSession( session_token="channel-1", wfl_user_id=user.id, login_snapshot=user.login, tournament_external_id="1437", game_external_id="902918", status="active", ) ) first = await hub.assign_current_match(user) assert first is not None assert first["game_id"] == "902918" first_assignment = first["assignment_id"] second = await hub.assign_match( wfl_user_id=user.id, tournament_external_id="1437", game_external_id="902950", ) assert second is not None assert second["game_id"] == "902950" assert second["assignment_id"] != first_assignment with database.session() as session: device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "GFX-PC-TEST")) assert device is not None assert device.current_match_external_id == "902950" assignments = list(session.scalars(select(VmixAssignment).order_by(VmixAssignment.id))) assert len(assignments) == 2 assert assignments[0].active is False assert assignments[1].active is True assert any(item.get("type") == "pairing.confirmed" for item in ws.sent) assert any(item.get("type") == "match.assign" and item.get("game_id") == "902950" for item in ws.sent) asyncio.run(scenario()) def test_set_text_command_waits_for_agent_ack(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "agent-command.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() user = HockeyUser(id="52", login="gfx", display_name="gfx") async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-PC-COMMAND", "device_secret": "z" * 40, "device_name": "Command GFX", "hostname": "COMMAND-PC", "agent_version": "1.1.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) await hub.pair_device("GFX-PC-COMMAND", user) assigned = await hub.assign_match( wfl_user_id=user.id, tournament_external_id="1437", game_external_id="902918", ) assert assigned is not None task = asyncio.create_task( hub.test_set_text( "GFX-PC-COMMAND", user, input_ref="Scorebug", selected_name="HomeTeam.Text", value="СКА", ) ) for _ in range(50): await asyncio.sleep(0) command = next((item for item in reversed(ws.sent) if item.get("type") == "vmix.command"), None) if command is not None: break assert command is not None assert command["assignment_id"] == assigned["assignment_id"] assert command["match_id"] == "902918" assert command["command"] == { "Function": "SetText", "Input": "Scorebug", "SelectedName": "HomeTeam.Text", "Value": "СКА", } await hub.receive_command_ack( "GFX-PC-COMMAND", { "type": "command.ack", "request_id": command["request_id"], "ok": True, "assignment_id": assigned["assignment_id"], "match_id": "902918", "vmix": {"connected": True, "response": "OK"}, }, ) result = await task assert result["ok"] is True assert result["function"] == "SetText" assert result["match_id"] == "902918" asyncio.run(scenario()) def test_vmix_inventory_creates_admin_mapping_profile(tmp_path: Path) -> None: from types import SimpleNamespace database = LocalTestDatabase(tmp_path / "agent-mapping.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() admin = HockeyUser(id="1", login="admin", display_name="admin", is_admin=True, role="admin") async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-MAPPING-TEST", "device_secret": "m" * 40, "device_name": "Mapping GFX", "agent_version": "1.3.0", "vmix": {"connected": True, "version": "29.0.0.48", "url": "http://127.0.0.1:8088/api/"}, }, ) fingerprint = "a" * 64 await hub.receive_inventory( "GFX-MAPPING-TEST", { "type": "vmix.inventory", "inventory": { "fingerprint": fingerprint, "vmix_version": "29.0.0.48", "inputs": [ { "key": "input-key-1", "number": "12", "title": "Scorebug", "type": "GT", "fields": [ {"name": "HomeTeam.Text", "type": "text", "index": "0"}, {"name": "HomeLogo.Source", "type": "image", "index": "0"}, ], } ], }, }, ) devices = await hub.list_mapping_devices() assert devices["devices"][0]["input_count"] == 1 assert devices["devices"][0]["field_count"] == 2 assert devices["devices"][0]["mapping"] is None profile = await hub.create_mapping_profile( SimpleNamespace(name="KHL_MAIN", device_id="GFX-MAPPING-TEST", description="Test"), admin, ) assert profile["project_fingerprint"] != fingerprint assert len(profile["project_fingerprint"]) == 64 assert profile["version"] == 1 assert any(item.get("type") == "mapping.assigned" and item.get("name") == "KHL_MAIN" for item in ws.sent) saved = await hub.replace_mapping_fields( profile["id"], SimpleNamespace(fields=[SimpleNamespace( graphic="score", data_key="game.home.name", vmix_input_key="input-key-1", vmix_input_number="12", vmix_input_title="Scorebug", vmix_field="HomeTeam.Text", field_type="text", enabled=True, )]), admin, ) assert saved["version"] == 2 assert saved["fields"][0]["vmix_field"] == "HomeTeam.Text" asyncio.run(scenario()) def test_admin_mapping_preview_uses_current_device_assignment(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "agent-mapping-preview.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() operator = HockeyUser(id="72", login="operator72", display_name="operator72") async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-MAPPING-PREVIEW", "device_secret": "p" * 40, "device_name": "Preview GFX", "agent_version": "1.3.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) await hub.pair_device("GFX-MAPPING-PREVIEW", operator) assigned = await hub.assign_match( wfl_user_id=operator.id, tournament_external_id="1437", game_external_id="902918", ) assert assigned is not None task = asyncio.create_task( hub.admin_test_mapping_value( "GFX-MAPPING-PREVIEW", input_ref="Scorebug", selected_name="HomeTeam.Text", value="СКА", field_type="text", ) ) command = None for _ in range(50): await asyncio.sleep(0) command = next((item for item in reversed(ws.sent) if item.get("type") == "vmix.command"), None) if command is not None: break assert command is not None assert command["assignment_id"] == assigned["assignment_id"] assert command["match_id"] == "902918" assert command["command"]["Function"] == "SetText" assert command["command"]["Value"] == "СКА" await hub.receive_command_ack( "GFX-MAPPING-PREVIEW", { "type": "command.ack", "request_id": command["request_id"], "ok": True, "assignment_id": assigned["assignment_id"], "match_id": "902918", "vmix": {"connected": True, "response": "OK"}, }, ) result = await task assert result["ok"] is True assert result["function"] == "SetText" asyncio.run(scenario()) def test_saved_mapping_is_resolved_and_pushed_to_vmix(tmp_path: Path) -> None: from types import SimpleNamespace database = LocalTestDatabase(tmp_path / "agent-mapping-apply.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() operator = HockeyUser(id="88", login="operator88", display_name="operator88") admin = HockeyUser(id="1", login="admin", display_name="admin", is_admin=True, role="admin") async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-MAPPING-APPLY", "device_secret": "a" * 40, "device_name": "Apply GFX", "agent_version": "1.3.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) fingerprint = "b" * 64 await hub.receive_inventory( "GFX-MAPPING-APPLY", { "type": "vmix.inventory", "inventory": { "fingerprint": fingerprint, "inputs": [{ "key": "score-key", "number": "5", "title": "Scorebug", "type": "GT", "fields": [{"name": "HomeTeam.Text", "type": "text", "index": "0"}], }], }, }, ) profile = await hub.create_mapping_profile( SimpleNamespace(name="AUTO", device_id="GFX-MAPPING-APPLY", description=""), admin ) await hub.pair_device("GFX-MAPPING-APPLY", operator) assigned = await hub.assign_match( wfl_user_id=operator.id, tournament_external_id="1437", game_external_id="902918", ) assert assigned is not None hub.mapping_data.data_catalog = lambda _user, _ctx=None: { # type: ignore[method-assign] "items": [{"key": "game.home.name", "value": "СКА", "kind": "text"}] } task = asyncio.create_task( hub.replace_mapping_fields( profile["id"], SimpleNamespace(fields=[SimpleNamespace( graphic="score", data_key="game.home.name", vmix_input_key="score-key", vmix_input_number="5", vmix_input_title="Scorebug", vmix_field="HomeTeam.Text", field_type="text", enabled=True, )]), admin, ) ) command = None for _ in range(100): await asyncio.sleep(0) command = next((item for item in reversed(ws.sent) if item.get("type") == "vmix.command"), None) if command is not None: break assert command is not None assert command["command"] == { "Function": "SetText", "Input": "score-key", "SelectedName": "HomeTeam.Text", "Value": "СКА", } await hub.receive_command_ack( "GFX-MAPPING-APPLY", { "type": "command.ack", "request_id": command["request_id"], "ok": True, "assignment_id": assigned["assignment_id"], "match_id": "902918", }, ) saved = await task assert saved["version"] == 2 asyncio.run(scenario()) def test_global_language_refresh_targets_all_active_operator_devices(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "agent-language-refresh.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] user_a = HockeyUser(id="201", login="operator-a", display_name="A") user_b = HockeyUser(id="202", login="operator-b", display_name="B") ws_a = FakeWebSocket() ws_b = FakeWebSocket() async def scenario() -> None: for device_id, secret, ws, user, game_id in ( ("GFX-LANG-A", "a" * 40, ws_a, user_a, "902918"), ("GFX-LANG-B", "b" * 40, ws_b, user_b, "902950"), ): await hub.register( ws, # type: ignore[arg-type] { "device_id": device_id, "device_secret": secret, "device_name": device_id, "agent_version": "1.3.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) await hub.pair_device(device_id, user) assigned = await hub.assign_match( wfl_user_id=user.id, tournament_external_id="1437", game_external_id=game_id, ) assert assigned is not None calls: list[tuple[str, str]] = [] async def fake_apply(device_id: str, *, reason: str = "manual") -> dict: calls.append((device_id, reason)) return {"ok": True, "device_id": device_id, "applied": 2, "total": 2} hub.apply_mapping_to_device = fake_apply # type: ignore[method-assign] result = await hub.apply_mapping_to_all_active_devices(reason="ui_language_changed") assert result["ok"] is True assert result["total_devices"] == 2 assert result["applied_devices"] == 2 assert {device_id for device_id, _ in calls} == {"GFX-LANG-A", "GFX-LANG-B"} assert all(reason == "ui_language_changed" for _, reason in calls) asyncio.run(scenario()) def test_mapping_apply_uses_operator_vmix_language_preference(tmp_path: Path) -> None: from hockey_data.models import UserPreference, VmixMappingField, VmixMappingProfile database = LocalTestDatabase(tmp_path / "agent-language-preference.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() operator = HockeyUser(id="303", login="operator-lang", display_name="Operator Lang") async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-LANG-PREF", "device_secret": "p" * 40, "device_name": "Language GFX", "agent_version": "1.3.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) fingerprint = "c" * 64 await hub.receive_inventory( "GFX-LANG-PREF", { "type": "vmix.inventory", "inventory": { "fingerprint": fingerprint, "inputs": [{ "key": "score-key", "number": "5", "title": "Scorebug", "type": "GT", "fields": [{"name": "HomeTeam.Text", "type": "text", "index": "0"}], }], }, }, ) await hub.pair_device("GFX-LANG-PREF", operator) assigned = await hub.assign_match( wfl_user_id=operator.id, tournament_external_id="1437", game_external_id="902918", ) assert assigned is not None with database.session() as session: session.add(UserPreference( wfl_user_id=operator.id, display_language="en", vmix_language="en", selected_tournament_external_id="1437", )) profile = VmixMappingProfile( name="LANG", project_fingerprint=fingerprint, active=True, version=1, created_by="admin", updated_by="admin", ) session.add(profile) session.flush() session.add(VmixMappingField( profile_id=profile.id, graphic="score", data_key="game.home.name", vmix_input_key="score-key", vmix_input_number="5", vmix_input_title="Scorebug", vmix_field="HomeTeam.Text", field_type="text", enabled=True, sort_order=0, )) captured: dict = {} def fake_catalog(_user, ctx=None): captured.update(ctx or {}) value = "Dynamo" if (ctx or {}).get("ui_language") == "en" else "Динамо" return {"items": [{"key": "game.home.name", "value": value, "kind": "text"}]} hub.mapping_data.data_catalog = fake_catalog # type: ignore[method-assign] task = asyncio.create_task(hub.apply_mapping_to_device("GFX-LANG-PREF", reason="language_switch")) command = None for _ in range(100): await asyncio.sleep(0) command = next((item for item in reversed(ws.sent) if item.get("type") == "vmix.command"), None) if command is not None: break assert command is not None assert captured["ui_language"] == "en" assert command["command"]["Value"] == "Dynamo" await hub.receive_command_ack( "GFX-LANG-PREF", { "type": "command.ack", "request_id": command["request_id"], "ok": True, "assignment_id": assigned["assignment_id"], "match_id": "902918", }, ) result = await task assert result["ok"] is True assert result["applied"] == 1 asyncio.run(scenario()) def test_auto_refresh_scheduler_targets_only_linked_due_sql_source(tmp_path: Path) -> None: from hockey_data.models import MappingSqlDataSource, VmixMappingField, VmixMappingProfile database = LocalTestDatabase(tmp_path / "agent-auto-refresh.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() operator = HockeyUser(id="404", login="clock-op", display_name="Clock") async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-AUTO-CLOCK", "device_secret": "c" * 40, "device_name": "Clock GFX", "agent_version": "1.3.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) fingerprint = "d" * 64 await hub.receive_inventory( "GFX-AUTO-CLOCK", { "type": "vmix.inventory", "inventory": { "fingerprint": fingerprint, "inputs": [{ "key": "clock-key", "number": "8", "title": "Clock", "type": "GT", "fields": [{"name": "Time.Text", "type": "text", "index": "0"}], }], }, }, ) await hub.pair_device("GFX-AUTO-CLOCK", operator) assigned = await hub.assign_match( wfl_user_id=operator.id, tournament_external_id="1437", game_external_id="902733", ) assert assigned is not None with database.session() as session: profile = VmixMappingProfile( name="CLOCK", project_fingerprint=fingerprint, active=True, version=1, created_by="admin", updated_by="admin", ) session.add(profile) session.flush() session.add(VmixMappingField( profile_id=profile.id, graphic="clock", data_key="clock.time", vmix_input_key="clock-key", vmix_input_number="8", vmix_input_title="Clock", vmix_field="Time.Text", field_type="text", enabled=True, sort_order=0, )) session.add(MappingSqlDataSource( code="clock", name="Часы", category="Система", description="", sql_text="SELECT 1 AS time", field_metadata_json="{}", enabled=True, auto_refresh_enabled=True, refresh_interval_ms=1000, sort_order=1, created_by="admin", updated_by="admin", )) # Another auto source exists but is not linked to this Mapping. session.add(MappingSqlDataSource( code="unused_auto", name="Unused", category="Система", description="", sql_text="SELECT 1 AS value", field_metadata_json="{}", enabled=True, auto_refresh_enabled=True, refresh_interval_ms=1000, sort_order=2, created_by="admin", updated_by="admin", )) calls: list[tuple[str, set[str] | None, bool]] = [] async def fake_apply(device_id: str, *, reason: str = "manual", source_codes=None, only_changed=False) -> dict: calls.append((device_id, set(source_codes or []), bool(only_changed))) return {"ok": True, "device_id": device_id, "applied": 1, "total": 1, "errors": []} hub.apply_mapping_to_device = fake_apply # type: ignore[method-assign] first = await hub.auto_refresh_once() assert first["applied"] == 1 assert calls == [("GFX-AUTO-CLOCK", {"clock"}, True)] # The 1 second source is not due again immediately. second = await hub.auto_refresh_once() assert second["applied"] == 0 assert len(calls) == 1 asyncio.run(scenario()) def test_agent_batch_transport_frame_and_ack(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "agent-batch.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-BATCH-TEST", "device_secret": "b" * 40, "device_name": "Batch GFX", "hostname": "BATCH-PC", "agent_version": "1.4.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) assert hub._agent_supports_batch("1.4.0") is True assert hub._agent_supports_batch("1.3.0") is False task = asyncio.create_task(hub.send_vmix_batch( "GFX-BATCH-TEST", assignment_id="assignment-1", match_id="902733", commands=[ {"Function": "SetText", "Input": "12", "SelectedName": "Name1.Text", "Value": "СКА"}, {"Function": "SetText", "Input": "12", "SelectedName": "Name2.Text", "Value": ""}, ], )) batch = None for _ in range(50): await asyncio.sleep(0) batch = next((item for item in reversed(ws.sent) if item.get("type") == "vmix.batch"), None) if batch is not None: break assert batch is not None assert len(batch["commands"]) == 2 assert batch["commands"][1]["Value"] == "" await hub.receive_command_ack( "GFX-BATCH-TEST", { "type": "command.batch.ack", "request_id": batch["request_id"], "ok": True, "results": [{"ok": True}, {"ok": True}], }, ) result = await task assert result["ok"] is True assert len(result["results"]) == 2 asyncio.run(scenario()) def test_runtime_vmix_sequence_preserves_command_order(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "agent-runtime-sequence.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() user = HockeyUser(id="91", login="operator91", display_name="operator91") async def scenario() -> None: await hub.register( ws, # type: ignore[arg-type] { "device_id": "GFX-PC-SEQUENCE", "device_secret": "s" * 40, "device_name": "Sequence GFX", "hostname": "SEQUENCE-PC", "agent_version": "1.4.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, }, ) await hub.pair_device("GFX-PC-SEQUENCE", user) assigned = await hub.assign_match( wfl_user_id=user.id, tournament_external_id="1437", game_external_id="902918", ) assert assigned is not None task = asyncio.create_task( hub.run_vmix_sequence_for_user( user, [ {"Function": "SetCountdown", "Input": "53", "Value": "00:18:42"}, {"Function": "StartCountdown", "Input": "53"}, ], ) ) first = None for _ in range(50): await asyncio.sleep(0) commands = [item for item in ws.sent if item.get("type") == "vmix.command"] if commands: first = commands[-1] break assert first is not None assert first["command"] == {"Function": "SetCountdown", "Input": "53", "Value": "00:18:42"} # The second command must not be emitted before the first ACK. assert len([item for item in ws.sent if item.get("type") == "vmix.command"]) == 1 await hub.receive_command_ack( "GFX-PC-SEQUENCE", {"type": "command.ack", "request_id": first["request_id"], "ok": True}, ) second = None for _ in range(50): await asyncio.sleep(0) commands = [item for item in ws.sent if item.get("type") == "vmix.command"] if len(commands) >= 2: second = commands[-1] break assert second is not None assert second["command"] == {"Function": "StartCountdown", "Input": "53"} await hub.receive_command_ack( "GFX-PC-SEQUENCE", {"type": "command.ack", "request_id": second["request_id"], "ok": True}, ) result = await task assert result["ok"] is True assert result["applied"] == 2 assert [item["function"] for item in result["results"]] == ["SetCountdown", "StartCountdown"] asyncio.run(scenario())