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>
81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""Tests for HAClient using aioresponses to mock aiohttp."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from aioresponses import aioresponses
|
|
|
|
from ha_diag.ha_client import HAClient, make_session
|
|
|
|
HA_URL = "http://homeassistant.test:8123"
|
|
TOKEN = "test-token"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_api_status_ok():
|
|
with aioresponses() as m:
|
|
m.get(f"{HA_URL}/api/", payload={"message": "API running."})
|
|
async with make_session(TOKEN) as session:
|
|
client = HAClient(HA_URL, session)
|
|
result = await client.get_api_status()
|
|
assert result == {"message": "API running."}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_api_status_unauthorized():
|
|
with aioresponses() as m:
|
|
m.get(f"{HA_URL}/api/", status=401)
|
|
async with make_session(TOKEN) as session:
|
|
client = HAClient(HA_URL, session)
|
|
with pytest.raises(Exception):
|
|
await client.get_api_status()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_states_returns_list():
|
|
payload = [{"entity_id": "light.living_room", "state": "on"}]
|
|
with aioresponses() as m:
|
|
m.get(f"{HA_URL}/api/states", payload=payload)
|
|
async with make_session(TOKEN) as session:
|
|
client = HAClient(HA_URL, session)
|
|
states = await client.get_states()
|
|
assert isinstance(states, list)
|
|
assert states[0]["entity_id"] == "light.living_room"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_config_returns_dict():
|
|
payload = {"version": "2024.1.0", "location_name": "Home"}
|
|
with aioresponses() as m:
|
|
m.get(f"{HA_URL}/api/config", payload=payload)
|
|
async with make_session(TOKEN) as session:
|
|
client = HAClient(HA_URL, session)
|
|
config = await client.get_config()
|
|
assert config["version"] == "2024.1.0"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_entity_registry_returns_list():
|
|
payload = [
|
|
{"entity_id": "light.hall", "platform": "zha", "area_id": "hallway"},
|
|
{"entity_id": "sensor.temp", "platform": "mqtt", "area_id": None},
|
|
]
|
|
with aioresponses() as m:
|
|
m.get(f"{HA_URL}/api/config/entity_registry", payload=payload)
|
|
async with make_session(TOKEN) as session:
|
|
client = HAClient(HA_URL, session)
|
|
registry = await client.get_entity_registry()
|
|
assert len(registry) == 2
|
|
assert registry[0]["platform"] == "zha"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_make_session_sets_auth_header():
|
|
"""make_session injects the Bearer token in all requests."""
|
|
with aioresponses() as m:
|
|
m.get(f"{HA_URL}/api/", payload={"message": "API running."})
|
|
async with make_session("my-secret-token") as session:
|
|
client = HAClient(HA_URL, session)
|
|
await client.get_api_status()
|
|
# Verify the Authorization header was sent
|
|
assert session.headers.get("Authorization") == "Bearer my-secret-token"
|