311 lines
16 KiB
Python
311 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
|
|
from hockey_data.agent_bridge import (
|
|
MappingConfigurationPayload, MappingFieldPayload, VmixAgentHub,
|
|
)
|
|
from hockey_data.auth_bridge import HockeyUser
|
|
from hockey_data.models import VmixAssignment, VmixDevice, VmixMappingField, VmixMappingProfile, OperatorSession
|
|
from hockey_data.match_access import MatchAccess
|
|
from tests.support import LocalTestDatabase
|
|
from tests.test_stage1_match_access import env
|
|
|
|
|
|
INVENTORY = {"inputs": [{"key": "title-key", "number": "1", "title": "Score",
|
|
"fields": [{"name": "Name.Text", "type": "text"}]}]}
|
|
|
|
|
|
def seed(db):
|
|
with db.session() as session:
|
|
profile = VmixMappingProfile(name="Shared", project_fingerprint="f" * 64,
|
|
inventory_json=json.dumps(INVENTORY), version=1)
|
|
session.add(profile)
|
|
session.flush()
|
|
session.add(VmixMappingField(profile_id=profile.id, data_key="old.name", vmix_input_key="title-key",
|
|
vmix_input_title="Score", vmix_field="Name.Text"))
|
|
for name, user, game in (("OUTPUT-A", "a", "g"), ("OUTPUT-B", "b", "g2")):
|
|
device = VmixDevice(device_uuid=name, device_secret_hash="test", wfl_user_id=user,
|
|
login_snapshot=user, project_fingerprint="f" * 64,
|
|
project_inventory_json=json.dumps(INVENTORY), vmix_connected=True,
|
|
is_active_for_account=True, current_match_external_id=game,
|
|
current_assignment_key=f"assignment-{name}")
|
|
session.add(device)
|
|
session.flush()
|
|
session.add(VmixAssignment(device_id=device.id, assignment_key=device.current_assignment_key,
|
|
wfl_user_id=user, game_external_id=game, tournament_external_id="t"))
|
|
return profile.id
|
|
|
|
|
|
def configuration(expected=1, key="new.name", **kwargs):
|
|
return MappingConfigurationPayload(expected_version=expected, name="Shared renamed", description="edited",
|
|
active=True, fields=[MappingFieldPayload(data_key=key, vmix_input_title="Score",
|
|
vmix_field="Name.Text")], **kwargs)
|
|
|
|
|
|
def mapped_key(db, profile_id):
|
|
with db.session() as session:
|
|
return session.scalar(select(VmixMappingField.data_key).where(VmixMappingField.profile_id == profile_id))
|
|
|
|
|
|
def test_template_save_freezes_legacy_devices_and_explicit_publication_updates_only_target(tmp_path, monkeypatch):
|
|
db = LocalTestDatabase(tmp_path / "publications.sqlite")
|
|
db.create_all()
|
|
profile_id = seed(db)
|
|
hub = VmixAgentHub(db)
|
|
admin = HockeyUser(id="a", login="a", display_name="a", is_admin=True)
|
|
# Neither legacy Agent has been read yet: save must retain its old links first.
|
|
result = asyncio.run(hub.save_mapping_configuration(profile_id, configuration(), admin))
|
|
assert result["version"] == 2
|
|
with db.session() as session:
|
|
a = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "OUTPUT-A"))
|
|
b = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "OUTPUT-B"))
|
|
old_a, old_b = a.mapping_profile_id, b.mapping_profile_id
|
|
assert old_a != old_b
|
|
assert a.mapping_profile_version == b.mapping_profile_version == 1
|
|
assert hub._active_mapping_profile_for_device(session, a).id == old_a
|
|
assert hub._active_mapping_profile_for_device(session, b).id == old_b
|
|
assert mapped_key(db, old_a) == mapped_key(db, old_b) == "old.name"
|
|
|
|
async def send(*_args, **_kwargs):
|
|
return True
|
|
async def apply(device_id, **_kwargs):
|
|
return {"ok": True, "device_id": device_id}
|
|
monkeypatch.setattr(hub, "send", send)
|
|
monkeypatch.setattr(hub, "apply_mapping_to_device", apply)
|
|
selected = asyncio.run(hub.use_mapping_profile_for_user(profile_id,
|
|
SimpleNamespace(device_id="OUTPUT-A", session_token="", game_id="g"), admin))
|
|
assert selected["profile"]["version"] == 2
|
|
with db.session() as session:
|
|
a = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "OUTPUT-A"))
|
|
b = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == "OUTPUT-B"))
|
|
assert a.mapping_profile_version == 2 and a.mapping_profile_id != old_a
|
|
assert b.mapping_profile_version == 1 and b.mapping_profile_id == old_b
|
|
new_a = a.mapping_profile_id
|
|
assert mapped_key(db, new_a) == "new.name"
|
|
assert mapped_key(db, old_a) == mapped_key(db, old_b) == "old.name"
|
|
db.engine.dispose()
|
|
|
|
|
|
def test_atomic_mapping_save_rejects_stale_version_and_invalid_fields_without_partial_metadata(tmp_path):
|
|
db = LocalTestDatabase(tmp_path / "conflict.sqlite")
|
|
db.create_all()
|
|
profile_id = seed(db)
|
|
hub = VmixAgentHub(db)
|
|
user = HockeyUser(id="a", login="a", display_name="a", is_admin=True)
|
|
asyncio.run(hub.save_mapping_configuration(profile_id, configuration(), user))
|
|
with pytest.raises(HTTPException) as stale:
|
|
asyncio.run(hub.save_mapping_configuration(profile_id, configuration(key="stale"), user))
|
|
assert stale.value.status_code == 409
|
|
assert mapped_key(db, profile_id) == "new.name"
|
|
invalid = configuration(expected=2)
|
|
invalid.name = "Must not commit"
|
|
invalid.fields[0].vmix_input_title = ""
|
|
with pytest.raises(HTTPException) as error:
|
|
asyncio.run(hub.save_mapping_configuration(profile_id, invalid, user))
|
|
assert error.value.status_code == 422
|
|
with db.session() as session:
|
|
profile = session.get(VmixMappingProfile, profile_id)
|
|
assert profile.version == 2 and profile.name == "Shared renamed"
|
|
db.engine.dispose()
|
|
|
|
|
|
def output_device(env, user="readonly", name="VIEW-OUTPUT", game="g"):
|
|
with env.db.session() as session:
|
|
device = VmixDevice(device_uuid=name, device_secret_hash="test", wfl_user_id=user,
|
|
is_active_for_account=True, vmix_connected=True,
|
|
current_match_external_id=game, current_assignment_key=f"assignment-{name}")
|
|
session.add(device)
|
|
session.flush()
|
|
session.add(VmixAssignment(device_id=device.id, assignment_key=device.current_assignment_key,
|
|
wfl_user_id=user, game_external_id=game, tournament_external_id="t"))
|
|
return name
|
|
|
|
|
|
def test_readonly_participant_keeps_private_output_but_stale_match_is_rejected(env, monkeypatch):
|
|
assert env.claim().status_code == 200
|
|
device = output_device(env)
|
|
calls = []
|
|
async def apply(device_id, **kwargs):
|
|
calls.append(device_id)
|
|
return {"ok": True, "applied": 1}
|
|
monkeypatch.setattr(env.hub, "apply_mapping_to_device", apply)
|
|
response = env.client.post(f"/api/hockey/agents/devices/{device}/apply-mapping",
|
|
headers=env.headers("readonly"))
|
|
assert response.status_code == 200, response.text
|
|
assert calls == [device]
|
|
stale = env.headers("readonly", game="g2")
|
|
response = env.client.post(f"/api/hockey/agents/devices/{device}/apply-mapping", headers=stale)
|
|
assert response.status_code == 409
|
|
assert calls == [device]
|
|
stale = env.headers("readonly", generation=0)
|
|
response = env.client.post(f"/api/hockey/agents/devices/{device}/apply-mapping", headers=stale)
|
|
assert response.status_code == 409
|
|
assert calls == [device]
|
|
|
|
|
|
def test_sequence_cannot_target_another_match_of_same_account(env, monkeypatch):
|
|
assert env.claim().status_code == 200
|
|
device = output_device(env, user="a", name="OTHER-GAME-OUTPUT", game="g2")
|
|
channel = env.service.open_session(wfl_user_id="a", tournament_external_id="t", game_external_id="g")
|
|
calls = []
|
|
async def send(*args, **kwargs):
|
|
calls.append(kwargs)
|
|
return {"ok": True}
|
|
monkeypatch.setattr(env.hub, "run_vmix_sequence_for_user", send)
|
|
response = env.client.post("/api/hockey/vmix/sequence", headers=env.headers(), json={
|
|
"session_token": channel["token"], "device_id": device,
|
|
"commands": [{"Function": "StartCountdown", "Input": "Score"}]})
|
|
assert response.status_code == 409
|
|
assert calls == []
|
|
|
|
|
|
def test_existing_device_schema_adds_publication_columns_without_changing_credentials(tmp_path):
|
|
from sqlalchemy import create_engine, text, MetaData, Table
|
|
from hockey_data.database import HockeyDatabase
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'legacy.sqlite'}")
|
|
new_columns = {"mapping_profile_id", "mapping_source_profile_id", "mapping_profile_version"}
|
|
legacy = Table("hockey_vmix_devices", MetaData(),
|
|
*(column._copy() for column in VmixDevice.__table__.columns if column.name not in new_columns))
|
|
legacy.create(engine)
|
|
with engine.begin() as connection:
|
|
connection.execute(legacy.insert().values(device_uuid="LEGACY-OUTPUT", device_secret_hash="keep-this-secret",
|
|
project_inventory_json=json.dumps(INVENTORY)))
|
|
database = HockeyDatabase.__new__(HockeyDatabase)
|
|
database.engine = engine
|
|
repaired = database._repair_additive_table(VmixDevice)
|
|
assert all(f"hockey_vmix_devices.{name}" in repaired for name in
|
|
("mapping_profile_id", "mapping_source_profile_id", "mapping_profile_version"))
|
|
with engine.connect() as connection:
|
|
row = connection.execute(text('SELECT * FROM hockey_vmix_devices')).mappings().one()
|
|
assert row['device_secret_hash'] == "keep-this-secret"
|
|
assert json.loads(row['project_inventory_json']) == INVENTORY
|
|
assert row['mapping_profile_id'] == row['mapping_source_profile_id'] == row['mapping_profile_version'] == 0
|
|
engine.dispose()
|
|
|
|
|
|
def test_configuration_route_requires_admin_and_version_and_commits_once(env):
|
|
profile_id = seed(env.db)
|
|
payload = configuration().model_dump()
|
|
path = f"/api/hockey/admin/vmix-mapping/profiles/{profile_id}/configuration"
|
|
denied = env.client.put(path, headers=env.headers("readonly"), json=payload)
|
|
assert denied.status_code == 403
|
|
missing = dict(payload)
|
|
missing.pop("expected_version")
|
|
assert env.client.put(path, headers=env.headers("admin"), json=missing).status_code == 422
|
|
response = env.client.put(path, headers=env.headers("admin"), json=payload)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["version"] == 2
|
|
assert env.client.put(path, headers=env.headers("admin"), json=payload).status_code == 409
|
|
|
|
|
|
def test_control_change_hook_observes_commit_and_never_runs_for_heartbeat_or_failure(env):
|
|
from fastapi import Request
|
|
from hockey_data.models import MatchControlLease
|
|
observed = []
|
|
def changed(game_id):
|
|
with env.db.session() as session:
|
|
observed.append((game_id, session.get(MatchControlLease, game_id).generation))
|
|
access = MatchAccess(env.db, on_control_change=changed)
|
|
def request(headers):
|
|
return Request({"type": "http", "headers": [(k.lower().encode(), str(v).encode()) for k,v in headers.items()]})
|
|
claimed = access.change("g", env.users['a'], request(env.headers()), "claim")
|
|
assert observed == [('g', claimed['generation'])]
|
|
access.change("g", env.users['a'], request(env.headers()), "heartbeat")
|
|
assert len(observed) == 1
|
|
with pytest.raises(HTTPException):
|
|
access.change("g", env.users['b'], request(env.headers('b')), "claim")
|
|
assert len(observed) == 1
|
|
released = access.change("g", env.users['a'], request(env.headers()), "release")
|
|
assert observed[-1] == ('g', released['generation'])
|
|
assert len(observed) == 2
|
|
|
|
|
|
def test_deleting_template_keeps_device_publications_and_removes_only_unused_snapshots(tmp_path, monkeypatch):
|
|
db = LocalTestDatabase(tmp_path / 'deleted-template.sqlite')
|
|
db.create_all()
|
|
profile_id = seed(db)
|
|
hub = VmixAgentHub(db)
|
|
with db.session() as session:
|
|
orphan = VmixMappingProfile(name=f'__AUTO_MAPPING__{profile_id}__unused', active=False,
|
|
project_fingerprint='f' * 64, version=1)
|
|
session.add(orphan)
|
|
session.flush()
|
|
orphan_id = orphan.id
|
|
session.add(VmixMappingField(profile_id=orphan_id, data_key='unused', vmix_input_title='Score',
|
|
vmix_field='Name.Text'))
|
|
messages = []
|
|
async def send(device_id, payload):
|
|
messages.append((device_id, payload))
|
|
return True
|
|
monkeypatch.setattr(hub, 'send', send)
|
|
result = asyncio.run(hub.delete_mapping_profile(profile_id))
|
|
assert result['ok']
|
|
with db.session() as session:
|
|
assert session.get(VmixMappingProfile, profile_id) is None
|
|
assert session.get(VmixMappingProfile, orphan_id) is None
|
|
assert session.scalar(select(VmixMappingField.id).where(VmixMappingField.profile_id == orphan_id)) is None
|
|
for device in session.scalars(select(VmixDevice)):
|
|
snapshot = hub._active_mapping_profile_for_device(session, device)
|
|
assert snapshot is not None and snapshot.id == device.mapping_profile_id
|
|
assert snapshot.version == device.mapping_profile_version == 1
|
|
display = hub._mapping_device_display_payload(session, snapshot)
|
|
assert display['source_deleted'] and display['name'] == 'Shared'
|
|
assert not messages
|
|
monkeypatch.setattr(hub.mapping_data, 'data_catalog', lambda *args, **kwargs:
|
|
{'items': [{'key': 'old.name', 'value': 'Publication still works'}]})
|
|
prepared = hub._prepare_mapping_to_device('OUTPUT-A', reason='device_recovered')
|
|
assert isinstance(prepared, tuple)
|
|
result, pending, *_ = prepared
|
|
assert result['profile_version'] == 1
|
|
assert pending[0]['command']['Value'] == 'Publication still works'
|
|
db.engine.dispose()
|
|
|
|
|
|
def test_inventory_refresh_freezes_legacy_links_before_rebinding(tmp_path):
|
|
db = LocalTestDatabase(tmp_path / 'inventory-publications.sqlite')
|
|
db.create_all()
|
|
profile_id = seed(db)
|
|
hub = VmixAgentHub(db)
|
|
changed_inventory = json.loads(json.dumps(INVENTORY))
|
|
changed_inventory['inputs'][0]['key'] = 'replacement-key'
|
|
with db.session() as session:
|
|
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == 'OUTPUT-A'))
|
|
device.project_inventory_json = json.dumps(changed_inventory)
|
|
user = HockeyUser(id='a', login='a', display_name='a', is_admin=True)
|
|
result = asyncio.run(hub.refresh_mapping_inventory(profile_id, 'OUTPUT-A', user))
|
|
assert result['version'] == 2
|
|
with db.session() as session:
|
|
assert session.scalar(select(VmixMappingField.vmix_input_key).where(
|
|
VmixMappingField.profile_id == profile_id)) == 'replacement-key'
|
|
for device in session.scalars(select(VmixDevice)):
|
|
assert device.mapping_profile_version == 1
|
|
assert session.scalar(select(VmixMappingField.vmix_input_key).where(
|
|
VmixMappingField.profile_id == device.mapping_profile_id)) == 'title-key'
|
|
db.engine.dispose()
|
|
|
|
|
|
def test_input_rename_keeps_published_titles_until_explicit_selection(tmp_path):
|
|
db = LocalTestDatabase(tmp_path / 'rename-publications.sqlite')
|
|
db.create_all()
|
|
profile_id = seed(db)
|
|
hub = VmixAgentHub(db)
|
|
user = HockeyUser(id='a', login='a', display_name='a', is_admin=True)
|
|
result = asyncio.run(hub.rename_mapping_input(profile_id,
|
|
SimpleNamespace(input_title='Score', title='Renamed Score', expected_version=1), user))
|
|
assert result['profile']['version'] == 2
|
|
with db.session() as session:
|
|
assert session.scalar(select(VmixMappingField.vmix_input_title).where(
|
|
VmixMappingField.profile_id == profile_id)) == 'Renamed Score'
|
|
for device in session.scalars(select(VmixDevice)):
|
|
assert device.mapping_profile_version == 1
|
|
assert session.scalar(select(VmixMappingField.vmix_input_title).where(
|
|
VmixMappingField.profile_id == device.mapping_profile_id)) == 'Score'
|
|
db.engine.dispose()
|