287 lines
14 KiB
Python
287 lines
14 KiB
Python
"""Connectivity messages must distinguish cached command errors from fresh loss.
|
|
|
|
These contracts run real status/overlay handlers and the real WebSocket reader.
|
|
The slow-state barrier pauses the actual DB save before SQL, leaving the reader
|
|
free to fence a late Mapping ACK without waiting for a database worker.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
|
|
import pytest
|
|
from fastapi import WebSocketDisconnect
|
|
from sqlalchemy import select
|
|
|
|
from hockey_data.models import VmixDevice
|
|
from tests.test_build192_mapping_delivery import DEVICE, profile, rows
|
|
|
|
|
|
async def _warm(tmp_path, *, connected=True):
|
|
database, hub, socket, user, catalog, assignment = await profile(tmp_path, rows(8))
|
|
await hub.receive_status(DEVICE, {"type": "heartbeat", "vmix": {"connected": True}})
|
|
applied = await hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True)
|
|
assert applied["ok"] and applied["applied"] == 8 and len(hub._mapping_value_cache) == 8
|
|
if not connected:
|
|
await hub.receive_status(DEVICE, {"type": "heartbeat", "vmix": {
|
|
"connected": False, "error": "dial tcp: connection refused"}})
|
|
assert not hub._mapping_value_cache
|
|
return database, hub, socket, user, catalog, assignment
|
|
|
|
|
|
async def _blocked_reader(hub, socket, monkeypatch):
|
|
release = threading.Event()
|
|
entered = asyncio.Event()
|
|
loop = asyncio.get_running_loop()
|
|
original_save = hub._save_device_status
|
|
blocked = False
|
|
|
|
def slow_save(*args):
|
|
nonlocal blocked
|
|
if not blocked:
|
|
blocked = True
|
|
loop.call_soon_threadsafe(entered.set)
|
|
assert release.wait(5), "test state DB worker was not released"
|
|
return original_save(*args)
|
|
|
|
monkeypatch.setattr(hub, "_save_device_status", slow_save)
|
|
hub._status_saved_at.pop(DEVICE, None)
|
|
reader = asyncio.create_task(hub.read_agent_messages(DEVICE, socket))
|
|
await socket.incoming.put({"type": "heartbeat", "vmix": {"connected": True}})
|
|
try:
|
|
await asyncio.wait_for(entered.wait(), 2)
|
|
except BaseException:
|
|
release.set()
|
|
reader.cancel()
|
|
await asyncio.gather(reader, return_exceptions=True)
|
|
raise
|
|
return reader, release
|
|
|
|
|
|
async def _stop_reader(hub, socket, reader, release):
|
|
release.set()
|
|
if not reader.done():
|
|
await socket.incoming.put(WebSocketDisconnect())
|
|
outcome = await asyncio.gather(reader, return_exceptions=True)
|
|
assert isinstance(outcome[0], WebSocketDisconnect), outcome
|
|
await hub.stop_auto_refresh()
|
|
|
|
|
|
async def _reader_barrier(hub, socket, message):
|
|
# Reading this matching ACK proves the previous status frame has passed
|
|
# through the fast reader, even while the state worker remains blocked.
|
|
request_id = "qa-connectivity-reader-barrier"
|
|
future = asyncio.get_running_loop().create_future()
|
|
hub._pending_commands[request_id] = (DEVICE, future)
|
|
try:
|
|
await socket.incoming.put(message)
|
|
await socket.incoming.put({"type": "command.ack", "request_id": request_id, "ok": True})
|
|
assert (await asyncio.wait_for(future, 1))["ok"] is True
|
|
finally:
|
|
hub._pending_commands.pop(request_id, None)
|
|
|
|
|
|
@pytest.mark.parametrize("kind,error_location,error", [
|
|
("heartbeat", "nested", "HTTP 500"),
|
|
("vmix.status", "nested", "HTTP 404 Not Found"),
|
|
("heartbeat", "top", "HTTP 503"),
|
|
("vmix.status", "reason", "HTTP 400 Bad Request"),
|
|
])
|
|
def test_cached_command_http_failure_preserves_real_status_cache_and_next_mapping(tmp_path, kind, error_location, error):
|
|
async def scenario():
|
|
database, hub, socket, _, catalog, _ = await _warm(tmp_path)
|
|
generation = hub._mapping_generations[DEVICE]
|
|
cached = dict(hub._mapping_value_cache)
|
|
message = {"type": kind, "vmix": {"connected": False}}
|
|
if error_location == "nested":
|
|
message["vmix"]["error"] = error
|
|
else:
|
|
message["error" if error_location == "top" else "reason"] = error
|
|
try:
|
|
await hub.receive_status(DEVICE, message)
|
|
assert hub._device_status[DEVICE]["vmix_connected"] is True
|
|
assert hub._mapping_generations[DEVICE] == generation
|
|
assert hub._mapping_value_cache == cached
|
|
with database.session() as session:
|
|
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == DEVICE))
|
|
assert device.vmix_connected is True
|
|
before = len(socket.executed)
|
|
unchanged = await hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True)
|
|
assert unchanged["unchanged"] == 8 and len(socket.executed) == before
|
|
catalog["sample.row.0"] = "new-after-command-error"
|
|
changed = await hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True)
|
|
assert changed["ok"] and changed["applied"] == 1 and changed["unchanged"] == 7
|
|
assert socket.executed[-1]["Value"] == "new-after-command-error"
|
|
assert hub._mapping_generations[DEVICE] == generation
|
|
finally:
|
|
await hub.stop_auto_refresh()
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("kind,error", [
|
|
("heartbeat", "dial tcp: connection refused"),
|
|
("vmix.status", "context deadline exceeded"),
|
|
("heartbeat", None),
|
|
("vmix.inventory", "HTTP 500"),
|
|
("overlay_status", "HTTP 500"),
|
|
])
|
|
def test_fresh_probe_transport_or_unknown_loss_invalidates_real_status_and_cache(tmp_path, kind, error):
|
|
async def scenario():
|
|
_, hub, _, _, _, _ = await _warm(tmp_path)
|
|
generation = hub._mapping_generations[DEVICE]
|
|
vmix = {"connected": False}
|
|
if error is not None:
|
|
vmix["error"] = error
|
|
try:
|
|
await hub.receive_status(DEVICE, {"type": kind, "vmix": vmix})
|
|
assert hub._device_status[DEVICE]["vmix_connected"] is False
|
|
assert hub._mapping_generations[DEVICE] > generation
|
|
assert not hub._mapping_value_cache
|
|
finally:
|
|
await hub.stop_auto_refresh()
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("initial,value", [(True, "false"), (False, "true"), (False, "false"), (True, 0), (False, 1), (True, None)])
|
|
def test_status_connectivity_requires_actual_boolean_not_truthiness(tmp_path, initial, value):
|
|
async def scenario():
|
|
_, hub, _, _, _, _ = await _warm(tmp_path, connected=initial)
|
|
generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
|
|
recoveries = []
|
|
hub._request_mapping_recovery = recoveries.append
|
|
try:
|
|
await hub.receive_status(DEVICE, {"type": "vmix.status", "vmix": {"connected": value}})
|
|
assert hub._device_status[DEVICE]["vmix_connected"] is initial
|
|
assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
|
|
assert recoveries == [], "A malformed connectivity value must not revive a disconnected Agent"
|
|
finally:
|
|
await hub.stop_auto_refresh()
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("initial,value", [(True, "false"), (False, "true"), (False, "false"), (True, 0), (False, 1), (True, None)])
|
|
def test_actual_overlay_handler_ignores_non_boolean_connectivity_without_fabricating_tally(tmp_path, initial, value):
|
|
async def scenario():
|
|
_, hub, _, _, _, _ = await _warm(tmp_path, connected=initial)
|
|
hub._runtime_overlay_state[DEVICE] = {"1": {"input": "Previous input"}}
|
|
hub._runtime_overlay_authoritative[DEVICE] = initial
|
|
previous = dict(hub._runtime_overlay_state[DEVICE])
|
|
generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
|
|
recoveries = []
|
|
hub._request_mapping_recovery = recoveries.append
|
|
try:
|
|
await hub.receive_overlay_state(DEVICE, {"type": "vmix.overlay_state", "vmix_connected": value,
|
|
"overlays": {"1": {"input": "Untrusted new input"}}})
|
|
assert hub._device_status[DEVICE]["vmix_connected"] is initial
|
|
assert hub._runtime_overlay_state[DEVICE] == previous
|
|
assert hub._runtime_overlay_authoritative[DEVICE] is initial
|
|
assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
|
|
assert recoveries == []
|
|
finally:
|
|
await hub.stop_auto_refresh()
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_actual_overlay_false_is_fresh_loss_even_when_error_is_http500(tmp_path):
|
|
async def scenario():
|
|
_, hub, _, _, _, _ = await _warm(tmp_path)
|
|
hub._runtime_overlay_state[DEVICE] = {"1": {"input": "Score"}}
|
|
hub._runtime_overlay_authoritative[DEVICE] = True
|
|
generation = hub._mapping_generations[DEVICE]
|
|
try:
|
|
await hub.receive_overlay_state(DEVICE, {"type": "vmix.overlay_state", "vmix_connected": False,
|
|
"error": "HTTP 500", "overlays": {}})
|
|
assert hub._device_status[DEVICE]["vmix_connected"] is False
|
|
assert hub._mapping_generations[DEVICE] > generation and not hub._mapping_value_cache
|
|
assert DEVICE not in hub._runtime_overlay_state and hub._runtime_overlay_authoritative[DEVICE] is False
|
|
finally:
|
|
await hub.stop_auto_refresh()
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["heartbeat", "vmix.status"])
|
|
def test_reader_does_not_fence_cached_http_command_failure_while_state_db_is_blocked(tmp_path, monkeypatch, kind):
|
|
async def scenario():
|
|
_, hub, socket, _, catalog, _ = await _warm(tmp_path)
|
|
generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
|
|
reader, release = await _blocked_reader(hub, socket, monkeypatch)
|
|
try:
|
|
await _reader_barrier(hub, socket, {"type": kind, "vmix": {"connected": False, "error": "HTTP 500"}})
|
|
assert not release.is_set()
|
|
assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
|
|
assert hub._device_status[DEVICE]["vmix_connected"] is True
|
|
catalog["sample.row.0"] = "still-delivered-while-state-worker-is-blocked"
|
|
result = await asyncio.wait_for(hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True), 2)
|
|
assert result["ok"] and result["applied"] == 1 and result["unchanged"] == 7
|
|
assert socket.executed[-1]["Value"] == "still-delivered-while-state-worker-is-blocked"
|
|
assert not release.is_set() and hub._mapping_generations[DEVICE] == generation
|
|
finally:
|
|
await _stop_reader(hub, socket, reader, release)
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("kind,value,after_loss", [
|
|
("vmix.status", 0, False),
|
|
("vmix.status", "true", True),
|
|
("vmix.status", 1, True),
|
|
("vmix.overlay_state", "false", True),
|
|
])
|
|
def test_reader_ignores_non_boolean_connectivity_without_loss_or_premature_recovery(tmp_path, monkeypatch, kind, value, after_loss):
|
|
async def scenario():
|
|
_, hub, socket, _, _, _ = await _warm(tmp_path)
|
|
reader, release = await _blocked_reader(hub, socket, monkeypatch)
|
|
recoveries = []
|
|
hub._request_mapping_recovery = recoveries.append
|
|
try:
|
|
if after_loss:
|
|
await _reader_barrier(hub, socket, {"type": "heartbeat", "vmix": {"connected": False}})
|
|
assert not hub._mapping_value_cache
|
|
generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
|
|
message = {"type": kind, "vmix_connected": value, "overlays": {}} if kind == "vmix.overlay_state" else {
|
|
"type": kind, "vmix": {"connected": value}}
|
|
await _reader_barrier(hub, socket, message)
|
|
assert not release.is_set()
|
|
assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
|
|
assert recoveries == [], "Unknown connectivity must not start recovery before a fresh positive probe"
|
|
finally:
|
|
await _stop_reader(hub, socket, reader, release)
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("loss", [
|
|
{"type": "vmix.inventory", "vmix": {"connected": False, "error": "HTTP 500"}, "inventory": {}},
|
|
{"type": "vmix.overlay_state", "vmix_connected": False, "error": "HTTP 500", "overlays": {}},
|
|
{"type": "heartbeat", "vmix": {"connected": False, "error": "dial tcp: connection refused"}},
|
|
{"type": "vmix.status", "vmix": {"connected": False, "error": "context deadline exceeded"}},
|
|
{"type": "heartbeat", "vmix": {"connected": False}},
|
|
], ids=["fresh-inventory-http500", "fresh-overlay-http500", "refused", "timeout", "legacy-unknown"])
|
|
def test_reader_fences_fresh_loss_before_state_db_and_late_ack_cannot_restore_mapping_cache(tmp_path, monkeypatch, loss):
|
|
async def scenario():
|
|
_, hub, socket, _, catalog, _ = await _warm(tmp_path)
|
|
generation = hub._mapping_generations[DEVICE]
|
|
reader, release = await _blocked_reader(hub, socket, monkeypatch)
|
|
socket.auto_ack = False
|
|
socket.command_arrived.clear()
|
|
catalog["sample.row.0"] = "late-ack-must-not-confirm"
|
|
task = asyncio.create_task(hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True))
|
|
try:
|
|
await asyncio.wait_for(socket.command_arrived.wait(), 2)
|
|
packet = socket.mapping_packets[-1]
|
|
await socket.incoming.put(loss)
|
|
# The reader sees loss before this ACK. The actual status saver is
|
|
# still blocked in a separate thread and cannot do the invalidation.
|
|
await socket.incoming.put(socket.payload(packet))
|
|
result = await asyncio.wait_for(asyncio.shield(task), 2)
|
|
assert not release.is_set()
|
|
assert hub._device_status[DEVICE]["vmix_connected"] is True
|
|
assert hub._mapping_generations[DEVICE] > generation
|
|
assert not result["ok"] and result["applied"] == 0 and not hub._mapping_value_cache
|
|
assert not hub._pending_commands
|
|
finally:
|
|
release.set()
|
|
if not task.done():
|
|
task.cancel()
|
|
await asyncio.gather(task, return_exceptions=True)
|
|
await _stop_reader(hub, socket, reader, release)
|
|
asyncio.run(scenario())
|