51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.main import _ha_event_listener, lifespan
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ha_event_listener_processes_state_change() -> None:
|
|
# Mock settings, store, engine
|
|
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_engine = MagicMock()
|
|
mock_engine.handle_state_change = MagicMock()
|
|
mock_app.state.behavior_engine = mock_engine
|
|
mock_store = MagicMock()
|
|
mock_app.state.actuator_store = mock_store
|
|
|
|
mock_client = MagicMock()
|
|
|
|
# Mock websockets connection and messages
|
|
mock_ws = AsyncMock()
|
|
mock_ws.recv.side_effect = [
|
|
'{"type":"auth_ok"}', # auth response
|
|
'{"type":"event","event":{"event_type":"state_changed","entity_id":"light.test","new_state":{"state":"on"}}}', # event
|
|
Exception("EOF"), # break loop
|
|
]
|
|
mock_ws.send.return_value = None
|
|
|
|
with patch("websockets.connect", return_value=mock_ws):
|
|
await _ha_event_listener(mock_app, mock_client)
|
|
|
|
mock_engine.handle_state_change.assert_called_once_with("light.test", {"state": "on"})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lifespan_starts_event_listener() -> None:
|
|
# This is a basic smoke test that lifespan creates the expected tasks
|
|
from fastapi import FastAPI
|
|
|
|
app = FastAPI()
|
|
app.state.settings = MagicMock()
|
|
app.state.settings.ha_configured = False
|
|
|
|
# Should not raise
|
|
async with lifespan(app):
|
|
pass
|