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"
|