291 lines
14 KiB
Python
291 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from fastapi import WebSocketDisconnect
|
|
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
|
|
|
|
|
|
class Socket:
|
|
def __init__(self) -> None:
|
|
self.client = SimpleNamespace(host='127.0.0.1')
|
|
self.sent: list[dict] = []
|
|
self.incoming: asyncio.Queue = asyncio.Queue()
|
|
self.hub = None
|
|
self.answers = [True]
|
|
self.auto_ack = True
|
|
self.command_arrived = asyncio.Event()
|
|
|
|
async def send_json(self, message: dict) -> None:
|
|
self.sent.append(message)
|
|
if message.get('type') in {'vmix.command', 'vmix.batch'}:
|
|
self.command_arrived.set()
|
|
if self.auto_ack:
|
|
commands = message['commands'] if message['type'] == 'vmix.batch' else [message['command']]
|
|
answers = [self.answers.pop(0) if len(self.answers) > 1 else self.answers[0]
|
|
for _command in commands]
|
|
await self.hub.receive_command_ack('GFX-DELIVERY', {
|
|
'type': 'command.batch.ack' if message['type'] == 'vmix.batch' else 'command.ack',
|
|
'request_id': message['request_id'],
|
|
'assignment_id': message['assignment_id'], 'match_id': message['match_id'],
|
|
'ok': all(answers), 'reason': '' if all(answers) else 'HTTP 500',
|
|
**({'results': [{'ok': answer, 'reason': '' if answer else 'HTTP 500'}
|
|
for answer in answers]} if message['type'] == 'vmix.batch' else {})})
|
|
|
|
async def receive_json(self):
|
|
item = await self.incoming.get()
|
|
if isinstance(item, Exception):
|
|
raise item
|
|
return item
|
|
|
|
async def close(self, **_kwargs) -> None:
|
|
pass
|
|
|
|
|
|
def mapping_packets(socket):
|
|
return [packet for packet in socket.sent if packet.get('type') in {'vmix.command', 'vmix.batch'}]
|
|
|
|
|
|
def packet_commands(packet):
|
|
return packet['commands'] if packet['type'] == 'vmix.batch' else [packet['command']]
|
|
|
|
|
|
def positive_ack(packet):
|
|
return {'type': 'command.batch.ack' if packet['type'] == 'vmix.batch' else 'command.ack',
|
|
'request_id': packet['request_id'], 'assignment_id': packet['assignment_id'],
|
|
'match_id': packet['match_id'], 'ok': True,
|
|
**({'results': [{'ok': True} for _command in packet['commands']]}
|
|
if packet['type'] == 'vmix.batch' else {})}
|
|
|
|
|
|
def setup(tmp_path: Path):
|
|
database = LocalTestDatabase(tmp_path / 'mapping.sqlite3')
|
|
database.create_all()
|
|
hub = VmixAgentHub(database)
|
|
socket = Socket()
|
|
socket.hub = hub
|
|
user = HockeyUser(id='mapping-operator', login='operator', display_name='Operator')
|
|
inventory = {'inputs': [{'key': 'score-key', 'number': '1', 'title': 'Score',
|
|
'fields': [{'name': 'HOME.Text', 'type': 'text'}, {'name': 'Clock.Text', 'type': 'text'}]}]}
|
|
|
|
async def init():
|
|
await hub.register(socket, {'device_id': 'GFX-DELIVERY', 'device_secret': 'x' * 40,
|
|
'agent_version': '1.5.3', 'vmix': {'connected': True}})
|
|
await hub.pair_device('GFX-DELIVERY', user)
|
|
await hub.assign_match(wfl_user_id=user.id, game_external_id='match-A',
|
|
device_id='GFX-DELIVERY', wait_for_mapping=False)
|
|
with database.session() as session:
|
|
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == 'GFX-DELIVERY'))
|
|
device.project_fingerprint = 'a' * 64
|
|
device.project_inventory_json = json.dumps(inventory)
|
|
profile = VmixMappingProfile(name='Mapping', project_fingerprint='a' * 64,
|
|
inventory_json=json.dumps(inventory), version=1, active=True, created_by='test', updated_by='test')
|
|
session.add(profile)
|
|
session.flush()
|
|
session.add(VmixMappingField(profile_id=profile.id, graphic='score', data_key='sample.home',
|
|
vmix_input_title='Score', vmix_field='HOME.Text', field_type='text', enabled=True, sort_order=0))
|
|
hub.mapping_data.data_catalog = lambda *_args, **_kwargs: {
|
|
'items': [{'key': 'sample.home', 'value': '3', 'source_code': 'sample'}], 'sources': []}
|
|
hub._persist_score_external_snapshot = lambda *_args, **_kwargs: None
|
|
socket.sent.clear()
|
|
return database, hub, socket, user, inventory, init
|
|
|
|
|
|
def test_failed_mapping_ack_is_not_cached_and_next_refresh_retries(tmp_path):
|
|
_, hub, socket, _, _, init = setup(tmp_path)
|
|
|
|
async def scenario():
|
|
await init()
|
|
socket.answers = [False]
|
|
first = await hub.apply_mapping_to_device('GFX-DELIVERY', source_codes={'sample'}, only_changed=True)
|
|
assert not first['ok'] and not hub._mapping_value_cache
|
|
assert sum(len(packet_commands(p)) for p in mapping_packets(socket)) == 2 # one bounded retry
|
|
socket.answers = [True]
|
|
recovered = await hub.apply_mapping_to_device('GFX-DELIVERY', source_codes={'sample'}, only_changed=True)
|
|
assert recovered['ok'] and recovered['applied'] == 1
|
|
assert recovered['mapping_ack_waited'] is True
|
|
unchanged = await hub.apply_mapping_to_device('GFX-DELIVERY', source_codes={'sample'}, only_changed=True)
|
|
assert unchanged['unchanged'] == 1
|
|
await hub.stop_auto_refresh()
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_non_idempotent_command_is_never_retried(tmp_path):
|
|
_, hub, socket, _, _, init = setup(tmp_path)
|
|
|
|
async def scenario():
|
|
await init()
|
|
socket.answers = [False]
|
|
with hub.database.session() as session:
|
|
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == 'GFX-DELIVERY'))
|
|
assignment = device.current_assignment_key
|
|
result = await hub._send_mapping_entries_resilient('GFX-DELIVERY', assignment_id=assignment,
|
|
match_id='match-A', use_batch=False, entries=[{'command': {'Function': 'StartCountdown',
|
|
'Input': 'Score', 'SelectedName': 'Clock.Text'}}])
|
|
assert not result['results'][0]['ok']
|
|
assert result['retries'] == 0
|
|
assert len([p for p in socket.sent if p.get('type') == 'vmix.command']) == 1
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_late_ack_from_previous_generation_cannot_confirm_cache(tmp_path):
|
|
_, hub, socket, _, _, init = setup(tmp_path)
|
|
|
|
async def scenario():
|
|
await init()
|
|
socket.auto_ack = False
|
|
pending = asyncio.create_task(hub.apply_mapping_to_device('GFX-DELIVERY', source_codes={'sample'}))
|
|
await asyncio.wait_for(socket.command_arrived.wait(), 3)
|
|
packet = mapping_packets(socket)[0]
|
|
hub._invalidate_mapping_delivery('GFX-DELIVERY')
|
|
await hub.receive_command_ack('GFX-DELIVERY', positive_ack(packet))
|
|
result = await pending
|
|
assert not result['ok'] and result['applied'] == 0
|
|
assert not hub._mapping_value_cache
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_large_mapping_queues_one_small_batch_and_urgent_command_can_interleave(tmp_path):
|
|
_, hub, socket, _, _, init = setup(tmp_path)
|
|
|
|
async def scenario():
|
|
await init()
|
|
socket.auto_ack = False
|
|
with hub.database.session() as session:
|
|
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == 'GFX-DELIVERY'))
|
|
assignment = device.current_assignment_key
|
|
entries = [{'command': {'Function': 'SetText', 'Input': 'Score', 'SelectedName': f'Field{n}', 'Value': str(n)}} for n in range(200)]
|
|
task = asyncio.create_task(hub._send_mapping_entries_resilient('GFX-DELIVERY', assignment_id=assignment,
|
|
match_id='match-A', entries=entries, use_batch=True))
|
|
await asyncio.wait_for(socket.command_arrived.wait(), 3)
|
|
await asyncio.sleep(0)
|
|
assert len(mapping_packets(socket)) == 1
|
|
original = mapping_packets(socket)[0]
|
|
assert original['type'] == 'vmix.batch' and len(original['commands']) == 6
|
|
assert len(hub._pending_commands) == 1
|
|
await hub.send('GFX-DELIVERY', {'type': 'vmix.command', 'command': {'Function': 'PauseCountdown'}})
|
|
assert socket.sent[-1]['command']['Function'] == 'PauseCountdown'
|
|
hub._invalidate_mapping_delivery('GFX-DELIVERY')
|
|
await hub.receive_command_ack('GFX-DELIVERY', positive_ack(original))
|
|
result = await task
|
|
assert len(result['results']) == 200
|
|
assert len(mapping_packets(socket)) == 2
|
|
assert sum(len(packet_commands(p)) for p in mapping_packets(socket)) == 7
|
|
assert not hub._pending_commands
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_reader_accepts_ack_while_status_worker_is_blocked(tmp_path):
|
|
_, hub, socket, _, _, init = setup(tmp_path)
|
|
|
|
async def scenario():
|
|
await init()
|
|
entered = asyncio.Event()
|
|
release = asyncio.Event()
|
|
async def slow_status(*_args):
|
|
entered.set()
|
|
await release.wait()
|
|
async def reconcile(*_args):
|
|
pass
|
|
hub.receive_status = slow_status
|
|
hub.reconcile_device = reconcile
|
|
socket.auto_ack = False
|
|
future = asyncio.get_running_loop().create_future()
|
|
hub._pending_commands['inflight-test'] = ('GFX-DELIVERY', future)
|
|
reader = asyncio.create_task(hub.read_agent_messages('GFX-DELIVERY', socket))
|
|
await socket.incoming.put({'type': 'vmix.status', 'vmix': {'connected': True}})
|
|
await asyncio.wait_for(entered.wait(), 3)
|
|
await socket.incoming.put({'type': 'command.ack', 'request_id': 'inflight-test', 'ok': True})
|
|
assert (await asyncio.wait_for(future, 1))['ok'] is True
|
|
await socket.incoming.put({'type': 'heartbeat', 'vmix': {'connected': True}})
|
|
for _ in range(20):
|
|
if any(p.get('type') == 'heartbeat.ack' for p in socket.sent):
|
|
break
|
|
await asyncio.sleep(0)
|
|
assert any(p.get('type') == 'heartbeat.ack' for p in socket.sent)
|
|
release.set()
|
|
await socket.incoming.put(WebSocketDisconnect())
|
|
try:
|
|
await reader
|
|
except WebSocketDisconnect:
|
|
pass
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_recovery_preserves_native_countdown_fields_and_reconnect_waits_inventory(tmp_path):
|
|
database, hub, socket, _, inventory, init = setup(tmp_path)
|
|
|
|
async def scenario():
|
|
await init()
|
|
with database.session() as session:
|
|
profile = session.scalar(select(VmixMappingProfile))
|
|
session.add(VmixMappingField(profile_id=profile.id, graphic='clock', data_key='sample.clock',
|
|
vmix_input_title='Score', vmix_field='Clock.Text', field_type='text', enabled=True, sort_order=1))
|
|
hub.mapping_data.data_catalog = lambda *_a, **_kw: {'items': [
|
|
{'key': 'sample.home', 'value': '3', 'source_code': 'sample'},
|
|
{'key': 'sample.clock', 'value': '20:00', 'source_code': 'sample'}], 'sources': []}
|
|
hub._remember_native_countdown('GFX-DELIVERY', {'Function': 'StartCountdown', 'Input': 'score-key', 'SelectedName': 'Clock.Text'})
|
|
await hub.reconcile_device('GFX-DELIVERY')
|
|
await asyncio.sleep(0)
|
|
assert not mapping_packets(socket)
|
|
await hub.receive_inventory('GFX-DELIVERY', {'type': 'vmix.inventory',
|
|
'inventory': {**inventory, 'fingerprint': 'b' * 64}})
|
|
await hub._mapping_recovery_jobs['GFX-DELIVERY']
|
|
commands = [command for p in mapping_packets(socket) for command in packet_commands(p)]
|
|
assert commands and all(p.get('SelectedName') != 'Clock.Text' for p in commands)
|
|
assert all(p['Function'] not in {'SetCountdown', 'StartCountdown', 'StopCountdown'} for p in commands)
|
|
hub._device_status['GFX-DELIVERY'] = {'vmix_connected': True}
|
|
await hub.receive_status('GFX-DELIVERY', {'type': 'vmix.status', 'vmix': {'connected': False}})
|
|
assert not hub._mapping_value_cache
|
|
await hub.receive_status('GFX-DELIVERY', {'type': 'vmix.status', 'vmix': {'connected': True}})
|
|
await hub._mapping_recovery_jobs['GFX-DELIVERY']
|
|
assert len(mapping_packets(socket)) == 2
|
|
await hub.stop_auto_refresh()
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_configured_timer_targets_are_protected_after_server_restart(tmp_path):
|
|
_, hub, _, _, inventory, _ = setup(tmp_path)
|
|
local = tmp_path / 'settings' / 'local'
|
|
local.mkdir(parents=True)
|
|
published = {'shortcut_sequences': [{'enabled': True, 'steps': [{
|
|
'type': 'hockey_vmix_timers_start', 'enabled': True,
|
|
'game_vmix_mode': 'countdown', 'sync_vmix_game': True,
|
|
'game_vmix_input': 'score-key', 'game_vmix_selected_name': 'Clock.Text',
|
|
'sync_vmix_penalties': True, 'home_penalty_targets': [
|
|
{'input': 'Score', 'selected_name': 'Penalty.Text'}],
|
|
'away_penalty_targets': []}]}]}
|
|
(local / 'ui_builder_published.json').write_text(json.dumps(published))
|
|
hub.mapping_data.settings = SimpleNamespace(settings_dir=local.parent)
|
|
assert hub._protected_timer_targets('GFX-DELIVERY', inventory['inputs']) == {
|
|
('Score', 'Clock.Text'), ('Score', 'Penalty.Text')}
|
|
|
|
|
|
def test_timeout_does_not_leave_pending_ack_or_accept_late_response(tmp_path):
|
|
_, hub, socket, _, _, init = setup(tmp_path)
|
|
|
|
async def scenario():
|
|
await init()
|
|
socket.auto_ack = False
|
|
try:
|
|
await hub.send_vmix_command('GFX-DELIVERY', assignment_id='assignment', match_id='match-A',
|
|
command={'Function': 'SetText', 'Input': 'Score', 'SelectedName': 'HOME.Text', 'Value': '3'},
|
|
timeout=0.01)
|
|
except Exception as error:
|
|
assert getattr(error, 'status_code', None) == 504
|
|
else:
|
|
raise AssertionError('Missing ACK must not count as success')
|
|
assert not hub._pending_commands
|
|
packet = next(p for p in socket.sent if p.get('type') == 'vmix.command')
|
|
await hub.receive_command_ack('GFX-DELIVERY', {'request_id': packet['request_id'], 'ok': True})
|
|
assert not hub._mapping_value_cache
|
|
asyncio.run(scenario())
|