from __future__ import annotations import asyncio import json from pathlib import Path from sqlalchemy import select from hockey_data.agent_bridge import VmixAgentHub from hockey_data.auth_bridge import HockeyUser from hockey_data.models import VmixDevice, VmixMappingField, VmixMappingProfile from tests.support import LocalTestDatabase ROOT = Path(__file__).resolve().parents[1] JS = (ROOT / "hockey_data/static/admin-directories.js").read_text(encoding="utf-8") CSS = (ROOT / "hockey_data/static/admin-directories.css").read_text(encoding="utf-8") BRIDGE = (ROOT / "hockey_data/agent_bridge.py").read_text(encoding="utf-8") def test_mapping_ui_is_config_list_with_one_click_my_agent_action() -> None: assert "Конфиги Mapping" in JS assert "Применить к моему Agent" in JS assert "data-map-use-profile" in JS assert "/profiles/${profileId}/use" in JS assert "Импорт готового Mapping" not in JS assert "Загрузить и применить" not in JS assert "Перенести на другой Agent…" not in JS assert "Экспорт JSON" not in JS assert "hockey-map-use-profile" in CSS assert "hockey-mapping-profile-row" in CSS def test_hidden_runtime_mapping_is_reused_and_not_listed(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "build74-mapping.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] user = HockeyUser(id="admin-1", login="admin", display_name="Admin") source_inventory = { "fingerprint": "a" * 64, "inputs": [{ "key": "source-key", "number": "3", "title": "SCORE BUG", "type": "GT", "fields": [{"name": "HomeScore.Text", "type": "text", "index": "0"}], }], } target_inventory = { "fingerprint": "b" * 64, "inputs": [{ "key": "target-key", "number": "18", "title": "SCORE BUG", "type": "GT", "fields": [{"name": "HomeScore.Text", "type": "text", "index": "9"}], }], } with database.session() as session: source = VmixMappingProfile( name="Чужой KHL Mapping", description="shared config", project_fingerprint="a" * 64, inventory_json=json.dumps(source_inventory), version=4, active=True, created_by="other-admin", updated_by="other-admin", ) session.add(source) session.flush() source_id = source.id session.add(VmixMappingField( profile_id=source.id, graphic="score", data_key="game.home_score", vmix_input_key="source-key", vmix_input_number="", vmix_input_title="SCORE BUG", vmix_field="HomeScore.Text", field_type="text", rule_json="{}", enabled=True, sort_order=0, )) device = VmixDevice( device_uuid="MY-GFX-AGENT-01", device_secret_hash="x" * 64, name="My Agent", wfl_user_id="admin-1", login_snapshot="admin", is_active_for_account=True, project_fingerprint="b" * 64, project_inventory_json=json.dumps(target_inventory), project_input_count=1, project_field_count=1, vmix_connected=True, ) session.add(device) with database.session() as session: source = session.get(VmixMappingProfile, source_id) device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "MY-GFX-AGENT-01")) fields = hub._mapping_profile_payload(session, source, include_inventory=False, include_fields=True)["fields"] runtime1, report1 = hub._upsert_runtime_mapping_for_device( session, source=source, source_fields=fields, device=device, user=user, ) runtime1_id = runtime1.id assert runtime1.name.startswith(f"__AUTO_MAPPING__{source_id}__") assert report1["mapped"] == 1 mapped = session.scalar(select(VmixMappingField).where(VmixMappingField.profile_id == runtime1.id)) assert mapped.vmix_input_key == "target-key" with database.session() as session: source = session.get(VmixMappingProfile, source_id) device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "MY-GFX-AGENT-01")) fields = hub._mapping_profile_payload(session, source, include_inventory=False, include_fields=True)["fields"] runtime2, report2 = hub._upsert_runtime_mapping_for_device( session, source=source, source_fields=fields, device=device, user=user, ) assert runtime2.id == runtime1_id assert report2["mapped"] == 1 profiles = list(session.scalars(select(VmixMappingProfile).order_by(VmixMappingProfile.id))) assert len(profiles) == 2 listed = asyncio.run(hub.list_mapping_profiles()) assert [item["name"] for item in listed["profiles"]] == ["Чужой KHL Mapping"] def test_backend_has_current_agent_use_endpoint_and_hidden_runtime_name() -> None: assert '"/api/hockey/admin/vmix-mapping/profiles/{profile_id}/use"' in BRIDGE assert "use_mapping_profile_for_user" in BRIDGE assert "__AUTO_MAPPING__" in BRIDGE assert "assign_current_match" in BRIDGE