SSE: publish uncompleted, child-day-complete and missed (#155)

This commit was merged in pull request #155.
This commit is contained in:
2026-09-22 23:25:06 -04:00
parent a07ce1b2f2
commit d4a55b6b40
5 changed files with 311 additions and 6 deletions
+14 -1
View File
@@ -69,6 +69,7 @@ from api.nestquest_core import (
core_settings,
core_snapshot,
)
from api import transitions as api_transitions
#: The instance DAO, imported on the app's core copy for the
#: child-mismatch guard on the complete route.
@@ -283,13 +284,14 @@ async def panel_complete_instance(
detail="Quest instance not found for this child",
)
now = _local_now()
try:
result = await core_completion.complete_instance(
database,
instance_id,
actor_source="panel",
actor_child_id=body.actor_child_id,
now=_local_now(),
now=now,
)
except ValueError as error:
message = str(error)
@@ -312,6 +314,17 @@ async def panel_complete_instance(
database, instance_id, was_on_time=result.was_on_time
)
request.app.state.publisher.publish(event_type, payload)
# THEN evaluate the day-complete rule off the completed event's
# payload (the core builder decides when the child's whole day
# is cleared — a zero-quest day and a non-today instance never
# fire it). ``today`` comes from the route's single clock read
# so the calendar date can never disagree with the completion's.
await api_transitions.publish_child_day_complete(
database,
payload,
now.date(),
request.app.state.publisher,
)
return {"status": str(result)}
+83
View File
@@ -0,0 +1,83 @@
"""Reusable SSE transition publishers for the API service (task 44c93cfd).
Each helper wraps the ONE publisher on ``app.state.publisher`` with a
single documented transition: it builds the payload through the core
event builders (``nestquest_core.events``) — never hand-built here —
and fans it out with :meth:`~api.events.TransitionPublisher.publish`.
The uncomplete route (follow-up task da0226b3) and the API-side missed
sweep do not exist yet; these helpers are the functions they will call,
and the SSE tests exercise them directly. Fan-out stays fire-and-forget
(see api/events.py): a slow subscriber drops the event and recovers by
polling the snapshot.
"""
from __future__ import annotations
import datetime
from api.nestquest_core import core_events
async def publish_quest_uncompleted(
database: object, instance_id: int, publisher: object
) -> None:
"""Publish ONE ``nestquest_quest_uncompleted`` event for an actual
reversal of ``instance_id``.
The payload is built by the core builder
(:func:`nestquest_core.events.build_quest_uncompleted_events`);
this helper only picks the single event out of the builder's list
and publishes it.
"""
events = await core_events.build_quest_uncompleted_events(
database, instance_id
)
for event_type, payload in events:
publisher.publish(event_type, payload)
async def publish_quest_missed(
database: object, instance_id: int, publisher: object
) -> None:
"""Publish ONE ``nestquest_quest_missed`` event for ``instance_id``.
The payload is built by the core builder
(:func:`nestquest_core.events.build_quest_missed_event`), which
mirrors the documented sweep payload exactly (§3 of
design/ENTITIES-AND-SERVICES.md).
"""
event_type, payload = await core_events.build_quest_missed_event(
database, instance_id
)
publisher.publish(event_type, payload)
async def publish_child_day_complete(
database: object,
completed_payload: dict,
today: datetime.date,
publisher: object,
) -> bool:
"""Publish ``nestquest_child_day_complete`` when a completion
cleared the child's whole day; return whether one was published.
The day-complete rule lives entirely in the core builder
(:func:`nestquest_core.events.build_child_day_complete_event`):
it returns ``None`` when the day did not clear (a zero-quest day
or a non-today instance) and this helper publishes nothing then.
"""
event = await core_events.build_child_day_complete_event(
database, completed_payload, today=today
)
if event is None:
return False
event_type, payload = event
publisher.publish(event_type, payload)
return True
__all__ = [
"publish_child_day_complete",
"publish_quest_missed",
"publish_quest_uncompleted",
]
+21 -1
View File
@@ -14,7 +14,9 @@ fires nothing), and ``nestquest_child_day_complete`` fires when a
completion clears the child's whole day (quests were owed and none
remain — a zero-quest day can never fire it, matching the all-done
binary sensor's rule). ``nestquest_quest_missed`` is the documented
contract Feature 11's nightly sweep fires; nothing here builds it.
contract Feature 11's nightly sweep fires; :func:`build_quest_missed_event`
builds it with exactly the payload the sweep hand-builds (§3), so the
API plane can publish the same shape without re-deriving it.
The completed-event and day-complete-event builders are split so the
shim can fire ``nestquest_quest_completed`` BEFORE it evaluates the
@@ -50,6 +52,7 @@ from .completion import derive_state
from .const import (
EVENT_CHILD_DAY_COMPLETE,
EVENT_QUEST_COMPLETED,
EVENT_QUEST_MISSED,
EVENT_QUEST_UNCOMPLETED,
)
from .dao_children import ChildrenDao
@@ -188,3 +191,20 @@ async def build_quest_uncompleted_events(
"""
payload = await _instance_payload(database, instance_id, _now_stamp())
return [(EVENT_QUEST_UNCOMPLETED, payload)]
async def build_quest_missed_event(
database: NestQuestDatabase, instance_id: int
) -> tuple[str, dict]:
"""Build the ``nestquest_quest_missed`` event for one instance.
The payload mirrors the nightly sweep's hand-built payload exactly
(the shared :func:`_instance_payload` shape: ``child_id``,
``child_name``, ``instance_id``, ``quest_title``, ``window``,
``due_date``, ``due_time``, ``occurred_at``), so the API plane's
missed path publishes the same event shape the integration's sweep
fires. Like the sweep, this builder never mutates the instance or
writes a completion event — it reads only.
"""
payload = await _instance_payload(database, instance_id, _now_stamp())
return (EVENT_QUEST_MISSED, payload)
+165 -4
View File
@@ -46,6 +46,9 @@ from tests.test_api_panel import PANEL_TOKEN, _seed_household
#: The documented quest-completed event name (core const, §3).
_core_const = importlib.import_module("nestquest_core.const")
EVENT_QUEST_COMPLETED = _core_const.EVENT_QUEST_COMPLETED
EVENT_QUEST_UNCOMPLETED = _core_const.EVENT_QUEST_UNCOMPLETED
EVENT_QUEST_MISSED = _core_const.EVENT_QUEST_MISSED
EVENT_CHILD_DAY_COMPLETE = _core_const.EVENT_CHILD_DAY_COMPLETE
#: Upper bound for any single stream read; bounds the whole test so a
#: wedged stream fails the assertion instead of hanging the suite.
@@ -70,12 +73,15 @@ class _Server:
def __init__(self, db_path: str) -> None:
self.port = _free_port()
self._server_loop: asyncio.AbstractEventLoop | None = None
self._server = uvicorn.Server(
uvicorn.Config(
# A closure factory: reads nothing at import time.
lambda: create_app(
ApiConfig(db_path=db_path, panel_token=PANEL_TOKEN)
),
# A closure factory that captures the app's event loop
# as it is created INSIDE the server thread: the
# lifespan's database and publisher belong to that
# thread's loop, so tests driving an API-side
# transition helper directly schedule it there.
self._make_app_factory(db_path),
host="127.0.0.1",
port=self.port,
log_level="warning",
@@ -85,6 +91,23 @@ class _Server:
target=self._server.run, daemon=True
)
def _make_app_factory(self, db_path: str):
"""The uvicorn app factory, closing over ``db_path``.
Records the running loop the first time uvicorn calls it (in
the server thread) so :meth:`run_on_loop` can target it.
"""
def factory():
app = create_app(
ApiConfig(db_path=db_path, panel_token=PANEL_TOKEN)
)
self._server_loop = asyncio.get_running_loop()
self.app = app
return app
return factory
def __enter__(self) -> "_Server":
self._thread.start()
deadline = time.monotonic() + READ_TIMEOUT
@@ -102,6 +125,20 @@ class _Server:
"""The full URL for ``path`` on this server."""
return f"http://127.0.0.1:{self.port}{path}"
def run_on_loop(self, coro) -> None:
"""Run ``coro`` on the SERVER's event loop and wait for it.
The app (its database and publisher) lives on the uvicorn
thread's loop, so a test that drives an API-side transition
helper directly must schedule it there: the helper awaits core
DAO calls and calls ``publisher.publish`` synchronously, and
both must run on the loop that owns them. Bounded by
READ_TIMEOUT so a wedged schedule fails instead of hanging.
"""
return asyncio.run_coroutine_threadsafe(
coro, self._server_loop
).result(timeout=READ_TIMEOUT)
class _EventStream:
"""A subscribed SSE reader over a live httpx connection.
@@ -284,6 +321,10 @@ async def test_no_op_recomplete_publishes_nothing(seeded_panel) -> None:
assert first.status_code == 200
event_type, _ = await stream.next_event()
assert event_type == EVENT_QUEST_COMPLETED
# Ada's only quest today: the completion also clears her whole
# day, so the day-complete event follows (task 44c93cfd).
event_type, _ = await stream.next_event()
assert event_type == EVENT_CHILD_DAY_COMPLETE
second = await seeded_panel.client.post(
url, headers=headers, json=body
@@ -293,3 +334,123 @@ async def test_no_op_recomplete_publishes_nothing(seeded_panel) -> None:
# Nothing arrives within the window: the no-op re-complete
# must publish nothing.
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
# --- the remaining documented transitions (task 44c93cfd) -----------------
async def test_publish_quest_uncompleted_streams_event(seeded_panel) -> None:
"""The reusable uncompleted publisher streams the documented
uncompleted transition with its documented payload fields.
The admin uncomplete route does not exist yet (deferred to task
da0226b3); the test drives the API-side helper the route will call.
The helper runs on the SERVER's event loop (``run_coroutine_threadsafe``
on the uvicorn thread's loop) against the app's own live database,
exactly the way the later route handler will call it.
"""
from api import transitions
seed = seeded_panel.seed
pack_bag = seed.pack_bag_instance
app = seeded_panel.server.app
database = app.state.db.database
stream = _stream(seeded_panel.url("/api/v1/panel/events"))
async with stream:
seeded_panel.server.run_on_loop(
transitions.publish_quest_uncompleted(
database, pack_bag.id, app.state.publisher
)
)
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_UNCOMPLETED
assert event_type == "nestquest_quest_uncompleted"
assert payload["child_id"] == seed.ada.id
assert payload["child_name"] == "Ada"
assert payload["instance_id"] == pack_bag.id
assert payload["quest_title"] == "Pack bag"
assert payload["window"] == "morning"
assert payload["due_date"] == pack_bag.due_date
assert payload["due_time"] == "10:00"
assert payload["occurred_at"]
assert "was_on_time" not in payload
async def test_last_quest_of_day_publishes_day_complete(seeded_panel) -> None:
"""Completing the child's LAST open quest of the day emits the
completed event AND the child-day-complete event carrying
``quests_due`` and ``quests_completed``.
Ada has exactly one instance today (Pack bag), so completing it
clears her whole day; a no-op re-complete afterwards publishes
nothing.
"""
seed = seeded_panel.seed
pack_bag = seed.pack_bag_instance
stream = _stream(seeded_panel.url("/api/v1/panel/events"))
async with stream:
response = await seeded_panel.client.post(
f"/api/v1/panel/instances/{pack_bag.id}/complete",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
json={"actor_child_id": seed.ada.id},
)
assert response.status_code == 200
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_COMPLETED
completed_occurred_at = payload["occurred_at"]
event_type, payload = await stream.next_event()
assert event_type == EVENT_CHILD_DAY_COMPLETE
assert event_type == "nestquest_child_day_complete"
assert payload["child_id"] == seed.ada.id
assert payload["child_name"] == "Ada"
assert payload["quests_due"] == 1
assert payload["quests_completed"] == 1
# One transition, one shared timestamp (the core builder
# reuses the completed event's stamp).
assert payload["occurred_at"] == completed_occurred_at
# No-op re-complete: nothing further on the stream.
second = await seeded_panel.client.post(
f"/api/v1/panel/instances/{pack_bag.id}/complete",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
json={"actor_child_id": seed.ada.id},
)
assert second.status_code == 200
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
async def test_publish_quest_missed_streams_event(seeded_panel) -> None:
"""The reusable missed publisher streams the documented missed
transition with its documented payload fields (the sweep payload).
The API-side missed sweep does not exist yet at this queue
position; the test drives the API-side helper the sweep will call
against the seeded past-due open instance, on the server's own
event loop and database.
"""
from api import transitions
seed = seeded_panel.seed
stale = seed.stale_instance
app = seeded_panel.server.app
database = app.state.db.database
stream = _stream(seeded_panel.url("/api/v1/panel/events"))
async with stream:
seeded_panel.server.run_on_loop(
transitions.publish_quest_missed(
database, stale.id, app.state.publisher
)
)
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_MISSED
assert event_type == "nestquest_quest_missed"
assert payload["child_id"] == seed.ada.id
assert payload["child_name"] == "Ada"
assert payload["instance_id"] == stale.id
assert payload["quest_title"] == "Stale chore"
assert payload["window"] == "morning"
assert payload["due_date"] == stale.due_date
assert payload["due_time"] == "09:00"
assert payload["occurred_at"]
+28
View File
@@ -791,3 +791,31 @@ async def test_shim_day_complete_uses_ha_local_timezone(
f"{far_today.isoformat()}, UTC today {utc_today.isoformat()}); "
f"got {len(day_complete)} events"
)
async def test_build_quest_missed_event_payload_key_set(
hass, make_entry
) -> None:
"""The missed-event payload carries exactly the documented keys —
the same shape the nightly sweep hand-builds (§3), with no
``was_on_time`` (completion-only)."""
from custom_components.nestquest.core.events import (
build_quest_missed_event,
)
entry, child, instances = await _setup_and_seed(hass, make_entry)
database = entry.runtime_data.database
event_type, payload = await build_quest_missed_event(
database, instances[0].id
)
assert event_type == EVENT_QUEST_MISSED
assert event_type == "nestquest_quest_missed"
assert set(payload) == _UNCOMPLETED_KEYS, (
f"missed payload keys drift: got {set(payload)} "
f"expected {_UNCOMPLETED_KEYS}"
)
assert payload["child_id"] == child.id
assert payload["child_name"] == "Ada"
assert payload["instance_id"] == instances[0].id
assert payload["quest_title"] == "Brush teeth"
assert _TIMESTAMP.match(payload["occurred_at"]), payload["occurred_at"]