from __future__ import annotations import asyncio from pathlib import Path from types import SimpleNamespace from sqlalchemy import and_, select from hockey_data.agent_bridge import VmixAgentHub, _stable_vmix_inventory_fingerprint from hockey_data.auth_bridge import HockeyUser from hockey_data.models import OperatorSession, VmixAssignment, VmixDevice, VmixMappingField, VmixMappingProfile 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 _hello(device_id: str, secret: str) -> dict: return { "device_id": device_id, "device_secret": secret, "device_name": device_id, "hostname": device_id, "agent_version": "1.4.0", "vmix": {"connected": True, "url": "http://127.0.0.1:8088/api/"}, } def test_same_account_can_keep_multiple_agents_enabled_without_broadcast_assignment(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "multi-agent.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] user = HockeyUser(id="multi-1", login="operator", display_name="Operator") ws_a = FakeWebSocket() ws_b = FakeWebSocket() async def scenario() -> None: await hub.register(ws_a, _hello("GFX-MULTI-A", "a" * 40)) # type: ignore[arg-type] await hub.register(ws_b, _hello("GFX-MULTI-B", "b" * 40)) # type: ignore[arg-type] await hub.pair_device("GFX-MULTI-A", user) await hub.pair_device("GFX-MULTI-B", user) listed = await hub.list_for_user(user) assert set(listed["active_device_ids"]) == {"GFX-MULTI-A", "GFX-MULTI-B"} with database.session() as session: session.add(OperatorSession( session_token="multi-session", wfl_user_id=user.id, login_snapshot=user.login, tournament_external_id="1437", game_external_id="902918", status="active", )) # With two enabled Agents there is deliberately no implicit target. assert await hub.assign_current_match(user) is None assert not any(item.get("type") == "match.assign" for item in ws_a.sent) assert not any(item.get("type") == "match.assign" for item in ws_b.sent) result = await hub.assign_match( wfl_user_id=user.id, tournament_external_id="1437", game_external_id="902918", device_id="GFX-MULTI-B", operator_session_token="multi-session", ) assert result is not None assert result["device_id"] == "GFX-MULTI-B" assert any(item.get("type") == "match.assign" for item in ws_b.sent) assert not any(item.get("type") == "match.assign" for item in ws_a.sent) with database.session() as session: a = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "GFX-MULTI-A")) b = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "GFX-MULTI-B")) assert a is not None and b is not None assert a.is_active_for_account is True assert b.is_active_for_account is True assert a.current_match_external_id == "" assert b.current_match_external_id == "902918" await hub.deactivate_device("GFX-MULTI-A", user) listed = await hub.list_for_user(user) assert listed["active_device_ids"] == ["GFX-MULTI-B"] asyncio.run(scenario()) def test_mapping_recovers_when_only_unrelated_vmix_inputs_change(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "mapping-compatible.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] ws = FakeWebSocket() old_inputs = [ { "key": "score-key", "number": "5", "title": "Score", "type": "GT", "fields": [{"name": "Clock.Text", "type": "text", "index": "0"}], }, { "key": "old-extra-key", "number": "6", "title": "Old Extra", "type": "Video", "fields": [], }, ] new_inputs = [ { "key": "score-key", "number": "17", "title": "Score Renamed", "type": "GT", "fields": [{"name": "Clock.Text", "type": "text", "index": "4"}], }, { "key": "new-extra-key", "number": "1", "title": "New Extra", "type": "Video", "fields": [], }, ] old_stable = _stable_vmix_inventory_fingerprint(old_inputs) new_stable = _stable_vmix_inventory_fingerprint(new_inputs) assert old_stable != new_stable async def scenario() -> None: await hub.register(ws, _hello("GFX-MAP-RECOVER", "c" * 40)) # type: ignore[arg-type] with database.session() as session: profile = VmixMappingProfile( name="RECOVER-MAP", description="", project_fingerprint=old_stable, inventory_json='{"inputs":[]}', version=1, active=True, created_by="admin", updated_by="admin", ) session.add(profile) session.flush() session.add(VmixMappingField( profile_id=profile.id, graphic="score", data_key="game.clock", vmix_input_key="score-key", vmix_input_number="5", vmix_input_title="Score", vmix_field="Clock.Text", field_type="text", enabled=True, sort_order=0, )) await hub.receive_inventory("GFX-MAP-RECOVER", { "type": "vmix.inventory", "inventory": { "fingerprint": "d" * 64, "inputs": new_inputs, "vmix_version": "29.0.0.48", }, }) assigned = [item for item in ws.sent if item.get("type") == "mapping.assigned"] missing = [item for item in ws.sent if item.get("type") == "mapping.missing"] assert assigned assert assigned[-1]["name"] == "RECOVER-MAP" assert not missing with database.session() as session: device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "GFX-MAP-RECOVER")) assert device is not None resolved = hub._active_mapping_profile_for_device(session, device) assert resolved is not None assert resolved.name == "RECOVER-MAP" # Compatibility recovery must not rewrite the profile's primary # fingerprint just because this Agent has an unrelated extra Input. assert resolved.project_fingerprint == old_stable asyncio.run(scenario())