Oskar Kapala
ab8895d28b
- new per-host service, follows node-agent pattern - 7 new HA event types defined (routing in supervisor — Phase 5) - HeartbeatCheck as pipeline validator (pings /api/, emits ha_websocket_dead) - service.yaml + host configs for piha (ken) and chelsty-infra (chelsty) - test scaffolding with aiohttp/aiosqlite mocks (15/15 passing) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
66 lines
1.9 KiB
Python
66 lines
1.9 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()
|
|
client.__aenter__ = AsyncMock(return_value=client)
|
|
client.__aexit__ = AsyncMock(return_value=None)
|
|
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():
|
|
client = _make_client(api_status={"message": "API running."})
|
|
check = HeartbeatCheck(client)
|
|
result = await check.run()
|
|
assert result.healthy is True
|
|
assert result.event_type is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_heartbeat_connection_error():
|
|
client = _make_client(side_effect=ConnectionError("refused"))
|
|
check = HeartbeatCheck(client)
|
|
result = await check.run()
|
|
assert result.healthy is False
|
|
assert result.event_type == HAEventType.ha_websocket_dead
|
|
assert result.severity == Severity.error
|
|
assert "refused" in result.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_heartbeat_unexpected_response():
|
|
client = _make_client(api_status={"unexpected": "key"})
|
|
check = HeartbeatCheck(client)
|
|
result = await check.run()
|
|
assert result.healthy is False
|
|
assert result.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)
|
|
result = await check.run()
|
|
assert result.healthy is False
|
|
assert result.event_type == HAEventType.ha_websocket_dead
|
|
assert "timed out" in result.message
|
|
|
|
|
|
def test_heartbeat_check_name():
|
|
client = MagicMock()
|
|
check = HeartbeatCheck(client)
|
|
assert check.name == "heartbeat"
|