SSE: publish uncompleted, child-day-complete and missed #155
+14
-1
@@ -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)}
|
||||
|
||||
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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
@@ -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"]
|
||||
|
||||
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user