diff --git a/tests/test_websocket_broadcast_regression.py b/tests/test_websocket_broadcast_regression.py new file mode 100644 index 00000000..2425e6cc --- /dev/null +++ b/tests/test_websocket_broadcast_regression.py @@ -0,0 +1,70 @@ +"""Regression test for the WebSocketServer.broadcast() runtime TypeError bug. + +Phase 5 of any_type_componentization_20260621 changed +WebSocketServer.broadcast(channel, payload) -> broadcast(message: WebSocketMessage) +but did not update internal callers in src/app_controller.py + src/events.py. +This produced worker[queue_fallback] TypeError spam on the GUI thread. + +This test catches the regression and is reused by code_path_audit_20260607 +as a structural assertion. + +CONVENTION: 1-space indentation. NO COMMENTS. +""" +from __future__ import annotations + +import inspect +from pathlib import Path +from typing import Any + +from src.api_hooks import WebSocketMessage, WebSocketServer + + +class _MockApp: + test_hooks_enabled: bool = True + + +def _make_server() -> WebSocketServer: + return WebSocketServer(_MockApp(), port=9001) + + +def test_websocket_server_broadcast_signature() -> None: + """WebSocketServer.broadcast must accept a single WebSocketMessage argument (self + message).""" + sig = inspect.signature(WebSocketServer.broadcast) + params = list(sig.parameters.keys()) + assert len(params) == 2, f"expected 2 params (self + message), got {len(params)}: {params}" + + +def test_websocket_server_broadcast_rejects_legacy_2arg_call() -> None: + """Calling broadcast with 2 positional args (legacy signature) must raise TypeError.""" + server = _make_server() + raised = False + try: + server.broadcast("channel", {"key": "value"}) + except TypeError: + raised = True + assert raised, "broadcast should reject legacy 2-arg call" + + +def test_websocket_server_broadcast_accepts_websocket_message_instance() -> None: + """The new signature accepts a WebSocketMessage instance (no-op when not started).""" + server = _make_server() + msg = WebSocketMessage(channel="test", payload={"key": "value"}) + server.broadcast(msg) + + +def test_internal_callers_use_websocket_message_signature() -> None: + """Grep all internal callers of broadcast() in src/ and assert they use the new signature.""" + src_root = Path(__file__).resolve().parents[1] / "src" + legacy_sites: list[str] = [] + for py_file in src_root.rglob("*.py"): + text = py_file.read_text(encoding="utf-8") + for lineno, line in enumerate(text.splitlines(), start=1): + if ".broadcast(" not in line: + continue + if "WebSocketMessage(" in line: + continue + if 'broadcast("' not in line and "broadcast('" not in line: + continue + rel = py_file.relative_to(src_root.parent) + legacy_sites.append(f"{rel}:{lineno}: {line.strip()}") + assert not legacy_sites, "legacy broadcast() callers found:\n" + "\n".join(legacy_sites) \ No newline at end of file