Pouzor
9b77e43ca1
Completes the split: `nodes` now holds only how a device is drawn on one canvas, and every device fact reaches the API from the inventory row. - The device columns are removed from `nodes` (SQLite table rebuild, the same shape as the existing device_inventory and canvas_state rebuilds). The drop is skipped, and logged, while any non-furniture node is still unlinked: those columns are the last copy of that node's facts. The backfill therefore reads them with raw SQL — by the time it runs, the model no longer declares them. - The status checker iterates devices, not nodes: one check per device however many canvases draw it, writing `status_live` / `last_seen` / `response_time_ms` on the row. `/ws/status` messages carry `device_id` and the node ids they light up. Hidden devices are not probed. - Readers repointed: scanner (last_scan lands on the row), proxmox (its node tier collapses into the inventory tier, keeping only the cluster handles), zigbee/zwave (one property refresh serves every canvas), rack inventory, liveview, stats, node dedupe. - `POST /scan/pending` merges into the row that already describes the host instead of minting a second one — one device is one row, whichever way it was documented. - Standalone keeps parity: the canvas blob gains `devices`, split on save and hydrated on load. A blob written before the split still reads. The rack inventory had a related bug: a mount that names a node explicitly printed the mount's device rather than the pinned node's. It now reads the node's own row. Tests that built a node with device columns are ported to the link; where a behaviour genuinely moved (properties refresh once on the row, last_scan is the device's) the assertion moved with it rather than being dropped. ha-relevant: yes
229 lines
7.5 KiB
Python
229 lines
7.5 KiB
Python
"""Tests for WebSocket status endpoint and broadcast helpers."""
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from app.api.routes.status import (
|
|
_connections,
|
|
_drop,
|
|
broadcast_scan_update,
|
|
broadcast_service_status,
|
|
broadcast_status,
|
|
)
|
|
from app.main import app
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_token() -> str:
|
|
from app.core.security import create_access_token
|
|
return create_access_token("admin")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WebSocket authentication
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_websocket_rejected_without_token():
|
|
"""Connection that sends no token field must be closed with 1008."""
|
|
with TestClient(app) as client, pytest.raises(WebSocketDisconnect), client.websocket_connect("/api/v1/status/ws/status") as ws:
|
|
ws.send_text(json.dumps({})) # missing token field
|
|
ws.receive_text() # triggers WebSocketDisconnect from server close
|
|
|
|
|
|
def test_websocket_rejected_with_invalid_token():
|
|
"""Connection that sends a garbage token must be closed."""
|
|
with TestClient(app) as client, pytest.raises(WebSocketDisconnect), client.websocket_connect("/api/v1/status/ws/status") as ws:
|
|
ws.send_text(json.dumps({"token": "not-a-valid-jwt"}))
|
|
ws.receive_text()
|
|
|
|
|
|
def test_websocket_rejected_with_malformed_json():
|
|
"""Connection that sends non-JSON as auth must be closed."""
|
|
with TestClient(app) as client, pytest.raises(WebSocketDisconnect), client.websocket_connect("/api/v1/status/ws/status") as ws:
|
|
ws.send_text("not-json")
|
|
ws.receive_text()
|
|
|
|
|
|
def test_websocket_accepted_with_valid_token():
|
|
"""Connection that sends a valid JWT as first message must be accepted."""
|
|
token = _make_token()
|
|
with TestClient(app) as client, client.websocket_connect("/api/v1/status/ws/status") as ws:
|
|
ws.send_text(json.dumps({"token": token}))
|
|
# Connection is open — subsequent messages should not raise
|
|
ws.send_text("ping")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# broadcast_status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_status_sends_to_connected_clients():
|
|
"""broadcast_status sends a JSON message to all active connections."""
|
|
received: list[str] = []
|
|
|
|
class FakeWS:
|
|
async def send_text(self, text: str) -> None:
|
|
received.append(text)
|
|
|
|
fake = FakeWS()
|
|
_connections.append(fake)
|
|
try:
|
|
await broadcast_status(
|
|
device_id="dev-1",
|
|
node_ids=["node-1", "node-2"],
|
|
status="online",
|
|
checked_at="2024-01-01T00:00:00",
|
|
response_time_ms=42,
|
|
)
|
|
finally:
|
|
_connections.remove(fake)
|
|
|
|
assert len(received) == 1
|
|
msg = json.loads(received[0])
|
|
assert msg["type"] == "status"
|
|
# One check, addressed by device, lighting up every canvas drawing it.
|
|
assert msg["device_id"] == "dev-1"
|
|
assert msg["node_ids"] == ["node-1", "node-2"]
|
|
assert msg["status"] == "online"
|
|
assert msg["response_time_ms"] == 42
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_status_no_response_time():
|
|
"""response_time_ms defaults to None."""
|
|
received: list[str] = []
|
|
|
|
class FakeWS:
|
|
async def send_text(self, text: str) -> None:
|
|
received.append(text)
|
|
|
|
fake = FakeWS()
|
|
_connections.append(fake)
|
|
try:
|
|
await broadcast_status(device_id="d", node_ids=["n"], status="offline", checked_at="t")
|
|
finally:
|
|
_connections.remove(fake)
|
|
|
|
msg = json.loads(received[0])
|
|
assert msg["response_time_ms"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_status_removes_dead_connection():
|
|
"""A connection that raises on send is removed from _connections."""
|
|
|
|
class DeadWS:
|
|
async def send_text(self, _: str) -> None:
|
|
raise RuntimeError("disconnected")
|
|
|
|
dead = DeadWS()
|
|
_connections.append(dead)
|
|
initial_len = len(_connections)
|
|
|
|
await broadcast_status(device_id="d", node_ids=["n"], status="online", checked_at="t")
|
|
|
|
assert dead not in _connections
|
|
assert len(_connections) == initial_len - 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# broadcast_scan_update
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_scan_update():
|
|
"""broadcast_scan_update sends scan_device_found payload."""
|
|
received: list[str] = []
|
|
|
|
class FakeWS:
|
|
async def send_text(self, text: str) -> None:
|
|
received.append(text)
|
|
|
|
fake = FakeWS()
|
|
_connections.append(fake)
|
|
try:
|
|
await broadcast_scan_update(run_id="run-42", devices_found=3)
|
|
finally:
|
|
_connections.remove(fake)
|
|
|
|
assert len(received) == 1
|
|
msg = json.loads(received[0])
|
|
assert msg["type"] == "scan_device_found"
|
|
assert msg["run_id"] == "run-42"
|
|
assert msg["devices_found"] == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_no_connections():
|
|
"""broadcast_* with no connections must not raise."""
|
|
assert len(_connections) == 0
|
|
await broadcast_status(device_id="d", node_ids=["n"], status="online", checked_at="t")
|
|
await broadcast_scan_update(run_id="r", devices_found=0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# broadcast_service_status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_service_status_payload():
|
|
received: list[str] = []
|
|
|
|
class FakeWS:
|
|
async def send_text(self, text: str) -> None:
|
|
received.append(text)
|
|
|
|
fake = FakeWS()
|
|
_connections.append(fake)
|
|
try:
|
|
await broadcast_service_status(
|
|
device_id="dev-7",
|
|
node_ids=["node-7"],
|
|
services=[{"port": 80, "protocol": "tcp", "status": "offline"}],
|
|
checked_at="2024-01-01T00:00:00",
|
|
)
|
|
finally:
|
|
_drop(fake)
|
|
|
|
msg = json.loads(received[0])
|
|
assert msg["type"] == "service_status"
|
|
assert msg["device_id"] == "dev-7"
|
|
assert msg["node_ids"] == ["node-7"]
|
|
assert msg["services"] == [{"port": 80, "protocol": "tcp", "status": "offline"}]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _drop — idempotent connection removal (regression for double-remove crash)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_drop_is_idempotent():
|
|
"""Dropping a connection twice must not raise (was a ValueError crash)."""
|
|
class FakeWS:
|
|
pass
|
|
|
|
fake = FakeWS()
|
|
_connections.append(fake)
|
|
_drop(fake)
|
|
_drop(fake) # second drop must be a no-op
|
|
assert fake not in _connections
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_dead_connection_dropped_once_safely():
|
|
"""A send failure removes the dead socket without a double-remove crash."""
|
|
class DeadWS:
|
|
async def send_text(self, _: str) -> None:
|
|
raise RuntimeError("disconnected")
|
|
|
|
dead = DeadWS()
|
|
_connections.append(dead)
|
|
await broadcast_status(device_id="d", node_ids=["n"], status="online", checked_at="t")
|
|
# A second broadcast must not raise even though dead is already gone.
|
|
await broadcast_status(device_id="d", node_ids=["n"], status="online", checked_at="t")
|
|
assert dead not in _connections
|