Oskar Kapala
3499b2f280
New checks: - SystemHealthCheck (15min interval): detects newly-failing HA integrations via /api/system_health snapshot diff; transition-based dedup (ok→error fires, sustained error silent, error→ok clears alert) - UpdatesAvailableCheck (daily cron 09:00): per-update ha_update_available events with 7-day dedup; release notes truncated at 2000 chars - UpdatesDigestCheck (Sunday cron 09:00): single digest event with all pending updates; weekly ISO-week dedup, independent of daily dedup key - AutomationFailuresCheck (30min interval): detects automations with N consecutive failures (default 3) via /api/trace/automation/<id>; 6h cooldown per automation Phase 3 flag fixes: - Flag #1 (since field): UnavailableEntitiesCheck now uses min(state.last_changed, baseline.first_seen) as effective "since", giving accurate duration when agent was offline at entity's first fail - Flag #3 (registry cache): HAClient.get_entity_registry() caches response in-process with configurable TTL (default 300s); avoids repeated API calls across concurrent check cycles; invalidate_registry_cache() for manual invalidation Storage: system_health_snapshot table (component, last_status, last_seen_at, payload) created automatically on next Storage.open() call Config additions (all with defaults): entity_registry_cache_ttl=300, system_health_check_interval=900, automation_check_interval=1800, automation_failure_threshold=3, updates_check_hour=9, updates_check_minute=0, updates_cooldown_days=7 Tests: 95 unit tests pass (49 new), 13 integration tests pass (9 new); 3 skipped (live-HA token not set in CI) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
257 lines
8.5 KiB
Python
257 lines
8.5 KiB
Python
"""Unit tests for UpdatesAvailableCheck and UpdatesDigestCheck."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from ha_diag.checks.updates_available import (
|
|
UpdatesAvailableCheck,
|
|
UpdatesDigestCheck,
|
|
_build_update_payload,
|
|
)
|
|
from ha_diag.config import Settings
|
|
from ha_diag.models import HAEventType
|
|
from ha_diag.storage import Storage
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_settings(**overrides) -> Settings:
|
|
defaults: dict = {
|
|
"ha_url": "http://test.local:8123",
|
|
"ha_token": "test",
|
|
"node_name": "test-node",
|
|
"location_tag": "test-loc",
|
|
"alert_cooldown_hours": 0.0,
|
|
"updates_cooldown_days": 0, # no dedup in most tests
|
|
"check_interval": 60,
|
|
"check_interval_unavailable": 3600,
|
|
}
|
|
defaults.update(overrides)
|
|
return Settings(**defaults)
|
|
|
|
|
|
def _make_client(states=None, error=None):
|
|
client = MagicMock()
|
|
if error:
|
|
client.get_states = AsyncMock(side_effect=error)
|
|
else:
|
|
client.get_states = AsyncMock(return_value=states or [])
|
|
return client
|
|
|
|
|
|
def _update_state(
|
|
entity_id: str = "update.homeassistant_core",
|
|
state: str = "on",
|
|
title: str = "Home Assistant Core",
|
|
installed: str = "2025.5.0",
|
|
latest: str = "2025.6.0",
|
|
release_summary: str | None = None,
|
|
release_url: str | None = None,
|
|
) -> dict:
|
|
attrs: dict = {
|
|
"title": title,
|
|
"installed_version": installed,
|
|
"latest_version": latest,
|
|
"in_progress": False,
|
|
"auto_update": False,
|
|
}
|
|
if release_summary:
|
|
attrs["release_summary"] = release_summary
|
|
if release_url:
|
|
attrs["release_url"] = release_url
|
|
return {"entity_id": entity_id, "state": state, "attributes": attrs}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _build_update_payload helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_update_payload_basic():
|
|
attrs = {"title": "HA Core", "installed_version": "1.0", "latest_version": "2.0"}
|
|
p = _build_update_payload("update.ha_core", attrs)
|
|
assert p["entity_id"] == "update.ha_core"
|
|
assert p["title"] == "HA Core"
|
|
assert p["installed_version"] == "1.0"
|
|
assert p["latest_version"] == "2.0"
|
|
|
|
|
|
def test_build_update_payload_release_summary_truncated():
|
|
long_notes = "x" * 3000
|
|
attrs = {"release_summary": long_notes}
|
|
p = _build_update_payload("update.ha_core", attrs)
|
|
assert len(p["release_summary"]) == 2000
|
|
|
|
|
|
def test_build_update_payload_release_url_omitted_when_absent():
|
|
p = _build_update_payload("update.ha_core", {})
|
|
assert "release_url" not in p
|
|
|
|
|
|
def test_build_update_payload_release_url_included_when_present():
|
|
attrs = {"release_url": "https://github.com/..."}
|
|
p = _build_update_payload("update.x", attrs)
|
|
assert p["release_url"] == "https://github.com/..."
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UpdatesAvailableCheck (daily individual events)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_updates_returns_empty(storage: Storage):
|
|
states = [{"entity_id": "light.living_room", "state": "on", "attributes": {}}]
|
|
check = UpdatesAvailableCheck(_make_client(states), storage, _make_settings())
|
|
assert await check.run() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_off_state_not_emitted(storage: Storage):
|
|
states = [_update_state(state="off")]
|
|
check = UpdatesAvailableCheck(_make_client(states), storage, _make_settings())
|
|
assert await check.run() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_update_emits_event(storage: Storage):
|
|
states = [_update_state()]
|
|
check = UpdatesAvailableCheck(_make_client(states), storage, _make_settings())
|
|
results = await check.run()
|
|
assert len(results) == 1
|
|
assert results[0].event_type == HAEventType.ha_update_available
|
|
assert "2025.5.0" in results[0].message
|
|
assert "2025.6.0" in results[0].message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_updates_emit_multiple_events(storage: Storage):
|
|
states = [
|
|
_update_state("update.ha_core"),
|
|
_update_state("update.mosquitto", title="Mosquitto"),
|
|
]
|
|
check = UpdatesAvailableCheck(_make_client(states), storage, _make_settings())
|
|
results = await check.run()
|
|
assert len(results) == 2
|
|
assert all(r.event_type == HAEventType.ha_update_available for r in results)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cooldown_prevents_same_update_next_day(storage: Storage):
|
|
states = [_update_state()]
|
|
settings = _make_settings(updates_cooldown_days=7)
|
|
check = UpdatesAvailableCheck(_make_client(states), storage, settings)
|
|
r1 = await check.run()
|
|
r2 = await check.run()
|
|
assert len(r1) == 1
|
|
assert r2 == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_cooldown_allows_repeat(storage: Storage):
|
|
states = [_update_state()]
|
|
check = UpdatesAvailableCheck(_make_client(states), storage, _make_settings(updates_cooldown_days=0))
|
|
r1 = await check.run()
|
|
r2 = await check.run()
|
|
assert len(r1) == 1
|
|
assert len(r2) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_payload_contains_version_fields(storage: Storage):
|
|
states = [_update_state(installed="2025.5.0", latest="2025.6.0")]
|
|
check = UpdatesAvailableCheck(_make_client(states), storage, _make_settings())
|
|
results = await check.run()
|
|
p = results[0].payload
|
|
assert p["installed_version"] == "2025.5.0"
|
|
assert p["latest_version"] == "2025.6.0"
|
|
assert p["in_progress"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ha_error_returns_empty(storage: Storage):
|
|
check = UpdatesAvailableCheck(
|
|
_make_client(error=ConnectionError("HA down")), storage, _make_settings()
|
|
)
|
|
assert await check.run() == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UpdatesDigestCheck (Sunday digest)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_no_updates_returns_empty(storage: Storage):
|
|
check = UpdatesDigestCheck(_make_client([]), storage, _make_settings())
|
|
assert await check.run() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_emits_single_event_for_all_updates(storage: Storage):
|
|
states = [
|
|
_update_state("update.ha_core"),
|
|
_update_state("update.mosquitto", title="Mosquitto"),
|
|
_update_state("update.esphome", title="ESPHome"),
|
|
]
|
|
check = UpdatesDigestCheck(_make_client(states), storage, _make_settings())
|
|
results = await check.run()
|
|
assert len(results) == 1
|
|
p = results[0].payload
|
|
assert p["digest"] is True
|
|
assert p["count"] == 3
|
|
assert len(p["updates"]) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_payload_has_digest_true(storage: Storage):
|
|
states = [_update_state()]
|
|
check = UpdatesDigestCheck(_make_client(states), storage, _make_settings())
|
|
results = await check.run()
|
|
assert results[0].payload["digest"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_weekly_dedup_prevents_same_week_refiring(storage: Storage):
|
|
states = [_update_state()]
|
|
check = UpdatesDigestCheck(_make_client(states), storage, _make_settings())
|
|
r1 = await check.run()
|
|
r2 = await check.run()
|
|
assert len(r1) == 1
|
|
assert r2 == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_fires_independently_of_daily_dedup(storage: Storage):
|
|
"""Daily cooldown on entity X doesn't suppress Sunday digest."""
|
|
states = [_update_state()]
|
|
settings = _make_settings(updates_cooldown_days=7)
|
|
|
|
# Daily check marks alert_key="update_available:update.homeassistant_core"
|
|
daily = UpdatesAvailableCheck(_make_client(states), storage, settings)
|
|
await daily.run()
|
|
|
|
# Digest uses different key "update_digest:{week}" — should still fire
|
|
digest = UpdatesDigestCheck(_make_client(states), storage, settings)
|
|
r = await digest.run()
|
|
assert len(r) == 1
|
|
assert r[0].payload["digest"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_name_is_updates_digest(storage: Storage):
|
|
check = UpdatesDigestCheck(_make_client([]), storage, _make_settings())
|
|
assert check.name == "updates_digest"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_daily_check_name_is_updates_available(storage: Storage):
|
|
check = UpdatesAvailableCheck(_make_client([]), storage, _make_settings())
|
|
assert check.name == "updates_available"
|