179 lines
8.6 KiB
Python
179 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
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 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_portable_rebind_uses_title_when_input_key_changed() -> None:
|
|
fields = [{
|
|
"graphic": "score",
|
|
"data_key": "game.home_score",
|
|
"vmix_input_key": "old-key",
|
|
"vmix_input_number": "5",
|
|
"vmix_input_title": "SCORE BUG",
|
|
"vmix_field": "HomeScore.Text",
|
|
"field_type": "text",
|
|
"enabled": True,
|
|
"rule": {"enabled": True, "operator": "eq", "right_value": "1"},
|
|
}]
|
|
inventory = {"inputs": [{
|
|
"key": "new-key",
|
|
"number": "17",
|
|
"title": "SCORE BUG",
|
|
"type": "GT",
|
|
"fields": [{"name": "HomeScore.Text", "type": "text", "index": "0"}],
|
|
}]}
|
|
|
|
rebound, report = VmixAgentHub._rebind_portable_mapping_fields(fields, inventory)
|
|
assert report["mapped"] == 1
|
|
assert report["skipped"] == 0
|
|
assert report["matched_by"]["title"] == 1
|
|
assert rebound[0]["vmix_input_key"] == "new-key"
|
|
assert rebound[0]["vmix_input_number"] == "" # current slot must never become a saved address
|
|
assert rebound[0]["vmix_input_title"] == "SCORE BUG"
|
|
assert rebound[0]["vmix_field"] == "HomeScore.Text"
|
|
assert rebound[0]["rule"]["operator"] == "eq"
|
|
|
|
|
|
def test_portable_rebind_reports_missing_field_without_guessing() -> None:
|
|
fields = [{
|
|
"data_key": "game.clock",
|
|
"vmix_input_key": "old-key",
|
|
"vmix_input_title": "SCORE BUG",
|
|
"vmix_field": "Clock.Text",
|
|
"field_type": "text",
|
|
"enabled": True,
|
|
}]
|
|
inventory = {"inputs": [{
|
|
"key": "new-key",
|
|
"number": "2",
|
|
"title": "SCORE BUG",
|
|
"fields": [{"name": "Period.Text", "type": "text", "index": "0"}],
|
|
}]}
|
|
rebound, report = VmixAgentHub._rebind_portable_mapping_fields(fields, inventory)
|
|
assert rebound == []
|
|
assert report["mapped"] == 0
|
|
assert report["skipped"] == 1
|
|
assert report["skipped_items"][0]["reason"] == "field_not_found"
|
|
|
|
|
|
def test_copy_profile_creates_target_specific_mapping_and_keeps_source(tmp_path: Path) -> None:
|
|
database = LocalTestDatabase(tmp_path / "portable-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-score-key", "number": "3", "title": "SCORE BUG", "type": "GT",
|
|
"fields": [{"name": "HomeScore.Text", "type": "text", "index": "0"}],
|
|
}],
|
|
"input_count": 1, "field_count": 1,
|
|
}
|
|
target_inventory = {
|
|
"fingerprint": "b" * 64,
|
|
"inputs": [{
|
|
"key": "target-score-key", "number": "11", "title": "SCORE BUG", "type": "GT",
|
|
"fields": [{"name": "HomeScore.Text", "type": "text", "index": "4"}],
|
|
}],
|
|
"input_count": 1, "field_count": 1,
|
|
}
|
|
|
|
with database.session() as session:
|
|
source = VmixMappingProfile(
|
|
name="KHL MAIN", description="portable", project_fingerprint="a" * 64,
|
|
inventory_json=json.dumps(source_inventory), version=7, active=True,
|
|
created_by="admin", updated_by="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-score-key", vmix_input_number="", vmix_input_title="SCORE BUG",
|
|
vmix_field="HomeScore.Text", field_type="text", rule_json="{}", enabled=True, sort_order=0,
|
|
))
|
|
session.add(VmixDevice(
|
|
device_uuid="GFX-PORTABLE-02", device_secret_hash="x" * 64, name="GFX 2",
|
|
project_fingerprint="b" * 64, project_inventory_json=json.dumps(target_inventory),
|
|
project_input_count=1, project_field_count=1, vmix_connected=True,
|
|
))
|
|
|
|
payload = SimpleNamespace(device_id="GFX-PORTABLE-02", name="", replace_existing=False, apply_now=False)
|
|
result = asyncio.run(hub.copy_mapping_profile_to_device(source_id, payload, user))
|
|
assert result["report"]["mapped"] == 1
|
|
assert result["report"]["matched_by"]["title"] == 1
|
|
assert result["profile"]["project_fingerprint"] == "b" * 64
|
|
assert result["profile"]["fields"][0]["vmix_input_key"] == "target-score-key"
|
|
assert result["profile"]["fields"][0]["vmix_input_title"] == "SCORE BUG"
|
|
|
|
with database.session() as session:
|
|
profiles = list(session.scalars(select(VmixMappingProfile).order_by(VmixMappingProfile.id)))
|
|
visible = [profile for profile in profiles if not profile.name.startswith("__AUTO_MAPPING__")]
|
|
assert len(visible) == 2
|
|
assert visible[0].id == source_id and visible[0].active is True
|
|
assert visible[0].project_fingerprint == "a" * 64
|
|
source_links = list(session.scalars(select(VmixMappingField).where(VmixMappingField.profile_id == source_id)))
|
|
assert len(source_links) == 1 and source_links[0].vmix_input_key == "source-score-key"
|
|
assert source_links[0].vmix_input_title == "SCORE BUG" and source_links[0].vmix_input_number == ""
|
|
assert visible[1].id == result["profile"]["id"] and visible[1].active is True
|
|
assert visible[1].project_fingerprint == "b" * 64
|
|
# Publishing the copy creates one independent runtime snapshot. The
|
|
# source and copied templates remain editable without changing it.
|
|
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "GFX-PORTABLE-02"))
|
|
assert device.mapping_source_profile_id == visible[1].id
|
|
assert device.mapping_profile_version == visible[1].version
|
|
snapshot = session.get(VmixMappingProfile, device.mapping_profile_id)
|
|
assert snapshot is not None and snapshot.name.startswith("__AUTO_MAPPING__")
|
|
assert snapshot.id not in {profile.id for profile in visible}
|
|
assert len(profiles) == 3 and snapshot.project_fingerprint == "b" * 64
|
|
published = list(session.scalars(select(VmixMappingField).where(VmixMappingField.profile_id == snapshot.id)))
|
|
assert len(published) == 1
|
|
assert published[0].vmix_input_title == "SCORE BUG"
|
|
assert published[0].vmix_field == "HomeScore.Text"
|
|
|
|
|
|
def test_portable_transfer_backend_remains_available_for_compatibility() -> None:
|
|
# Build 74 intentionally simplified the operator UI, but the Build 72 API stays
|
|
# available so existing integrations/exported tools do not break.
|
|
assert "/copy-to-device" in JS
|
|
assert "/api/hockey/admin/vmix-mapping/import" in JS
|
|
assert "hockey-mapping.json" in JS
|
|
assert "portable vMix Mapping transfer/import" in CSS
|
|
assert '"/api/hockey/admin/vmix-mapping/profiles/{profile_id}/export"' in BRIDGE
|
|
assert '"/api/hockey/admin/vmix-mapping/import"' in BRIDGE
|
|
|
|
|
|
def test_portable_rebind_ignores_conflicting_legacy_selectors_and_rejects_ambiguous_titles():
|
|
source = {"data_key": "game.clock", "vmix_input_key": "reused-guid", "vmix_input_number": "5",
|
|
"vmix_input_title": "Score", "vmix_field": "Clock.Text", "enabled": True}
|
|
right = {"key": "new-guid", "number": "17", "title": "Score",
|
|
"fields": [{"name": "Clock.Text", "type": "text"}]}
|
|
wrong = {"key": "reused-guid", "number": "5", "title": "Other",
|
|
"fields": [{"name": "Clock.Text", "type": "text"}]}
|
|
rebound, report = VmixAgentHub._rebind_portable_mapping_fields([source], {"inputs": [wrong, right]})
|
|
assert report["matched_by"] == {"key": 0, "title": 1, "number": 0}
|
|
assert rebound[0]["vmix_input_title"] == "Score" and rebound[0]["vmix_input_key"] == "new-guid"
|
|
assert rebound[0]["vmix_input_number"] == ""
|
|
for inputs, reason in [([wrong], "input_title_not_found"),
|
|
([right, {**right, "key": "duplicate", "number": "18"}], "ambiguous_input_title")]:
|
|
rebound, report = VmixAgentHub._rebind_portable_mapping_fields([source], {"inputs": inputs})
|
|
assert rebound == [] and report["mapped"] == 0 and report["skipped"] == 1
|
|
assert report["skipped_items"][0]["reason"] == reason
|