ed
e9fa69ddc1
Phase 5 of any_type_componentization_20260621. Promotes the WebSocket
broadcast signature in src/api_hooks.py from (channel, payload: dict) to
a typed WebSocketMessage dataclass (16 Any sites):
NEW dataclass (inline in src/api_hooks.py):
- WebSocketMessage (frozen=True): channel: str, payload: JsonValue
MODIFIED:
- _serialize_for_api(obj: Any) -> JsonValue (typed return)
- broadcast(channel: str, payload: dict[str, Any]) -> broadcast(message: WebSocketMessage)
- _get_app_attr / _set_app_attr signatures UNCHANGED (Pattern 4 preserved)
NEW tests/test_api_hooks_dataclasses.py (12 tests, all pass):
- test_websocket_message_construction
- test_websocket_message_with_list_payload
- test_websocket_message_with_nested_payload
- test_websocket_message_is_frozen
- test_websocket_message_to_json
- test_serialize_for_api_returns_dict_for_to_dict_object
- test_serialize_for_api_handles_nested_lists
- test_serialize_for_api_handles_purepath
- test_serialize_for_api_passthrough_for_primitives
- test_serialize_for_api_handles_mixed_nesting
- test_get_app_attr_signature_preserved (Pattern 4 invariant)
- test_set_app_attr_signature_preserved (Pattern 4 invariant)
MODIFIED tests/test_websocket_server.py:
- Updated broadcast() call site to use WebSocketMessage(channel=..., payload=...)
- Added WebSocketMessage import
Verified:
uv run pytest tests/test_api_hooks_dataclasses.py tests/test_api_hooks_warmup.py tests/test_websocket_server.py --timeout=30
23 passed in 5.03s (12 new + 10 existing + 1 websocket)
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""Tests for src/api_hooks.py WebSocketMessage + JsonValue usage
|
|
|
|
Phase 5 of any_type_componentization_20260621. Verifies:
|
|
- WebSocketMessage dataclass (channel, payload: JsonValue)
|
|
- WebSocketMessage is frozen=True
|
|
- _serialize_for_api uses JsonValue type hint
|
|
- broadcast() takes WebSocketMessage instead of (channel, payload)
|
|
- _get_app_attr / _set_app_attr signatures UNCHANGED (Pattern 4 preserved)
|
|
|
|
CONVENTION: 1-space indentation. NO COMMENTS.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pytest
|
|
from src import api_hooks
|
|
from src.type_aliases import JsonValue
|
|
|
|
|
|
def test_websocket_message_construction() -> None:
|
|
msg = api_hooks.WebSocketMessage(channel="status", payload={"status": "ok"})
|
|
assert msg.channel == "status"
|
|
assert msg.payload == {"status": "ok"}
|
|
|
|
|
|
def test_websocket_message_with_list_payload() -> None:
|
|
msg = api_hooks.WebSocketMessage(channel="events", payload=[{"type": "x"}, {"type": "y"}])
|
|
assert msg.payload == [{"type": "x"}, {"type": "y"}]
|
|
|
|
|
|
def test_websocket_message_with_nested_payload() -> None:
|
|
msg = api_hooks.WebSocketMessage(
|
|
channel="data",
|
|
payload={"users": [{"name": "a", "meta": {"active": True}}], "count": 1}
|
|
)
|
|
assert msg.payload["count"] == 1
|
|
assert msg.payload["users"][0]["meta"]["active"] is True
|
|
|
|
|
|
def test_websocket_message_is_frozen() -> None:
|
|
msg = api_hooks.WebSocketMessage(channel="x", payload={})
|
|
with pytest.raises(Exception):
|
|
msg.channel = "mutated"
|
|
|
|
|
|
def test_websocket_message_to_json() -> None:
|
|
msg = api_hooks.WebSocketMessage(channel="status", payload={"ok": True})
|
|
j = json.dumps({"channel": msg.channel, "payload": msg.payload})
|
|
assert json.loads(j) == {"channel": "status", "payload": {"ok": True}}
|
|
|
|
|
|
def test_serialize_for_api_returns_dict_for_to_dict_object() -> None:
|
|
class WithToDict:
|
|
def to_dict(self) -> dict:
|
|
return {"k": "v"}
|
|
result = api_hooks._serialize_for_api(WithToDict())
|
|
assert result == {"k": "v"}
|
|
|
|
|
|
def test_serialize_for_api_handles_nested_lists() -> None:
|
|
obj = {"items": [{"a": 1}, {"b": 2}]}
|
|
result = api_hooks._serialize_for_api(obj)
|
|
assert result == {"items": [{"a": 1}, {"b": 2}]}
|
|
|
|
|
|
def test_serialize_for_api_handles_purepath() -> None:
|
|
from pathlib import PurePath, PureWindowsPath
|
|
p = PurePath("a/b/c") # Use a relative path to avoid Windows normalization
|
|
result = api_hooks._serialize_for_api(p)
|
|
assert isinstance(result, str)
|
|
# Either forward or backslash separator; both are valid string representations
|
|
assert result.replace("\\", "/") == "a/b/c"
|
|
|
|
|
|
def test_serialize_for_api_passthrough_for_primitives() -> None:
|
|
assert api_hooks._serialize_for_api(42) == 42
|
|
assert api_hooks._serialize_for_api("hello") == "hello"
|
|
assert api_hooks._serialize_for_api(None) is None
|
|
|
|
|
|
def test_serialize_for_api_handles_mixed_nesting() -> None:
|
|
obj = {"list": [1, 2, {"nested": "deep"}], "scalar": True}
|
|
result = api_hooks._serialize_for_api(obj)
|
|
assert result == obj
|
|
|
|
|
|
def test_get_app_attr_signature_preserved() -> None:
|
|
"""Pattern 4: _get_app_attr / _set_app_attr must NOT change signature."""
|
|
import inspect
|
|
sig = inspect.signature(api_hooks._get_app_attr)
|
|
params = list(sig.parameters.keys())
|
|
assert params == ["app", "name", "default"]
|
|
|
|
|
|
def test_set_app_attr_signature_preserved() -> None:
|
|
import inspect
|
|
sig = inspect.signature(api_hooks._set_app_attr)
|
|
params = list(sig.parameters.keys())
|
|
assert params == ["app", "name", "value"] |