113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
import asyncio
|
|
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.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]] = []
|
|
|
|
def handle_state_change(self, entity_id: str, new_state: dict[str, object] | None) -> None:
|
|
self.state_changes.append((entity_id, new_state))
|
|
|
|
|
|
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",'
|
|
'"entity_id":"light.test","new_state":{"state":"on"}}}'
|
|
),
|
|
asyncio.CancelledError(),
|
|
]
|
|
)
|
|
|
|
with patch("websockets.connect", return_value=fake_ws):
|
|
try:
|
|
await _ha_event_listener(mock_app, mock_client)
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
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_store = ActuatorStore(tmp_path / "store")
|
|
mock_app.state.actuator_store = mock_store
|
|
mock_client = MagicMock()
|
|
|
|
anyio.run(run_test)
|
|
assert mock_engine.state_changes == [("light.test", {"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",
|
|
}
|