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_config_list_is_compact_and_has_copy_action() -> None: assert 'data-map-copy-profile="${item.id}"' in JS assert 'title="Копировать конфиг"' in JS assert '>⧉' in JS assert '${isUsed ? "✓" : "▶"}' in JS assert 'hockey-map-profile-actions' in CSS assert 'hockey-map-profile-icon' in CSS assert 'width:28px;height:28px' in CSS assert '/profiles/${profileId}/duplicate' in JS assert 'Копия «${created.name}» создана' in JS def test_duplicate_creates_independent_inactive_profile_with_all_links(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "build76-copy.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] user = HockeyUser(id="admin-1", login="admin", display_name="Admin") inventory = { "inputs": [{ "key": "score", "number": "3", "title": "SCORE", "type": "GT", "fields": [{"name": "Home.Text", "type": "text"}, {"name": "BG.Color", "type": "color"}], }], "input_count": 1, "field_count": 2, } with database.session() as session: source = VmixMappingProfile( name="KHL MAIN", description="base package", project_fingerprint="a" * 64, inventory_json=json.dumps(inventory), version=7, active=True, created_by="other-admin", updated_by="other-admin", ) session.add(source); session.flush(); source_id = source.id session.add_all([ VmixMappingField( profile_id=source.id, graphic="score", data_key="game.home_score", vmix_input_key="score", vmix_input_number="", vmix_input_title="SCORE", vmix_field="Home.Text", field_type="text", rule_json='{"mode":"upper"}', enabled=True, sort_order=0, ), VmixMappingField( profile_id=source.id, graphic="score", data_key="home.color", vmix_input_key="score", vmix_input_number="", vmix_input_title="SCORE", vmix_field="BG.Color", field_type="color", rule_json="{}", enabled=False, sort_order=1, ), ]) first = asyncio.run(hub.duplicate_mapping_profile(source_id, user)) second = asyncio.run(hub.duplicate_mapping_profile(source_id, user)) assert first["name"] == "KHL MAIN — копия" assert second["name"] == "KHL MAIN — копия (2)" assert first["active"] is False assert first["version"] == 1 assert first["description"] == "base package" assert first["copied_from_profile_id"] == source_id assert first["copied_fields"] == 2 assert len(first["fields"]) == 2 assert first["fields"][0]["rule"] == {"mode": "upper"} assert first["fields"][1]["enabled"] is False with database.session() as session: source = session.get(VmixMappingProfile, source_id) clone = session.get(VmixMappingProfile, first["id"]) assert source.active is True assert source.version == 7 assert clone.active is False assert clone.project_fingerprint == source.project_fingerprint assert len(list(session.scalars(select(VmixMappingField).where(VmixMappingField.profile_id == clone.id)))) == 2 def test_inactive_copy_can_rescan_to_project_that_already_has_live_mapping(tmp_path: Path) -> None: database = LocalTestDatabase(tmp_path / "build76-rescan-copy.sqlite3") database.create_all() hub = VmixAgentHub(database) # type: ignore[arg-type] user = HockeyUser(id="admin-1", login="admin", display_name="Admin") old_inventory = {"inputs": [{"key": "old", "number": "1", "title": "TITLE", "fields": [{"name": "Name.Text", "type": "text"}]}]} new_inventory = {"inputs": [{"key": "new", "number": "2", "title": "TITLE", "fields": [{"name": "Name.Text", "type": "text"}, {"name": "Photo.Source", "type": "image"}]}], "input_count": 1, "field_count": 2} with database.session() as session: draft = VmixMappingProfile( name="COPY DRAFT", description="", project_fingerprint="a" * 64, inventory_json=json.dumps(old_inventory), version=1, active=False, created_by="admin", updated_by="admin", ) live = VmixMappingProfile( name="OTHER LIVE", description="", project_fingerprint="b" * 64, inventory_json=json.dumps(new_inventory), version=1, active=True, created_by="admin", updated_by="admin", ) session.add_all([draft, live]); session.flush(); draft_id = draft.id; live_id = live.id session.add(VmixMappingField( profile_id=draft.id, graphic="lt", data_key="player.name", vmix_input_key="old", vmix_input_number="", vmix_input_title="TITLE", vmix_field="Name.Text", field_type="text", rule_json="{}", enabled=True, sort_order=0, )) session.add(VmixDevice( device_uuid="GFX-COPY-76", device_secret_hash="x" * 64, name="GFX Copy", project_fingerprint="b" * 64, project_inventory_json=json.dumps(new_inventory), project_input_count=1, project_field_count=2, vmix_connected=True, )) result = asyncio.run(hub.refresh_mapping_inventory(draft_id, "GFX-COPY-76", user)) assert result["project_fingerprint"] == "b" * 64 assert result["active"] is False assert result["fields"][0]["vmix_input_key"] == "new" assert result["refresh_report"]["new_fields"] == 1 with database.session() as session: assert session.get(VmixMappingProfile, live_id).active is True def test_backend_exposes_duplicate_endpoint() -> None: assert '"/api/hockey/admin/vmix-mapping/profiles/{profile_id}/duplicate"' in BRIDGE assert 'duplicate_mapping_profile' in BRIDGE