147 lines
7.0 KiB
Python
147 lines
7.0 KiB
Python
"""Exercise actual acknowledged Mapping delivery with protocol-shaped replies."""
|
|
import asyncio
|
|
from datetime import datetime
|
|
|
|
from hockey_data.agent_bridge import LiveAgent, VmixAgentHub
|
|
from hockey_data.models import VmixAssignment, VmixDevice
|
|
from tests.support import LocalTestDatabase
|
|
|
|
|
|
def transport(tmp_path, fail_at=None):
|
|
db = LocalTestDatabase(tmp_path / "mapping.sqlite")
|
|
db.create_all()
|
|
with db.session() as s:
|
|
device = VmixDevice(device_uuid="GFX-OWNER", device_secret_hash="x", wfl_user_id="owner",
|
|
is_active_for_account=True, agent_version="1.5.3", current_match_external_id="g", current_assignment_key="assignment")
|
|
s.add(device)
|
|
s.flush()
|
|
s.add(VmixAssignment(device_id=device.id, assignment_key="assignment", wfl_user_id="owner",
|
|
game_external_id="g", active=True))
|
|
class Socket:
|
|
def __init__(self, device_id):
|
|
self.device_id = device_id
|
|
self.frames = []
|
|
self.auto_ack = True
|
|
self.command_received = asyncio.Event()
|
|
self.max_pending = 0
|
|
|
|
async def acknowledge(self, message, *, results=None, ok=None, reason=""):
|
|
is_batch = message.get("type") == "vmix.batch"
|
|
if is_batch and results is None:
|
|
results = [{"ok": True, "reason": ""} for _ in message["commands"]]
|
|
payload = {
|
|
"type": "command.batch.ack" if is_batch else "command.ack", "request_id": message["request_id"],
|
|
"assignment_id": message["assignment_id"], "match_id": message["match_id"],
|
|
"ok": (all(item.get("ok") is True for item in results) if is_batch else True) if ok is None else ok,
|
|
"reason": reason,
|
|
}
|
|
if is_batch:
|
|
payload["results"] = results
|
|
await hub.receive_command_ack(self.device_id, payload)
|
|
|
|
async def send_json(self, message):
|
|
if fail_at is not None and len(self.frames) == fail_at:
|
|
raise ConnectionError("fake disconnected socket")
|
|
self.frames.append(message)
|
|
if message.get("type") in {"vmix.command", "vmix.batch"}:
|
|
self.max_pending = max(self.max_pending, sum(
|
|
pending[0] == self.device_id for pending in hub._pending_commands.values()))
|
|
self.command_received.set()
|
|
if self.auto_ack:
|
|
await self.acknowledge(message)
|
|
own, other = Socket("GFX-OWNER"), Socket("GFX-VIEWER")
|
|
hub = VmixAgentHub(db)
|
|
hub._live["GFX-OWNER"] = LiveAgent(own, datetime.utcnow())
|
|
hub._live["GFX-VIEWER"] = LiveAgent(other, datetime.utcnow())
|
|
return hub, own, other
|
|
|
|
|
|
def entries(count):
|
|
return [{"data_key": str(i), "command": {"Function": "SetText", "Input": f"input-{i // 31}",
|
|
"SelectedName": f"Field{i}.Text", "Value": str(i)}} for i in range(count)]
|
|
|
|
|
|
def frame_commands(frame):
|
|
"""Read actual wire frames without bypassing the hub's ACK transport."""
|
|
if frame.get("type") == "vmix.batch":
|
|
return frame["commands"]
|
|
if frame.get("type") == "vmix.command":
|
|
return [frame["command"]]
|
|
return []
|
|
|
|
|
|
def sent_commands(socket):
|
|
return [command for frame in socket.frames for command in frame_commands(frame)]
|
|
|
|
|
|
def send(hub, rows):
|
|
return asyncio.run(hub._send_mapping_entries_resilient("GFX-OWNER", assignment_id="assignment",
|
|
match_id="g", entries=rows, use_batch=True))
|
|
|
|
|
|
def test_372_mapping_fields_are_acknowledged_ordered_and_isolated(tmp_path):
|
|
hub, own, other = transport(tmp_path)
|
|
rows = entries(372)
|
|
result = send(hub, rows)
|
|
assert result["commands_sent"] == 372 and result["input_groups"] == 12
|
|
assert all(item["ok"] for item in result["results"])
|
|
assert result["ack_waited"] is True and result["transport"] == "bounded_batch_ack"
|
|
assert own.max_pending == 1 and result["retries"] == 0
|
|
assert own.frames[0]["type"] == "match.assign"
|
|
assert sent_commands(own) == [r["command"] for r in rows]
|
|
assert all(f["type"] == "vmix.batch" and 1 <= len(f["commands"]) <= 6
|
|
and len({(c["Input"], c["SelectedName"]) for c in f["commands"]}) == len(f["commands"])
|
|
and f["match_id"] == "g" and f["assignment_id"] == "assignment"
|
|
for f in own.frames[1:])
|
|
assert len(own.frames[1:]) == 62
|
|
assert len({f["request_id"] for f in own.frames[1:]}) == 62
|
|
assert other.frames == []
|
|
|
|
|
|
def test_disconnect_stops_remaining_mapping_and_reports_failure(tmp_path):
|
|
hub, own, other = transport(tmp_path, fail_at=4) # assignment plus three acknowledged six-command batches
|
|
result = send(hub, entries(35))
|
|
assert result["commands_attempted"] == result["commands_sent"] == 24
|
|
assert result["commands_confirmed"] == 18 and result["commands_failed"] == 17
|
|
assert [row["ok"] for row in result["results"]] == [True] * 18 + [False] * 17
|
|
assert result["retries"] == 0, "An offline Agent must not receive speculative retries"
|
|
assert len(sent_commands(own)) == 18
|
|
assert len(own.frames) == 4 and other.frames == []
|
|
assert not hub._pending_commands
|
|
|
|
|
|
def test_another_agent_completes_while_first_agent_ack_is_pending(tmp_path):
|
|
hub, own, other = transport(tmp_path)
|
|
with hub.database.session() as session:
|
|
device = VmixDevice(device_uuid="GFX-VIEWER", device_secret_hash="y", wfl_user_id="viewer",
|
|
is_active_for_account=True, current_match_external_id="g2", current_assignment_key="assignment2")
|
|
session.add(device)
|
|
session.flush()
|
|
session.add(VmixAssignment(device_id=device.id, assignment_key="assignment2", wfl_user_id="viewer",
|
|
game_external_id="g2", active=True))
|
|
own.auto_ack = False
|
|
|
|
async def scenario():
|
|
first = asyncio.create_task(hub._send_mapping_entries_resilient(
|
|
"GFX-OWNER", assignment_id="assignment", match_id="g", entries=entries(1), use_batch=True))
|
|
await asyncio.wait_for(own.command_received.wait(), 3)
|
|
second = await asyncio.wait_for(hub._send_mapping_entries_resilient(
|
|
"GFX-VIEWER", assignment_id="assignment2", match_id="g2", entries=entries(1), use_batch=True), 3)
|
|
assert second["commands_sent"] == 1 and second["results"][0]["ok"]
|
|
assert second["transport"] == "bounded_batch_ack" and other.frames[-1]["type"] == "vmix.batch"
|
|
assert not first.done(), "A slow first Agent must not block the second Agent"
|
|
await own.acknowledge(own.frames[-1])
|
|
assert (await first)["results"][0]["ok"]
|
|
assert other.frames[-1]["match_id"] == "g2"
|
|
assert own.frames[-1]["match_id"] == "g"
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_unconfirmed_assignment_does_not_send_mapping(tmp_path):
|
|
hub, own, other = transport(tmp_path, fail_at=0)
|
|
result = send(hub, entries(8))
|
|
assert result["assignment_refreshed"] is False
|
|
assert result["commands_sent"] == 0 and not any(row["ok"] for row in result["results"])
|
|
assert own.frames == [] and other.frames == []
|