151 lines
4.6 KiB
Python
151 lines
4.6 KiB
Python
import asyncio
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import anyio
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.actuators.store import ActuatorStore
|
|
from app.behavior.engine import BehaviorEngine
|
|
from app.ha.models import HaEntitySummary
|
|
from app.ha.reader import HaReader
|
|
from app.main import _ha_event_listener, app as fastapi_app, lifespan
|
|
|
|
|
|
class _FakeWebSocket:
|
|
def __init__(self, messages: list[str | BaseException]) -> None:
|
|
self._messages = messages
|
|
self.sent: list[dict[str, object]] = []
|
|
|
|
async def __aenter__(self) -> "_FakeWebSocket":
|
|
return self
|
|
|
|
async def __aexit__(self, *args: object) -> None:
|
|
return None
|
|
|
|
async def recv(self) -> str:
|
|
message = self._messages.pop(0)
|
|
if isinstance(message, BaseException):
|
|
raise message
|
|
return message
|
|
|
|
async def send(self, message: str) -> None:
|
|
import json
|
|
|
|
self.sent.append(json.loads(message))
|
|
|
|
|
|
class _RecordingBehaviorEngine(BehaviorEngine):
|
|
def __init__(self, tmp_path: Path) -> None:
|
|
super().__init__(
|
|
ha_reader=MagicMock(),
|
|
store=ActuatorStore(tmp_path / "actuators"),
|
|
settings=MagicMock(),
|
|
)
|
|
self.state_changes: list[
|
|
tuple[str, dict[str, object] | None, Sequence[HaEntitySummary] | None]
|
|
] = []
|
|
|
|
def handle_state_change(
|
|
self,
|
|
entity_id: str,
|
|
new_state: dict[str, object] | None,
|
|
*,
|
|
current_entities: Sequence[HaEntitySummary] | None = None,
|
|
) -> None:
|
|
self.state_changes.append((entity_id, new_state, current_entities))
|
|
|
|
|
|
class _FakeHaReader(HaReader):
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def read_entities(self) -> list[HaEntitySummary]:
|
|
return [
|
|
HaEntitySummary(
|
|
entity_id="light.test",
|
|
domain="light",
|
|
state="off",
|
|
)
|
|
]
|
|
|
|
|
|
def test_ha_event_listener_processes_state_change(tmp_path: Path) -> None:
|
|
async def run_test() -> None:
|
|
fake_ws = _FakeWebSocket(
|
|
[
|
|
'{"type":"auth_required"}',
|
|
'{"type":"auth_ok"}',
|
|
(
|
|
'{"type":"event","event":{"event_type":"state_changed",'
|
|
'"data":{"entity_id":"light.test","new_state":{"state":"on"}}}}'
|
|
),
|
|
asyncio.CancelledError(),
|
|
]
|
|
)
|
|
|
|
with patch("websockets.connect", return_value=fake_ws) as connect:
|
|
try:
|
|
await _ha_event_listener(mock_app, mock_client)
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
connect.assert_called_once_with(
|
|
"ws://homeassistant:8123/api/websocket",
|
|
ping_interval=20,
|
|
ping_timeout=10,
|
|
)
|
|
assert fake_ws.sent == [
|
|
{"type": "auth", "access_token": "test-token"},
|
|
{"id": 1, "type": "subscribe_events", "event_type": "state_changed"},
|
|
]
|
|
|
|
mock_app = MagicMock()
|
|
mock_app.state.settings = MagicMock()
|
|
mock_app.state.settings.ha_url = "http://homeassistant:8123"
|
|
mock_app.state.settings.ha_token = "test-token"
|
|
mock_app.state.ws_status = MagicMock()
|
|
mock_engine = _RecordingBehaviorEngine(tmp_path)
|
|
mock_app.state.behavior_engine = mock_engine
|
|
mock_app.state.ha_reader = _FakeHaReader()
|
|
mock_store = ActuatorStore(tmp_path / "store")
|
|
mock_store.configure("light.test")
|
|
mock_app.state.actuator_store = mock_store
|
|
mock_client = MagicMock()
|
|
|
|
anyio.run(run_test)
|
|
assert len(mock_engine.state_changes) == 1
|
|
entity_id, new_state, current_entities = mock_engine.state_changes[0]
|
|
assert entity_id == "light.test"
|
|
assert new_state == {"state": "on"}
|
|
assert current_entities == [
|
|
HaEntitySummary(entity_id="light.test", domain="light", state="on")
|
|
]
|
|
assert mock_app.state.ws_status.status == "connected"
|
|
assert mock_app.state.ws_status.error is None
|
|
|
|
|
|
def test_lifespan_skips_event_listener_without_ha_config() -> None:
|
|
app = FastAPI()
|
|
app.state.settings = MagicMock()
|
|
app.state.settings.ha_configured = False
|
|
|
|
async def run_test() -> None:
|
|
async with lifespan(app):
|
|
pass
|
|
|
|
anyio.run(run_test)
|
|
|
|
|
|
def test_websocket_health_returns_unavailable_without_listener() -> None:
|
|
with TestClient(fastapi_app) as client:
|
|
response = client.get("/health/websocket")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"status": "unavailable",
|
|
"error": "WebSocket-Listener nicht initialisiert",
|
|
}
|