Maniacs_Mansion/NestQuest
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ec7c8d4bc2a1484effdb8de6d18d22952c433ee9
NestQuest/tests/test_coordinator.py
T
overseer ec7c8d4bc2 Coordinator polls the API snapshot and rebuilds the sensors' data 
Feature 18: the shared coordinator's refresh pass now fetches
GET /api/v1/panel/snapshot through the NestQuestApiClient and
reconstructs the exact core dataclasses (NestQuestSnapshot,
ChildDaySnapshot, QuestInstanceView) the Feature 10 entities read, so
sensor.py and binary_sensor.py are untouched and every entity_id and
attribute payload keeps its shape.

Why: the integration becomes an API client of the NestQuest service
instead of a second reader of the SQLite file, and the panel cards and
the four blueprints keep working without changes.

- snapshot_from_api_payload maps the route's JSON back onto the
  dataclasses field by field (state 'completed' -> 'done' so
  instance_payload renders 'completed' again; counts, presence,
  next_present, overdue, on_time, completed_at pass through verbatim;
  due_date is stamped from today_iso, the only date the route serves).
- The panel payload omits missed instances (D-009), so
  admin_instances can no longer carry missed rows — the admin surface
  for missed quests is the PWA reading the API directly. Documented in
  the module docstring and pinned by tests.
- An unconfigured panel token (the shipped default) gets a typed-error
  stub instead of a construction crash: setup succeeds, the first
  refresh fails cleanly, entities come up unavailable and HA's retry
  cycle applies. Fetch failures log the typed error and mark the
  entities unavailable while data keeps its last good snapshot.
- The runtime record registers before the first refresh so the refresh
  path always finds the live entry state; the update interval and the
  serialized forced-refresh behaviour are unchanged.
- Tests: a fixture snapshot drives the coordinator (stub client, no
  network) asserting per-child values, household rollups, cycle day,
  and the exact instance payload shape, plus unconfigured and
  failed-fetch cases; DB-seeded tests ride a lazy local-snapshot
  client through the same JSON path.
2026-09-23 06:02:26 -04:00

531 lines
19 KiB
Python
Raw Blame History

