111 lines
5.0 KiB
Python
111 lines
5.0 KiB
Python
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
|