Oskar Kapala
20f6761a67
- shared aiohttp ClientSession in HAClient (Phase 1 Flag #2 fixed): make_session() factory, session injected at startup, closed on shutdown - Check.run() → list[CheckResult]: clean multi-event interface - first real diagnostic check: entity unavailable > 24h (INSERT OR IGNORE baseline preserves first-seen timestamp) - root cause grouping: emit ha_integration_failed instead of N entity events when ≥50% of integration's entities are unavailable (≥3 min) - alert deduplication via SQLite cooldown window (default 6h) - recovery clears baseline + dedup for immediate re-alert - configurable thresholds: duration, integration %, cooldown - 38 unit tests + 7 integration tests (42 pass, 3 skip w/o live HA) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
"""Tests for HeartbeatCheck."""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from ha_diag.checks.heartbeat import HeartbeatCheck
|
|
from ha_diag.models import HAEventType, Severity
|
|
|
|
|
|
def _make_client(api_status=None, side_effect=None):
|
|
client = MagicMock()
|
|
if side_effect:
|
|
client.get_api_status = AsyncMock(side_effect=side_effect)
|
|
else:
|
|
client.get_api_status = AsyncMock(return_value=api_status)
|
|
return client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_heartbeat_ok_returns_empty_list():
|
|
client = _make_client(api_status={"message": "API running."})
|
|
check = HeartbeatCheck(client)
|
|
results = await check.run()
|
|
assert results == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_heartbeat_connection_error():
|
|
client = _make_client(side_effect=ConnectionError("refused"))
|
|
check = HeartbeatCheck(client)
|
|
results = await check.run()
|
|
assert len(results) == 1
|
|
assert results[0].healthy is False
|
|
assert results[0].event_type == HAEventType.ha_websocket_dead
|
|
assert results[0].severity == Severity.error
|
|
assert "refused" in results[0].message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_heartbeat_unexpected_response():
|
|
client = _make_client(api_status={"unexpected": "key"})
|
|
check = HeartbeatCheck(client)
|
|
results = await check.run()
|
|
assert len(results) == 1
|
|
assert results[0].event_type == HAEventType.ha_websocket_dead
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_heartbeat_timeout():
|
|
client = _make_client(side_effect=TimeoutError("timed out"))
|
|
check = HeartbeatCheck(client)
|
|
results = await check.run()
|
|
assert len(results) == 1
|
|
assert results[0].event_type == HAEventType.ha_websocket_dead
|
|
assert "timed out" in results[0].message
|
|
|
|
|
|
def test_heartbeat_check_name():
|
|
check = HeartbeatCheck(MagicMock())
|
|
assert check.name == "heartbeat"
|