"""Tests for the shared NestQuest coordinator (Feature 10, Feature 18).
The coordinator now polls the API service's panel snapshot through the
API client (Feature 18) and reconstructs the core snapshot dataclasses
every entity reads. The tests stub the client — NO network, NO aiohttp
— with a FIXTURE snapshot payload (the documented shape of the Feature
16 snapshot route, ``api/routes_panel.py``) and assert the
reconstruction: per-child counts and presence, the instance views
(``completed`` mapped back to ``done``), the household rollups, the
cycle day, and the unconfigured / failed-fetch behaviour.
"""
from __future__ import annotations
import datetime
import pytest
from conftest import set_coordinator_client, wire_entry_to_registry
from custom_components.nestquest import async_setup_entry, async_unload_entry
from custom_components.nestquest.api_client import (
NestQuestApiClient,
NestQuestApiError,
)
from custom_components.nestquest.const import (
CONF_UPDATE_INTERVAL,
DEFAULT_UPDATE_INTERVAL,
DOMAIN,
)
from custom_components.nestquest.coordinator import (
NestQuestCoordinator,
coordinator_client_from_entry,
snapshot_from_api_payload,
)
from custom_components.nestquest.core.snapshot import instance_payload
def _today_iso() -> str:
return datetime.date.today().isoformat()
class StubSnapshotClient:
"""A no-network stand-in for the API client's snapshot surface.
Answers ``get_snapshot()`` from a scripted queue of outcomes — a
payload dict or an exception per call, consumed in order — and
records every call, so each test drives exactly one refresh.
"""
def __init__(self, *outcomes) -> None:
self.calls = 0
self._outcomes = list(outcomes)
self._last = outcomes[-1] if outcomes else None
async def get_snapshot(self):
self.calls += 1
if not self._outcomes:
# Re-serve the last outcome: setup's first refresh and the
# test's explicit pass both poll the same scripted state.
outcome = self._last
else:
outcome = self._outcomes.pop(0)
self._last = outcome
if isinstance(outcome, BaseException):
raise outcome
return outcome
#: A fixture panel snapshot (the documented Feature 16 route shape):
#: two children — Ada owes one, owes none of it yet, is away today with
#: a returns date; Bo completed his only quest and is present.
FIXTURE_SNAPSHOT = {
"today_iso": _today_iso(),
"cycle_day": 3,
"children": [
{
"child_id": 1,
"child_name": "Ada",
"present": False,
"next_present": "2026-09-24",
"due_today": 2,
"completed_today": 1,
"remaining_today": 1,
"completion_pct": 50,
"instances": [
{
"id": 7,
"definition_id": 3,
"child_id": 1,
"title": "Brush teeth",
"icon": "🦷",
"window": "morning",
"due_time": "08:00",
"state": "completed",
"overdue": False,
"completed_at": f"{_today_iso()}T07:55:00+00:00",
"on_time": True,
},
{
"id": 8,
"definition_id": 3,
"child_id": 1,
"title": "Pack bag",
"icon": None,
"window": "morning",
"due_time": "10:00",
"state": "open",
"overdue": True,
"completed_at": None,
"on_time": None,
},
],
},
{
"child_id": 2,
"child_name": "Bo",
"present": True,
"next_present": None,
"due_today": 1,
"completed_today": 1,
"remaining_today": 0,
"completion_pct": 100,
"instances": [
{
"id": 9,
"definition_id": 4,
"child_id": 2,
"title": "Make bed",
"icon": "🛏",
"window": "morning",
"due_time": "07:30",
"state": "completed",
"overdue": False,
"completed_at": f"{_today_iso()}T07:31:00+00:00",
"on_time": True,
}
],
},
],
}
async def _wire_entry(hass, make_entry, client, options=None):
"""Build an entry whose coordinator uses the scripted API client.
The client is registered with the conftest override BEFORE setup:
the first refresh at setup already polls the snapshot, so the
platform setup sees real data and creates the whole entity set.
"""
from conftest import set_coordinator_client
entry = wire_entry_to_registry(
make_entry(
options=options,
data={"api_base_url": "http://api.test:8000", "panel_token": "tok"},
),
hass.registry,
)
set_coordinator_client(entry, client)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
assert coordinator.api_client is client
return entry, coordinator
async def test_coordinator_builds_snapshot_from_fixture(hass, make_entry) -> None:
"""One refresh = one snapshot GET; the fixture's values land on the
snapshot the entities read, per child and household-wide."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
calls_after_setup = client.calls
snapshot = await coordinator._async_update_data()
assert client.calls == calls_after_setup + 1
assert snapshot.today_iso == _today_iso()
assert snapshot.cycle_day == 3
assert len(snapshot.children) == 2
ada, bo = snapshot.children
assert (ada.child_id, ada.child_name) == (1, "Ada")
assert ada.due_today == 2
assert ada.completed_today == 1
assert ada.remaining_today == 1
assert ada.completion_pct == 50
assert ada.present is False
assert ada.next_present == "2026-09-24"
done_view, open_view = ada.instances
assert done_view.instance_id == 7
assert done_view.title == "Brush teeth"
assert done_view.icon == "🦷"
assert done_view.window == "morning"
assert done_view.due_time == "08:00"
assert done_view.due_date == _today_iso()
# The API's "completed" maps back to the core view's "done".
assert done_view.state == "done"
assert done_view.overdue is False
assert done_view.completed_at == f"{_today_iso()}T07:55:00+00:00"
assert done_view.was_on_time is True
assert open_view.state == "open"
assert open_view.overdue is True
assert open_view.completed_at is None
assert open_view.was_on_time is None
assert (bo.child_id, bo.child_name) == (2, "Bo")
assert bo.due_today == 1
assert bo.completed_today == 1
assert bo.remaining_today == 0
assert bo.completion_pct == 100
assert bo.present is True
assert bo.next_present is None
# Household rollups: the sums the household sensors compute.
assert sum(child.due_today for child in snapshot.children) == 3
assert sum(child.completed_today for child in snapshot.children) == 2
assert sum(child.remaining_today for child in snapshot.children) == 1
async def test_coordinator_refresh_publishes_to_listeners(hass, make_entry) -> None:
"""The refresh pass notifies listeners and reports success — the
ONE refresh pass every entity reads from."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
called = []
coordinator.async_add_listener(lambda: called.append("entity"))
await coordinator.async_refresh()
assert called == ["entity"]
assert coordinator.last_update_success is True
assert coordinator.data is not None
assert coordinator.data.children[0].child_name == "Ada"
async def test_instance_payload_shape_is_unchanged(hass, make_entry) -> None:
"""The per-child instance attribute payload the due-today sensor
emits keeps its exact documented shape, and admin_instances now
equals instances (the API omits missed rows — D-009; the admin
surface for missed quests is the PWA)."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
snapshot = await coordinator._async_update_data()
ada = snapshot.children[0]
panel = instance_payload(ada.instances, include_missed=False)
assert panel == [
{
"id": 7,
"definition_id": 3,
"child_id": 1,
"title": "Brush teeth",
"icon": "🦷",
"window": "morning",
"due_time": "08:00",
"state": "completed",
"overdue": False,
"completed_at": f"{_today_iso()}T07:55:00+00:00",
"on_time": True,
},
{
"id": 8,
"definition_id": 3,
"child_id": 1,
"title": "Pack bag",
"icon": None,
"window": "morning",
"due_time": "10:00",
"state": "open",
"overdue": True,
"completed_at": None,
"on_time": None,
},
]
# The API payload omits missed rows, so include_missed cannot
# re-add them: admin_instances == instances, byte for byte.
assert instance_payload(ada.instances, include_missed=True) == panel
async def test_coordinator_entities_read_the_api_snapshot(hass, make_entry) -> None:
"""The registered sensors and binary sensors read the rebuilt
snapshot: entity values come from the fixture, with the SAME
entity_ids as ever."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
entry, coordinator = await _wire_entry(hass, make_entry, client)
await coordinator.async_refresh()
ada_due = hass.entities[f"{DOMAIN}_child_1_quests_due_today"]
assert ada_due.native_value == 2
attributes = ada_due.extra_state_attributes
assert attributes["child_id"] == 1
assert attributes["child_name"] == "Ada"
assert attributes["present"] is False
assert [row["id"] for row in attributes["instances"]] == [7, 8]
assert (
hass.entities[f"{DOMAIN}_child_1_quests_completed_today"].native_value
== 1
)
assert (
hass.entities[f"{DOMAIN}_child_1_quests_remaining_today"].native_value
== 1
)
assert (
hass.entities[f"{DOMAIN}_child_1_completion_pct_today"].native_value
== 50
)
assert hass.entities[f"{DOMAIN}_child_1_next_quest"].native_value == (
"Pack bag"
)
# Ada owes one still: not all done; away today.
assert hass.entities[f"{DOMAIN}_child_1_all_done"].is_on is False
assert hass.entities[f"{DOMAIN}_child_1_present_today"].is_on is False
assert (
hass.entities[f"{DOMAIN}_child_2_completion_pct_today"].native_value
== 100
)
# Bo had a quest and owes none of it: all done, present today.
assert hass.entities[f"{DOMAIN}_child_2_all_done"].is_on is True
assert hass.entities[f"{DOMAIN}_child_2_present_today"].is_on is True
bo_attributes = hass.entities[
f"{DOMAIN}_child_2_present_today"
].extra_state_attributes
assert bo_attributes["next_present"] is None
assert (
hass.entities[f"{DOMAIN}_household_quests_due_today"].native_value == 3
)
assert (
hass.entities[f"{DOMAIN}_household_quests_completed_today"].native_value
== 2
)
cycle = hass.entities[f"{DOMAIN}_cycle_day"]
assert cycle.native_value == 3
assert cycle.extra_state_attributes == {
"cycle_day": 3,
"today": _today_iso(),
}
await async_unload_entry(hass, entry)
async def test_coordinator_failed_fetch_marks_unavailable(hass, make_entry) -> None:
"""A failed fetch (transport failure or non-2xx — both typed) raises
out of the refresh pass: last_update_success goes False (entities
unavailable), data keeps its last good value, and the typed error is
logged, never swallowed."""
error = NestQuestApiError(
"NestQuest API GET /api/v1/panel/snapshot failed with HTTP 503",
status=503,
)
# The stub serves the good snapshot for setup's first refresh, then
# the scripted failure.
client = StubSnapshotClient(FIXTURE_SNAPSHOT, error)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
# First refresh (at setup) served the good snapshot.
assert coordinator.data is not None
assert coordinator.last_update_success is True
# The next pass hits the scripted failure: it raises out of the
# refresh, marks the entities unavailable, and keeps the last GOOD
# snapshot as data.
with pytest.raises(NestQuestApiError):
await coordinator.async_refresh()
assert coordinator.last_update_success is False
assert coordinator.data.children[0].child_name == "Ada"
async def test_coordinator_unconfigured_api_does_not_crash_setup(
hass, make_entry, monkeypatch
) -> None:
"""An entry with no panel token set up fine: the coordinator gets
the typed-error stub, the first refresh fails cleanly (unavailable
entities), and setup does NOT crash."""
# Restore the production factory: the conftest default would serve
# the local snapshot, but this test exercises the token guard.
import custom_components.nestquest as nq
monkeypatch.setattr(
nq, "coordinator_client_from_entry", coordinator_client_from_entry
)
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
from custom_components.nestquest.coordinator import _UnconfiguredApiClient
assert isinstance(coordinator.api_client, _UnconfiguredApiClient)
# A real NestQuestApiClient was never built (it would ValueError).
assert not isinstance(coordinator.api_client, NestQuestApiClient)
assert coordinator.last_update_success is False
assert coordinator.data is None
with pytest.raises(NestQuestApiError, match="not configured"):
await coordinator.async_refresh()
async def test_coordinator_configured_token_builds_real_client(
hass, make_entry
) -> None:
"""An entry WITH a panel token builds the real client through the
production factory (a stub transport injected via session_factory,
the mock-only harness having no aiohttp module to hand in), and the
coordinator's refresh polls the snapshot route end to end."""
import json as json_module
class _OneShotSession:
"""A session-shaped stub answering every request with the fixture."""
def __init__(self, payload) -> None:
self._body = json_module.dumps(payload)
def request(self, method, url, **kwargs):
body = self._body
class _Response:
def __init__(self) -> None:
self.status = 200
async def text(self):
return body
class _Context:
async def __aenter__(self):
return _Response()
async def __aexit__(self, *exc):
return False
return _Context()
def _session_factory(_hass):
return _OneShotSession(FIXTURE_SNAPSHOT)
entry = wire_entry_to_registry(
make_entry(
data={"api_base_url": "http://api.test:8000", "panel_token": "tok"}
),
hass.registry,
)
real_client = coordinator_client_from_entry(
hass, entry, session_factory=_session_factory
)
assert isinstance(real_client, NestQuestApiClient)
set_coordinator_client(entry, real_client)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
assert isinstance(coordinator.api_client, NestQuestApiClient)
# The real client polls the stub transport; the snapshot lands.
await coordinator.async_refresh()
assert coordinator.data.children[0].child_name == "Ada"
async def test_setup_creates_coordinator_with_configured_interval(
hass, make_entry
) -> None:
entry = wire_entry_to_registry(
make_entry(options={CONF_UPDATE_INTERVAL: 60}), hass.registry
)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
assert isinstance(coordinator, NestQuestCoordinator)
assert coordinator.update_interval == datetime.timedelta(seconds=60)
async def test_setup_defaults_to_five_minute_interval(hass, make_entry) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
assert coordinator.update_interval == datetime.timedelta(
seconds=DEFAULT_UPDATE_INTERVAL
)
async def test_unload_shuts_coordinator_down(hass, make_entry) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
torn_down = []
async def _shutdown():
torn_down.append(coordinator)
coordinator.async_shutdown = _shutdown
assert await async_unload_entry(hass, entry) is True
assert torn_down == [coordinator]
async def test_failed_first_refresh_shuts_coordinator_down(
hass, make_entry, monkeypatch
) -> None:
"""A setup failure after coordinator creation must not leak its
listeners."""
from custom_components.nestquest import async_setup_entry
import custom_components.nestquest.coordinator as coordinator_module
shutdown_calls = []
async def _record_shutdown(self):
shutdown_calls.append(self.entry_id)
monkeypatch.setattr(
coordinator_module.NestQuestCoordinator,
"async_shutdown",
_record_shutdown,
)
async def _boom(self):
raise RuntimeError("refresh failed")
monkeypatch.setattr(
coordinator_module.NestQuestCoordinator,
"_async_update_data",
_boom,
)
entry = wire_entry_to_registry(make_entry(), hass.registry)
with pytest.raises(RuntimeError, match="refresh failed"):
await async_setup_entry(hass, entry)
assert shutdown_calls == [entry.entry_id]
assert getattr(entry, "runtime_data", None) is None
async def test_snapshot_from_api_payload_empty_household() -> None:
"""An empty household (no children) is a valid snapshot: empty
children tuple, cycle day passes through."""
snapshot = snapshot_from_api_payload(
{"today_iso": "2026-09-23", "cycle_day": 0, "children": []}
)
assert snapshot.children == ()
assert snapshot.cycle_day == 0
assert snapshot.today_iso == "2026-09-23"
Reference in New Issue View Git Blame Copy Permalink