249 lines
11 KiB
Python
249 lines
11 KiB
Python
"""Real reader/register/inventory races use SQLite and protocol-shaped ACKs only."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime
|
|
import json
|
|
import threading
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from hockey_data import agent_bridge as bridge
|
|
from hockey_data.broadcast_readiness import broadcast_readiness
|
|
from hockey_data.auth_bridge import HockeyUser
|
|
from hockey_data.models import VmixAssignment, VmixDevice, VmixMappingField, VmixMappingProfile
|
|
from tests.support import LocalTestDatabase
|
|
|
|
|
|
DEVICE = "GFX-INVENTORY-SESSION"
|
|
SECRET = "offline-inventory-session-" + "x" * 32
|
|
|
|
|
|
def inventory(marker):
|
|
return {"fingerprint": ("a" if marker == "OldOnly" else "b") * 64, "inputs": [
|
|
{"key": "score", "number": "1", "title": "Score", "fields": [{"name": "Home.Text", "type": "text"}]},
|
|
{"key": marker, "number": "2", "title": marker, "fields": []}]}
|
|
|
|
|
|
class Socket:
|
|
client = SimpleNamespace(host="offline")
|
|
|
|
def __init__(self, hub):
|
|
self.hub, self.queue, self.sent = hub, asyncio.Queue(), []
|
|
self.close_sent = False
|
|
|
|
async def receive_json(self):
|
|
return await self.queue.get()
|
|
|
|
async def close(self, **kwargs):
|
|
# Sending a WebSocket close frame need not synchronously deliver the
|
|
# peer's disconnect. The old state worker can still be in a DB await.
|
|
self.close_sent = True
|
|
|
|
async def send_json(self, packet):
|
|
self.sent.append(packet)
|
|
if packet.get("type") not in {"vmix.command", "vmix.batch"}:
|
|
return
|
|
batch = packet["type"] == "vmix.batch"
|
|
commands = packet["commands"] if batch else [packet["command"]]
|
|
await self.hub.receive_command_ack(DEVICE, {
|
|
"type": "command.batch.ack" if batch else "command.ack", "request_id": packet["request_id"],
|
|
"assignment_id": packet["assignment_id"], "match_id": packet["match_id"], "ok": True,
|
|
**({"results": [{"ok": True, "reason": ""} for _ in commands]} if batch else {})})
|
|
|
|
|
|
@pytest.fixture
|
|
def rig(tmp_path):
|
|
db = LocalTestDatabase(tmp_path / "inventory-session.sqlite")
|
|
db.create_all()
|
|
with db.session() as session:
|
|
profile = VmixMappingProfile(name="Published", project_fingerprint="a" * 64,
|
|
inventory_json=json.dumps(inventory("OldOnly")), version=1)
|
|
session.add(profile); session.flush()
|
|
device = VmixDevice(device_uuid=DEVICE, device_secret_hash=bridge._secret_hash(SECRET),
|
|
wfl_user_id="operator", is_active_for_account=True, current_match_external_id="match-A",
|
|
current_assignment_key="assignment-A", mapping_profile_id=profile.id, mapping_profile_version=1,
|
|
project_fingerprint="a" * 64, project_inventory_json=json.dumps(inventory("OldOnly")),
|
|
agent_version="1.5.3", vmix_connected=True)
|
|
session.add(device); session.flush()
|
|
session.add(VmixAssignment(device_id=device.id, assignment_key="assignment-A", wfl_user_id="operator",
|
|
game_external_id="match-A", active=True))
|
|
session.add(VmixMappingField(profile_id=profile.id, data_key="probe.home", vmix_input_title="Score",
|
|
vmix_field="Home.Text", field_type="text"))
|
|
hub = bridge.VmixAgentHub(db)
|
|
hub.mapping_data.data_catalog = lambda *_a, **_k: {
|
|
"items": [{"key": "probe.home", "value": "1", "source_code": "probe"}], "sources": []}
|
|
hub._persist_score_external_snapshot = lambda *_a, **_k: None
|
|
state = SimpleNamespace(db=db, hub=hub, gates=[], old=Socket(hub), new=Socket(hub),
|
|
user=HockeyUser(id="operator", login="operator", display_name="operator"))
|
|
hub._live[DEVICE] = bridge.LiveAgent(state.old, datetime.utcnow())
|
|
hub._device_status[DEVICE] = {"vmix_connected": True}
|
|
yield state
|
|
db.engine.dispose()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def old_reader(rig):
|
|
reader = asyncio.create_task(rig.hub.read_agent_messages(DEVICE, rig.old))
|
|
try:
|
|
yield reader
|
|
finally:
|
|
for gate in rig.gates:
|
|
gate.set()
|
|
reader.cancel()
|
|
await asyncio.gather(reader, return_exceptions=True)
|
|
await rig.hub.stop_auto_refresh()
|
|
|
|
|
|
def hold_worker(rig, monkeypatch, function_name, *, after_sql=False):
|
|
"""Hold an actual DB worker around SQL, retaining run_db cancellation join."""
|
|
original = bridge.run_db
|
|
loop = asyncio.get_running_loop()
|
|
gate = SimpleNamespace(entered=asyncio.Event(), release=threading.Event(), finished=asyncio.Event(), calls=0)
|
|
rig.gates.append(gate.release)
|
|
|
|
async def dispatched(function, *args, **kwargs):
|
|
if function.__name__ != function_name:
|
|
return await original(function, *args, **kwargs)
|
|
gate.calls += 1
|
|
if gate.calls != 1:
|
|
return await original(function, *args, **kwargs)
|
|
|
|
def blocked():
|
|
result = function(*args, **kwargs) if after_sql else None
|
|
loop.call_soon_threadsafe(gate.entered.set)
|
|
try:
|
|
assert gate.release.wait(10), "Test worker was not released"
|
|
return result if after_sql else function(*args, **kwargs)
|
|
finally:
|
|
loop.call_soon_threadsafe(gate.finished.set)
|
|
return await original(blocked)
|
|
monkeypatch.setattr(bridge, "run_db", dispatched)
|
|
return gate
|
|
|
|
|
|
def observe_inventory(rig, monkeypatch):
|
|
original = rig.hub.receive_inventory
|
|
observed = SimpleNamespace(old_calls=0, old_finished=asyncio.Event())
|
|
async def receive(device_id, message, **kwargs):
|
|
old = message.get("inventory", {}).get("fingerprint") == "a" * 64
|
|
if old:
|
|
observed.old_calls += 1
|
|
try:
|
|
return await original(device_id, message, **kwargs)
|
|
finally:
|
|
if old:
|
|
observed.old_finished.set()
|
|
monkeypatch.setattr(rig.hub, "receive_inventory", receive)
|
|
return observed
|
|
|
|
|
|
async def reconnect(rig, *, heartbeat=True):
|
|
await rig.hub.register(rig.new, {"device_id": DEVICE, "device_secret": SECRET,
|
|
"agent_version": "1.5.3", "vmix": {"connected": True}})
|
|
assert rig.old.close_sent and rig.hub._live[DEVICE].websocket is rig.new
|
|
assert DEVICE not in rig.hub._mapping_inventory_ready
|
|
if heartbeat:
|
|
await rig.hub.receive_status(DEVICE, {"type": "heartbeat", "vmix": {"connected": True}})
|
|
|
|
|
|
async def checks(rig):
|
|
result = await broadcast_readiness(rig.hub, rig.user, device_id=DEVICE, game_id="match-A")
|
|
return result, {item["key"]: item for item in result["checks"]}
|
|
|
|
|
|
async def fresh_inventory(rig, *, expect_ready=True):
|
|
await rig.hub.receive_inventory(DEVICE, {"type": "vmix.inventory", "inventory": inventory("NewOnly")}, websocket=rig.new)
|
|
for _ in range(10):
|
|
await asyncio.sleep(0)
|
|
jobs = [job for job in rig.hub._mapping_recovery_jobs.values() if not job.done()]
|
|
if not jobs:
|
|
break
|
|
await asyncio.wait_for(asyncio.gather(*jobs), 5)
|
|
result, state = await checks(rig)
|
|
assert state["inventory"]["details"]["current_connection"] is True
|
|
if expect_ready:
|
|
assert result["status"] == "ready"
|
|
assert state["delivery"]["status"] == "ready"
|
|
with rig.db.session() as session:
|
|
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == DEVICE))
|
|
titles = {row["title"] for row in json.loads(device.project_inventory_json)["inputs"]}
|
|
assert "NewOnly" in titles and "OldOnly" not in titles
|
|
|
|
|
|
def test_old_inventory_db_completion_cannot_confirm_current_socket_then_fresh_inventory_can(rig, monkeypatch):
|
|
async def scenario():
|
|
# SQL has committed while this is still the old connection; only the
|
|
# post-await identity fence can reject its late completion.
|
|
gate = hold_worker(rig, monkeypatch, "persist_inventory", after_sql=True)
|
|
observed = observe_inventory(rig, monkeypatch)
|
|
async with old_reader(rig):
|
|
await rig.old.queue.put({"type": "vmix.inventory", "inventory": inventory("OldOnly")})
|
|
await asyncio.wait_for(gate.entered.wait(), 5)
|
|
await reconnect(rig)
|
|
gate.release.set()
|
|
await asyncio.wait_for(observed.old_finished.wait(), 5)
|
|
result, state = await checks(rig)
|
|
assert DEVICE not in rig.hub._mapping_inventory_ready
|
|
assert result["status"] != "ready" and state["inventory"]["status"] == "unknown"
|
|
assert state["delivery"]["status"] == "unknown"
|
|
assert not any(packet.get("type") in {"vmix.command", "vmix.batch", "mapping.assigned"} for packet in rig.new.sent)
|
|
await fresh_inventory(rig)
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_new_inventory_waits_out_old_worker_without_old_completion_overwriting_it(rig, monkeypatch):
|
|
async def scenario():
|
|
gate = hold_worker(rig, monkeypatch, "persist_inventory")
|
|
async with old_reader(rig):
|
|
await rig.old.queue.put({"type": "vmix.inventory", "inventory": inventory("OldOnly")})
|
|
await asyncio.wait_for(gate.entered.wait(), 5)
|
|
await reconnect(rig)
|
|
newer = asyncio.create_task(fresh_inventory(rig))
|
|
try:
|
|
await asyncio.sleep(.03)
|
|
assert not newer.done() and gate.calls == 1, "New inventory dispatched another writer before old worker joined"
|
|
gate.release.set()
|
|
await asyncio.wait_for(newer, 5)
|
|
assert gate.calls == 2 and not rig.hub._pending_commands
|
|
finally:
|
|
gate.release.set()
|
|
if not newer.done():
|
|
newer.cancel()
|
|
await asyncio.gather(newer, return_exceptions=True)
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("cancel_reader", [False, True], ids=["old-reader-stays-open", "cancel-joins-status-worker"])
|
|
def test_inventory_queued_behind_old_status_never_captures_new_socket(rig, monkeypatch, cancel_reader):
|
|
async def scenario():
|
|
gate = hold_worker(rig, monkeypatch, "_save_device_status")
|
|
observed = observe_inventory(rig, monkeypatch)
|
|
async with old_reader(rig) as reader:
|
|
await rig.old.queue.put({"type": "vmix.inventory", "inventory": inventory("OldOnly")})
|
|
await asyncio.wait_for(gate.entered.wait(), 5)
|
|
worker = next(task for task in asyncio.all_tasks() if task.get_name() == f"hockey-agent-state:{DEVICE}")
|
|
# The old status DB await still owns the per-device status lock.
|
|
# New inventory can proceed, while its heartbeat waits until join.
|
|
await reconnect(rig, heartbeat=False)
|
|
await fresh_inventory(rig, expect_ready=False)
|
|
if cancel_reader:
|
|
reader.cancel()
|
|
await asyncio.sleep(.03)
|
|
assert not reader.done(), "Reader cancellation abandoned the running DB worker"
|
|
gate.release.set()
|
|
await asyncio.wait_for(gate.finished.wait(), 5)
|
|
if cancel_reader:
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await asyncio.wait_for(reader, 5)
|
|
else:
|
|
await asyncio.wait_for(worker, 5)
|
|
assert observed.old_calls == 0, "Old inventory was invoked after status await using the new connection"
|
|
await rig.hub.receive_status(DEVICE, {"type": "heartbeat", "vmix": {"connected": True}})
|
|
result, state = await checks(rig)
|
|
assert result["status"] == "ready" and state["inventory"]["details"]["current_connection"] is True
|
|
asyncio.run(scenario())
|