Files
ledgrab/server/tests/test_events_ws_parity.py
T
alexei.dolgolyov ddae5719cf chore(frontend-infra): inbound-event allowlist + storage/state touch-ups
events-ws gains an inbound-event allowlist matching the new server-side
allowlist; test_events_ws_parity pins the two lists in sync.
state + storage modules and the streams / integrations /
z2m-light-targets / streams-*-templates editors absorb the
closeIfPristine guard alongside small UX fixes. css-editor template
picks up the new MiniSelect markup for the filter-kind picker.
2026-05-23 00:50:15 +03:00

95 lines
3.6 KiB
Python

"""Cross-cutting parity guard between server event emitters and the TS allowlist.
The frontend ``events-ws.ts`` module enforces a typed allowlist on inbound
WebSocket message ``type`` values. If a new server-side ``fire_event(...)``
call introduces a type that isn't in the JS allowlist, the corresponding
UI silently breaks. This test rederives the set of types the server emits
from the Python source and asserts every one is present in the TS
allowlist source, catching the drift at CI time rather than in production.
"""
from __future__ import annotations
import re
from pathlib import Path
import pytest
_REPO_ROOT = Path(__file__).resolve().parents[1]
_SERVER_SRC = _REPO_ROOT / "src" / "ledgrab"
_EVENTS_WS = _SERVER_SRC / "static" / "js" / "core" / "events-ws.ts"
# Scan the small set of files that funnel into the global ``events-ws.ts``
# dispatcher (``fire_event`` / ``_fire_event`` / ``_emit`` / the wrapper
# ``fire_entity_event``). Everything else is per-WS-handler protocol and
# is read off its own connection — not relevant here.
#
# Patterns we recognise as broadcast emit sites:
# fire_event({"type": "<x>" ...})
# _fire_event({"type": "<x>" ...})
# self._emit("<x>", ...) — discovery_watcher.py
# fire_entity_event(...) — always "entity_changed"
_EMIT_PATTERNS = (
re.compile(r'fire_event\(\s*\{\s*"type"\s*:\s*"([a-z_]+)"'),
re.compile(r'_fire_event\(\s*\{\s*"type"\s*:\s*"([a-z_]+)"'),
re.compile(r'self\._emit\(\s*"([a-z_]+)"'),
)
_FIRE_ENTITY_EVENT_RE = re.compile(r"fire_entity_event\(")
def _server_event_types() -> set[str]:
"""Extract event types that flow into the events-ws.ts dispatcher."""
discovered: set[str] = set()
for path in _SERVER_SRC.rglob("*.py"):
try:
text = path.read_text(encoding="utf-8")
except OSError:
continue
for rx in _EMIT_PATTERNS:
discovered.update(rx.findall(text))
if _FIRE_ENTITY_EVENT_RE.search(text):
# fire_entity_event always dispatches the "entity_changed" type.
discovered.add("entity_changed")
return discovered
def _ts_allowlist_types() -> set[str]:
"""Parse the allowlist constant out of events-ws.ts source text."""
text = _EVENTS_WS.read_text(encoding="utf-8")
match = re.search(
r"_ALLOWED_SERVER_EVENT_TYPES[^=]*=\s*new\s+Set\(\[(.*?)\]\)",
text,
re.DOTALL,
)
if not match:
pytest.fail("Could not locate _ALLOWED_SERVER_EVENT_TYPES in events-ws.ts")
body = match.group(1)
return set(re.findall(r"'([a-z_]+)'", body))
def test_every_server_event_type_is_in_ts_allowlist() -> None:
"""Catch the regression where the JS allowlist forgets a new event.
Failure mode: a server-side ``fire_event({"type": "foo"})`` ships, the
UI never sees it because ``events-ws.ts`` filters it out, and the bug
surfaces only via "feature X stopped working" reports.
"""
server_types = _server_event_types()
allowlist = _ts_allowlist_types()
missing = sorted(server_types - allowlist)
assert not missing, (
"Server emits these event types but the TS allowlist excludes "
f"them — features that listen will silently break: {missing}. "
"Either add them to _ALLOWED_SERVER_EVENT_TYPES in "
"static/js/core/events-ws.ts or move them to _LOCAL_TYPES in this "
"test if they're a per-WS protocol envelope."
)
def test_ts_allowlist_is_not_empty() -> None:
"""Sanity check: the parse helper actually found the constant."""
allowlist = _ts_allowlist_types()
assert allowlist, "Parser failed to extract allowlist from events-ws.ts"