Maniacs_Mansion/NestQuest
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove the HA admin services and the service permission gate

Browse Source 
Since Feature 18 the integration is a client of the NestQuest API
service and the admin surface is the PWA plus the API's admin routes,
so the integration must not keep HA services for admin operations.
Drop every admin service handler and its registration (uncomplete,
create/update/set-active quest definition, presence pattern and
overrides, CSV export, manage_child, regenerate) together with the
Feature 09 permission gate that existed only for them: permissions.py,
service_policy.py and the core SERVICE_POLICY registry, the
regenerate handler plumbing in __init__, the dead payload schemas and
services.yaml entries, and the service-name constants they alone
used. Only the panel completion path (nestquest.complete_quest)
remains, still proxying to the entry's shared API client. The four
automation blueprints are untouched — they only read sensors and
listen to bus events.
This commit is contained in:
overseer
2026-09-23 08:21:22 -04:00
parent 684f52ece8
commit 403dd736b8
15 changed files with 157 additions and 2042 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/nestquest/__init__.py
+15 -39
View File
@@ -282,9 +282,10 @@ async def _run_horizon_materialization(
The walk itself runs through :func:`~.materialize.materialize`, whose DB
access is executor-only.
This is the ONE generation path: the startup backfill, the daily rollover
listener and the ``regenerate`` service all funnel through it, so there is
no duplicated materialization logic. ``settings`` (REQUIRED — there is no
This is the ONE generation path inside the integration: the
startup backfill and the daily rollover listener both funnel
through it, so there is no duplicated materialization logic.
``settings`` (REQUIRED — there is no
silent default; every caller already passes the entry's validated
:class:`NestQuestSettings`) carries the configured ``horizon_days``. The
window itself comes from :meth:`settings.horizon_window` so the helper is
@@ -326,13 +327,10 @@ def _register_day_rollover_listener(
def _find_live_runtime_data(hass: HomeAssistant) -> NestQuestRuntimeData | None:
"""Return a live runtime record with a connected database, or None.
The domain-global services (e.g. ``regenerate``) have no database of
their own; they resolve whichever config entry's runtime data is
currently live at call time, so none of them can hold a stale handle
to a closed connection after another entry unloads. The record also
carries that entry's validated ``settings`` (e.g. the configured
horizon) so service handlers read horizon days off it rather than
re-reading HA config.
The domain-global service (the panel completion proxy) has no
database of its own; it resolves whichever config entry's runtime
data is currently live at call time, so it cannot hold a stale
handle to a closed connection after another entry unloads.
"""
for runtime_data in hass.data.get(DOMAIN, {}).values():
database = getattr(runtime_data, "database", None)
@@ -341,39 +339,17 @@ def _find_live_runtime_data(hass: HomeAssistant) -> NestQuestRuntimeData | None:
return None
def _make_regenerate_handler(hass: HomeAssistant):
"""Return the Feature 07 ``regenerate`` handler closed over ``hass``."""
async def _regenerate(_call: Any) -> None:
runtime_data = _find_live_runtime_data(hass)
if runtime_data is None:
_LOGGER.warning(
"NestQuest regenerate requested but no config entry has a "
"live database; ignoring"
)
return
await _run_horizon_materialization(
hass, runtime_data.database, runtime_data.settings
)
return _regenerate
def _register_services(hass: HomeAssistant) -> None:
"""Register domain-global NestQuest services once.
"""Register the domain-global NestQuest service once.
Services are DOMAIN-global — registered exactly once (guarded by
``has_service`` on regenerate) and NOT bound to any config entry —
so each handler resolves the currently-live database at call time
The service is DOMAIN-global — registered exactly once (guarded by
``has_service``) and NOT bound to any config entry — so the panel
completion proxy resolves the currently-live runtime at call time
and a name collision or duplicate setup is impossible. Removal is
handled by the unload path only once the LAST entry is removed
(see :func:`_deregister_services`).
"""
async_register_services(
hass,
find_runtime=_find_live_runtime_data,
regenerate_handler=_make_regenerate_handler(hass),
)
async_register_services(hass, find_runtime=_find_live_runtime_data)
def _deregister_services(hass: HomeAssistant) -> None:
@@ -483,8 +459,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Backfill any days the daily listener missed while the integration
# was off (or freshly installed): run the one idempotent walk over
# [today, today + horizon_days] once at setup, using the entry's
# configured horizon. This runs through the SAME path as the daily
# listener and regenerate service.
# configured horizon. This runs through the SAME path as the
# daily listener.
await _run_horizon_materialization(hass, database, settings)
# The missed sweep runs at startup too (Feature 11): if HA was
# down at the rollover time, the previous night's still-open
custom_components/nestquest/core/const.py
+10 -25
View File
@@ -20,40 +20,25 @@ TIME_PATTERN = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d")
PLATFORMS: list[str] = ["sensor", "binary_sensor"]
# Home Assistant bus events (design/ENTITIES-AND-SERVICES.md §3). The
# panel and admin cards listen for these; Feature 10 fires completed,
# uncompleted, and child_day_complete from the service paths, and
# Feature 11's nightly sweep fires quest_missed.
# panel and admin cards listen for these; since Feature 18 the
# completed/uncompleted/day-complete transitions are published by the
# API service on its SSE stream and re-fired on the bus by the
# integration's subscription, and Feature 11's nightly sweep fires
# quest_missed.
EVENT_QUEST_COMPLETED = "nestquest_quest_completed"
EVENT_QUEST_UNCOMPLETED = "nestquest_quest_uncompleted"
EVENT_QUEST_MISSED = "nestquest_quest_missed"
EVENT_CHILD_DAY_COMPLETE = "nestquest_child_day_complete"
#: Canonical domain-global service names (design/ENTITIES-AND-SERVICES.md §2
#: plus the existing Feature 07 regenerate service).
#: Canonical domain-global service name (design/ENTITIES-AND-SERVICES.md §2).
#: Since Feature 18 only the panel completion service remains: the admin
#: surface is the PWA and the API service's admin routes, so the former
#: HA admin services (uncomplete, quest-definition and presence
#: management, CSV export, child management, regenerate) are gone.
SERVICE_COMPLETE_QUEST = "complete_quest"
SERVICE_UNCOMPLETE_QUEST = "uncomplete_quest"
SERVICE_CREATE_QUEST_DEFINITION = "create_quest_definition"
SERVICE_UPDATE_QUEST_DEFINITION = "update_quest_definition"
SERVICE_SET_QUEST_DEFINITION_ACTIVE = "set_quest_definition_active"
SERVICE_SET_PRESENCE_PATTERN = "set_presence_pattern"
SERVICE_CREATE_PRESENCE_OVERRIDE = "create_presence_override"
SERVICE_DELETE_PRESENCE_OVERRIDE = "delete_presence_override"
SERVICE_EXPORT_HISTORY_CSV = "export_history_csv"
SERVICE_MANAGE_CHILD = "manage_child"
SERVICE_REGENERATE = "regenerate"
DOMAIN_SERVICES: tuple[str, ...] = (
SERVICE_COMPLETE_QUEST,
SERVICE_UNCOMPLETE_QUEST,
SERVICE_CREATE_QUEST_DEFINITION,
SERVICE_UPDATE_QUEST_DEFINITION,
SERVICE_SET_QUEST_DEFINITION_ACTIVE,
SERVICE_SET_PRESENCE_PATTERN,
SERVICE_CREATE_PRESENCE_OVERRIDE,
SERVICE_DELETE_PRESENCE_OVERRIDE,
SERVICE_EXPORT_HISTORY_CSV,
SERVICE_MANAGE_CHILD,
SERVICE_REGENERATE,
)
SQLITE_DB_FILENAME = "nestquest.db"
custom_components/nestquest/core/service_policy.py
-46
View File
@@ -1,46 +0,0 @@
"""Explicit admin-only versus open service registry (Feature 09).
One mapping, one file: every registered ``nestquest`` service name is
either ``open`` or ``admin_only``. The permission decorator (later
Feature 09 tasks) reads this table; it does not take per-handler
policy arguments.
"""
from __future__ import annotations
from .const import (
DOMAIN_SERVICES,
SERVICE_COMPLETE_QUEST,
SERVICE_CREATE_PRESENCE_OVERRIDE,
SERVICE_CREATE_QUEST_DEFINITION,
SERVICE_DELETE_PRESENCE_OVERRIDE,
SERVICE_EXPORT_HISTORY_CSV,
SERVICE_MANAGE_CHILD,
SERVICE_REGENERATE,
SERVICE_SET_PRESENCE_PATTERN,
SERVICE_SET_QUEST_DEFINITION_ACTIVE,
SERVICE_UNCOMPLETE_QUEST,
SERVICE_UPDATE_QUEST_DEFINITION,
)
OPEN = "open"
ADMIN_ONLY = "admin_only"
SERVICE_POLICY: dict[str, str] = {
SERVICE_COMPLETE_QUEST: OPEN,
SERVICE_UNCOMPLETE_QUEST: ADMIN_ONLY,
SERVICE_CREATE_QUEST_DEFINITION: ADMIN_ONLY,
SERVICE_UPDATE_QUEST_DEFINITION: ADMIN_ONLY,
SERVICE_SET_QUEST_DEFINITION_ACTIVE: ADMIN_ONLY,
SERVICE_SET_PRESENCE_PATTERN: ADMIN_ONLY,
SERVICE_CREATE_PRESENCE_OVERRIDE: ADMIN_ONLY,
SERVICE_DELETE_PRESENCE_OVERRIDE: ADMIN_ONLY,
SERVICE_EXPORT_HISTORY_CSV: ADMIN_ONLY,
SERVICE_MANAGE_CHILD: ADMIN_ONLY,
SERVICE_REGENERATE: ADMIN_ONLY,
}
if frozenset(SERVICE_POLICY) != frozenset(DOMAIN_SERVICES):
raise RuntimeError(
"SERVICE_POLICY must name exactly the registered nestquest services, "
f"got {sorted(SERVICE_POLICY)} vs {list(DOMAIN_SERVICES)}"
)
custom_components/nestquest/events.py
+4 -2
View File
@@ -4,8 +4,10 @@ The transition-event payload construction lives in
:mod:`custom_components.nestquest.core.events` (extracted in
task 510c1f78) and is HA-free: it builds and RETURNS the events. This
shim keeps the integration's old ``fire_quest_completed`` /
``fire_quest_uncompleted`` API (the complete/uncomplete service
handlers call them with ``hass``), reads the HA-local timezone from
``fire_quest_uncompleted`` API for direct callers — since Feature 18
the HA service surface no longer fires transition events itself (the
API service publishes them on its SSE stream and the integration's
subscription re-fires them) — reads the HA-local timezone from
``hass.config.time_zone`` to derive ``today``, asks the core builder
for the events, and fires each on ``hass.bus`` with the documented
payloads. The four blueprints keep working with identical payloads.
custom_components/nestquest/permissions.py
-90
View File
@@ -1,90 +0,0 @@
"""Permission gate for NestQuest services (Feature 09).
A wrapper factory, applied to EVERY registered service at registration
time (no handler is exempt), reads ``call.context.user_id``, consults
the :data:`~.service_policy.SERVICE_POLICY` registry and
:func:`~.admin_allowlist.is_admin`, and raises
:class:`homeassistant.exceptions.Unauthorized` for an admin-only
operation called by a non-admin — including a call with NO user
context, which fails closed. ``complete_quest`` (policy ``open``)
proceeds regardless of the allowlist.
Logging (later Feature 09 tasks): every denied admin-only call logs a
warning and every successful one logs a debug line, both carrying the
service, the user id (or ``none``), and a UTC timestamp.
"""
from __future__ import annotations
import datetime
import logging
from collections.abc import Awaitable, Callable
from typing import Any
from homeassistant.exceptions import Unauthorized
from .admin_allowlist import is_admin
from .service_policy import ADMIN_ONLY
LOGGER = logging.getLogger(__name__)
ServiceHandler = Callable[[Any], Awaitable[None]]
def _context_user_id(call: Any) -> str | None:
context = getattr(call, "context", None)
user_id = getattr(context, "user_id", None) if context is not None else None
if isinstance(user_id, str) and user_id:
return user_id
return None
def _stamp() -> str:
return datetime.datetime.now(datetime.timezone.utc).isoformat(
timespec="seconds"
)
def permission_gate(
service: str,
policy: str,
database_of: Callable[[Any], Any],
) -> Callable[[ServiceHandler], ServiceHandler]:
"""Wrap ``handler`` with the Feature 09 permission check.
``database_of`` resolves the live database from the call-time
runtime (the same ``find_runtime`` the handler uses), so the gate
never opens its own connection. The verdict is fail-closed:
absent allowlist row, unknown id, no user context — all not admin.
"""
def _wrap(handler: ServiceHandler) -> ServiceHandler:
if policy != ADMIN_ONLY:
return handler
async def _gated(call: Any) -> None:
user_id = _context_user_id(call)
allowed = user_id is not None and await is_admin(
database_of(call), user_id
)
if not allowed:
LOGGER.warning(
"NestQuest denied %s: caller user_id=%s at %s",
service,
user_id if user_id is not None else "none",
_stamp(),
)
raise Unauthorized(
f"User '{user_id if user_id is not None else 'none'}' "
f"is not allowed to call nestquest.{service}"
)
LOGGER.debug(
"NestQuest allowed %s: caller user_id=%s at %s",
service,
user_id,
_stamp(),
)
await handler(call)
return _gated
return _wrap
custom_components/nestquest/service_policy.py
-11
View File
@@ -1,11 +0,0 @@
"""Re-export of :mod:`custom_components.nestquest.core.service_policy` for the integration package.
The service policy registry lives in
:mod:`custom_components.nestquest.core.service_policy` (extracted in
task 510c1f78); this module re-exports it so existing
``from .service_policy import ...`` references inside the integration
keep working.
"""
from __future__ import annotations
from .core.service_policy import * # noqa: F401,F403
custom_components/nestquest/services.py
+24 -475
View File
@@ -1,65 +1,36 @@
"""Domain-global NestQuest service schemas and handlers (Feature 09).
"""The domain-global ``nestquest.complete_quest`` service (Feature 18).
Registration is domain-scoped and happens once (see
:func:`async_register_services`); handlers resolve a live database at
call time through the injected ``find_runtime`` callable. Permission
checks are Feature 09 later tasks — this module only validates payload
shape and forwards to the Feature 03/06/08 business layers (the
presence writes included, through :mod:`.presence_management` — the
ONE implementation the API admin plane shares). Since Feature 18 the
integration is a client of the NestQuest API service, so
``complete_quest`` proxies to the API's panel complete route through
the entry's shared API client instead of writing any database.
``export_history_csv`` is registered with a schema but raises until
Feature 13. ``regenerate`` is supplied by the caller so
the Feature 07 handler stays the one in ``__init__``.
:func:`async_register_services`). Since Feature 18 the integration is
a client of the NestQuest API service, so ``complete_quest`` proxies
to the API's panel complete route through the entry's shared API
client instead of writing any database. The former HA admin services
(uncomplete, quest-definition and presence management, CSV export,
child management, regenerate) are removed: the admin surface is the
PWA and the API service's admin routes, so the only HA service left is
the panel completion path.
"""
from __future__ import annotations
import datetime
from collections.abc import Awaitable, Callable
from typing import Any
from zoneinfo import ZoneInfo
import voluptuous as vol
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from . import children as children_layer
from . import completion as completion_layer
from . import presence_management as presence_management_layer
from . import quest_definitions as quest_definitions_layer
from .api_client import NestQuestApiError
from .const import (
DOMAIN,
DOMAIN_SERVICES,
QUEST_WINDOWS,
SERVICE_COMPLETE_QUEST,
SERVICE_CREATE_PRESENCE_OVERRIDE,
SERVICE_CREATE_QUEST_DEFINITION,
SERVICE_DELETE_PRESENCE_OVERRIDE,
SERVICE_EXPORT_HISTORY_CSV,
SERVICE_MANAGE_CHILD,
SERVICE_REGENERATE,
SERVICE_SET_PRESENCE_PATTERN,
SERVICE_SET_QUEST_DEFINITION_ACTIVE,
SERVICE_UNCOMPLETE_QUEST,
SERVICE_UPDATE_QUEST_DEFINITION,
)
from .events import fire_quest_uncompleted
from .permissions import permission_gate
from .recurrence import ScheduleRule
from .service_policy import SERVICE_POLICY
FindRuntime = Callable[[HomeAssistant], Any]
ServiceHandler = Callable[[Any], Awaitable[None]]
ACTOR_PANEL = "panel"
ACTOR_USER = "user"
MANAGE_CHILD_CREATE = "create"
MANAGE_CHILD_EDIT = "edit"
MANAGE_CHILD_SET_ACTIVE = "set_active"
MANAGE_CHILD_REORDER = "reorder"
def _strict_int(value: object) -> int:
@@ -69,82 +40,6 @@ def _strict_int(value: object) -> int:
return value
def _strict_bool(value: object) -> bool:
"""Reject ints and other truthy values; require a real bool."""
if type(value) is not bool:
raise vol.Invalid(f"expected bool, got {value!r}")
return value
def _weekday_list(value: object) -> list[int]:
if type(value) is not list:
raise vol.Invalid(f"expected a list of weekday ints, got {value!r}")
return [_strict_int(item) for item in value]
def _presence_pattern(value: object) -> dict[int, list[int]]:
if not isinstance(value, dict):
raise vol.Invalid(f"pattern must be a mapping, got {value!r}")
pattern: dict[int, list[int]] = {}
for week, days in value.items():
if isinstance(week, str):
try:
week_index = int(week)
except ValueError as err:
raise vol.Invalid(
f"pattern week index must be an integer, got {week!r}"
) from err
if str(week_index) != week:
raise vol.Invalid(
f"pattern week index must be an integer, got {week!r}"
)
else:
week_index = _strict_int(week)
pattern[week_index] = _weekday_list(days)
return pattern
def _window_entry(value: object) -> str | tuple[str, str | None]:
if isinstance(value, str):
if value not in QUEST_WINDOWS:
raise vol.Invalid(f"unknown window {value!r}")
return value
if isinstance(value, dict):
parsed = vol.Schema(
{
vol.Required("window"): vol.In(QUEST_WINDOWS),
vol.Optional("due_time"): str,
}
)(value)
due_time = parsed.get("due_time")
if due_time is None:
return parsed["window"]
return (parsed["window"], due_time)
raise vol.Invalid(
f"windows entry must be a window name or object, got {value!r}"
)
def _windows(value: object) -> list[str | tuple[str, str | None]]:
if type(value) is not list:
raise vol.Invalid(f"windows must be a list, got {value!r}")
if not value:
raise vol.Invalid("windows must not be empty")
return [_window_entry(entry) for entry in value]
def _int_list(value: object) -> list[int]:
if type(value) is not list:
raise vol.Invalid(f"expected a list of ints, got {value!r}")
return [_strict_int(item) for item in value]
def _rule_dict(value: object) -> dict:
if not isinstance(value, dict):
raise vol.Invalid(f"rule must be a dict, got {value!r}")
return value
SCHEMA_COMPLETE_QUEST = vol.Schema(
{
vol.Required("instance_id"): _strict_int,
@@ -154,86 +49,6 @@ SCHEMA_COMPLETE_QUEST = vol.Schema(
}
)
SCHEMA_UNCOMPLETE_QUEST = SCHEMA_COMPLETE_QUEST
SCHEMA_CREATE_QUEST_DEFINITION = vol.Schema(
{
vol.Required("title"): str,
vol.Required("rule"): _rule_dict,
vol.Required("assignee_child_ids"): vol.All(
_int_list, vol.Length(min=1)
),
vol.Required("windows"): _windows,
vol.Optional("description"): str,
vol.Optional("icon"): str,
}
)
SCHEMA_UPDATE_QUEST_DEFINITION = vol.Schema(
{
vol.Required("definition_id"): _strict_int,
vol.Optional("title"): str,
vol.Optional("description"): vol.Any(str, None),
vol.Optional("icon"): vol.Any(str, None),
vol.Optional("rule"): _rule_dict,
vol.Optional("windows"): _windows,
}
)
SCHEMA_SET_QUEST_DEFINITION_ACTIVE = vol.Schema(
{
vol.Required("definition_id"): _strict_int,
vol.Required("is_active"): _strict_bool,
}
)
SCHEMA_SET_PRESENCE_PATTERN = vol.Schema(
{
vol.Required("child_id"): _strict_int,
vol.Required("cycle_length_weeks"): _strict_int,
vol.Required("anchor_date"): str,
vol.Required("pattern"): _presence_pattern,
}
)
SCHEMA_CREATE_PRESENCE_OVERRIDE = vol.Schema(
{
vol.Required("child_id"): _strict_int,
vol.Required("start_date"): str,
vol.Required("end_date"): str,
vol.Required("is_present"): _strict_bool,
vol.Optional("note"): str,
}
)
SCHEMA_DELETE_PRESENCE_OVERRIDE = vol.Schema(
{
vol.Required("override_id"): _strict_int,
}
)
SCHEMA_EXPORT_HISTORY_CSV = vol.Schema({})
SCHEMA_MANAGE_CHILD = vol.Schema(
{
vol.Required("action"): vol.In(
[
MANAGE_CHILD_CREATE,
MANAGE_CHILD_EDIT,
MANAGE_CHILD_SET_ACTIVE,
MANAGE_CHILD_REORDER,
]
),
vol.Optional("child_id"): _strict_int,
vol.Optional("display_name"): str,
vol.Optional("colour"): str,
vol.Optional("avatar_ref"): str,
vol.Optional("sort_order"): _strict_int,
vol.Optional("is_active"): _strict_bool,
vol.Optional("ordered_ids"): _int_list,
}
)
def _require_runtime(hass: HomeAssistant, find_runtime: FindRuntime):
runtime = find_runtime(hass)
@@ -245,11 +60,11 @@ def _require_runtime(hass: HomeAssistant, find_runtime: FindRuntime):
async def _refresh_entities(runtime: Any) -> None:
"""Force an immediate coordinator refresh after a service mutation.
"""Force an immediate coordinator refresh after the service call.
Only the SERVICE paths call this — the scheduled materialize and
Only the SERVICE path calls this — the scheduled materialize and
rollover internals are direct function calls and never widen this
gate — so entities reflect a completion the moment the service
gate — so the entities reflect a completion the moment the service
call returns instead of waiting for the poll interval.
"""
coordinator = getattr(runtime, "coordinator", None)
@@ -257,12 +72,6 @@ async def _refresh_entities(runtime: Any) -> None:
await coordinator.async_refresh()
def _generation_kwargs(hass: HomeAssistant, runtime: Any) -> dict[str, Any]:
time_zone = ZoneInfo(hass.config.time_zone)
today = datetime.datetime.now(time_zone).date()
return {"today": today, "horizon_days": runtime.settings.horizon_days}
def _actor_kwargs(call: Any) -> dict[str, Any]:
actor = call.data["actor"]
if actor == ACTOR_PANEL:
@@ -353,289 +162,29 @@ def _complete_quest(
return _with_errors(_handler)
def _uncomplete_quest(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
runtime = _require_runtime(hass, find_runtime)
instance_id = call.data["instance_id"]
result = await completion_layer.uncomplete_instance(
runtime.database,
instance_id,
**_actor_kwargs(call),
)
if result.appended:
await fire_quest_uncompleted(
hass, runtime.database, instance_id
)
await _refresh_entities(runtime)
return _with_errors(_handler)
def _create_quest_definition(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
database = _require_runtime(hass, find_runtime).database
data = call.data
await quest_definitions_layer.create_quest_definition(
database,
data["title"],
ScheduleRule.from_dict(data["rule"]),
data["assignee_child_ids"],
data["windows"],
description=data.get("description"),
icon=data.get("icon"),
)
return _with_errors(_handler)
def _update_quest_definition(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
runtime = _require_runtime(hass, find_runtime)
data = call.data
kwargs: dict[str, Any] = _generation_kwargs(hass, runtime)
if "title" in data:
kwargs["title"] = data["title"]
if "description" in data:
kwargs["description"] = data["description"]
if "icon" in data:
kwargs["icon"] = data["icon"]
if "rule" in data:
kwargs["rule"] = ScheduleRule.from_dict(data["rule"])
if "windows" in data:
kwargs["windows"] = data["windows"]
await quest_definitions_layer.edit_quest_definition(
runtime.database, data["definition_id"], **kwargs
)
return _with_errors(_handler)
def _set_quest_definition_active(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
runtime = _require_runtime(hass, find_runtime)
await quest_definitions_layer.set_quest_definition_active(
runtime.database,
call.data["definition_id"],
call.data["is_active"],
**_generation_kwargs(hass, runtime),
)
return _with_errors(_handler)
def _set_presence_pattern(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
runtime = _require_runtime(hass, find_runtime)
data = call.data
await presence_management_layer.set_presence_schedule(
runtime.database,
data["child_id"],
data["cycle_length_weeks"],
data["anchor_date"],
data["pattern"],
**_generation_kwargs(hass, runtime),
)
return _with_errors(_handler)
def _create_presence_override(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
runtime = _require_runtime(hass, find_runtime)
data = call.data
await presence_management_layer.create_presence_override(
runtime.database,
data["child_id"],
data["start_date"],
data["end_date"],
data["is_present"],
note=data.get("note"),
**_generation_kwargs(hass, runtime),
)
return _with_errors(_handler)
def _delete_presence_override(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
runtime = _require_runtime(hass, find_runtime)
await presence_management_layer.delete_presence_override(
runtime.database,
call.data["override_id"],
**_generation_kwargs(hass, runtime),
)
return _with_errors(_handler)
async def _export_history_csv(_call: Any) -> None:
raise HomeAssistantError(
"CSV history export is not implemented until Feature 13"
)
def _manage_child(
hass: HomeAssistant, find_runtime: FindRuntime
) -> ServiceHandler:
async def _handler(call: Any) -> None:
database = _require_runtime(hass, find_runtime).database
data = call.data
action = data["action"]
if action == MANAGE_CHILD_CREATE:
display_name = data.get("display_name")
if display_name is None:
raise HomeAssistantError(
"display_name is required when action is create"
)
kwargs: dict[str, Any] = {}
if "colour" in data:
kwargs["colour"] = data["colour"]
if "avatar_ref" in data:
kwargs["avatar_ref"] = data["avatar_ref"]
if "sort_order" in data:
kwargs["sort_order"] = data["sort_order"]
await children_layer.create_child(
database, display_name, **kwargs
)
return
if action == MANAGE_CHILD_EDIT:
child_id = data.get("child_id")
if child_id is None:
raise HomeAssistantError(
"child_id is required when action is edit"
)
kwargs = {}
if "display_name" in data:
kwargs["display_name"] = data["display_name"]
if "colour" in data:
kwargs["colour"] = data["colour"]
if "avatar_ref" in data:
kwargs["avatar_ref"] = data["avatar_ref"]
if "sort_order" in data:
kwargs["sort_order"] = data["sort_order"]
await children_layer.edit_child(database, child_id, **kwargs)
return
if action == MANAGE_CHILD_SET_ACTIVE:
child_id = data.get("child_id")
is_active = data.get("is_active")
if child_id is None:
raise HomeAssistantError(
"child_id is required when action is set_active"
)
if is_active is None:
raise HomeAssistantError(
"is_active is required when action is set_active"
)
await children_layer.set_child_active(
database, child_id, is_active
)
return
ordered_ids = data.get("ordered_ids")
if ordered_ids is None:
raise HomeAssistantError(
"ordered_ids is required when action is reorder"
)
await children_layer.reorder_children(database, ordered_ids)
return _with_errors(_handler)
def async_register_services(
hass: HomeAssistant,
*,
find_runtime: FindRuntime,
regenerate_handler: ServiceHandler,
) -> None:
"""Register every canonical service once at domain scope.
"""Register the panel completion service once at domain scope.
Every service — regenerate included — passes through the Feature
09 permission gate before its handler runs; the gate reads
:data:`~.service_policy.SERVICE_POLICY`, so no service can be
registered exempt from the policy.
The one remaining HA service is the panel completion path; every
admin operation moved to the API service's admin routes, so there
is no permission gate to apply here.
"""
if hass.services.has_service(DOMAIN, SERVICE_REGENERATE):
if hass.services.has_service(DOMAIN, SERVICE_COMPLETE_QUEST):
return
gate = permission_gate
specs: list[tuple[str, ServiceHandler, object]] = [
(SERVICE_REGENERATE, regenerate_handler, None),
(
SERVICE_COMPLETE_QUEST,
_complete_quest(hass, find_runtime),
SCHEMA_COMPLETE_QUEST,
),
(
SERVICE_UNCOMPLETE_QUEST,
_uncomplete_quest(hass, find_runtime),
SCHEMA_UNCOMPLETE_QUEST,
),
(
SERVICE_CREATE_QUEST_DEFINITION,
_create_quest_definition(hass, find_runtime),
SCHEMA_CREATE_QUEST_DEFINITION,
),
(
SERVICE_UPDATE_QUEST_DEFINITION,
_update_quest_definition(hass, find_runtime),
SCHEMA_UPDATE_QUEST_DEFINITION,
),
(
SERVICE_SET_QUEST_DEFINITION_ACTIVE,
_set_quest_definition_active(hass, find_runtime),
SCHEMA_SET_QUEST_DEFINITION_ACTIVE,
),
(
SERVICE_SET_PRESENCE_PATTERN,
_set_presence_pattern(hass, find_runtime),
SCHEMA_SET_PRESENCE_PATTERN,
),
(
SERVICE_CREATE_PRESENCE_OVERRIDE,
_create_presence_override(hass, find_runtime),
SCHEMA_CREATE_PRESENCE_OVERRIDE,
),
(
SERVICE_DELETE_PRESENCE_OVERRIDE,
_delete_presence_override(hass, find_runtime),
SCHEMA_DELETE_PRESENCE_OVERRIDE,
),
(
SERVICE_EXPORT_HISTORY_CSV,
_export_history_csv,
SCHEMA_EXPORT_HISTORY_CSV,
),
(
SERVICE_MANAGE_CHILD,
_manage_child(hass, find_runtime),
SCHEMA_MANAGE_CHILD,
),
]
def _database_of(_call: Any) -> Any:
return _require_runtime(hass, find_runtime).database
for service, handler, schema in specs:
gated = gate(
service, SERVICE_POLICY[service], _database_of
)(handler)
hass.services.async_register(DOMAIN, service, gated, schema=schema)
hass.services.async_register(
DOMAIN,
SERVICE_COMPLETE_QUEST,
_complete_quest(hass, find_runtime),
schema=SCHEMA_COMPLETE_QUEST,
)
def async_deregister_services(hass: HomeAssistant) -> None:
"""Remove every canonical service when the last entry unloads."""
for service in DOMAIN_SERVICES:
if hass.services.has_service(DOMAIN, service):
hass.services.async_remove(DOMAIN, service)
hass.services.async_remove(DOMAIN, service)
custom_components/nestquest/services.yaml
+4 -291
View File
@@ -1,15 +1,8 @@
# Service declarations for the NestQuest integration (Home Assistant
# services.yaml metadata). Services are domain-scoped: nested under the
# "nestquest" domain, not repeated per config entry.
regenerate:
name: Regenerate quest instances
description: >-
Regenerate every future quest instance for the household over the rolling
horizon, using the same idempotent materialization as the daily rollover
and the startup backfill. Safe to call at any time: a re-run refreshes the
same rows instead of duplicating them, and completed instances are never
touched.
# "nestquest" domain, not repeated per config entry. Since Feature 18
# only the panel completion service remains — every admin operation
# moved to the PWA and the API service's admin routes.
complete_quest:
name: Complete quest
@@ -43,284 +36,4 @@ complete_quest:
description: Home Assistant user id when actor is user and no call context is present.
required: false
selector:
text:
uncomplete_quest:
name: Uncomplete quest
description: Undo a quest instance completion.
fields:
instance_id:
name: Instance ID
description: The quest instance to uncomplete.
required: true
selector:
number:
mode: box
actor:
name: Actor
description: Who is undoing the completion (panel or user).
required: true
selector:
select:
options:
- panel
- user
actor_child_id:
name: Actor child ID
description: The tapped child profile. Required when actor is panel.
required: false
selector:
number:
mode: box
actor_user_id:
name: Actor user ID
description: Home Assistant user id when actor is user and no call context is present.
required: false
selector:
text:
create_quest_definition:
name: Create quest definition
description: Create a quest definition with its schedule rule, assignees and windows.
fields:
title:
name: Title
description: Display title for the quest.
required: true
selector:
text:
rule:
name: Rule
description: Recurrence rule as a mapping (rule_type plus shape fields).
required: true
selector:
object:
assignee_child_ids:
name: Assignee child IDs
description: Child ids this definition is assigned to.
required: true
selector:
object:
windows:
name: Windows
description: Day windows (morning, afternoon, evening), optionally with a due time.
required: true
selector:
object:
description:
name: Description
description: Optional longer description.
required: false
selector:
text:
icon:
name: Icon
description: Optional icon name.
required: false
selector:
text:
update_quest_definition:
name: Update quest definition
description: Edit a quest definition's title, description, icon, rule or windows.
fields:
definition_id:
name: Definition ID
description: The quest definition to update.
required: true
selector:
number:
mode: box
title:
name: Title
description: Replacement display title.
required: false
selector:
text:
description:
name: Description
description: Replacement description, or null to clear.
required: false
selector:
text:
icon:
name: Icon
description: Replacement icon, or null to clear.
required: false
selector:
text:
rule:
name: Rule
description: Replacement recurrence rule as a mapping.
required: false
selector:
object:
windows:
name: Windows
description: Replacement day windows.
required: false
selector:
object:
set_quest_definition_active:
name: Set quest definition active
description: Deactivate or reactivate a quest definition.
fields:
definition_id:
name: Definition ID
description: The quest definition to update.
required: true
selector:
number:
mode: box
is_active:
name: Active
description: Whether the definition should generate future instances.
required: true
selector:
boolean:
set_presence_pattern:
name: Set presence pattern
description: Upsert a child's repeating presence schedule.
fields:
child_id:
name: Child ID
description: The child whose schedule is being set.
required: true
selector:
number:
mode: box
cycle_length_weeks:
name: Cycle length weeks
description: Number of weeks in the repeating cycle (1 to 4).
required: true
selector:
number:
mode: box
min: 1
max: 4
anchor_date:
name: Anchor date
description: Cycle anchor as a YYYY-MM-DD date.
required: true
selector:
date:
pattern:
name: Pattern
description: Mapping of week index to present weekday ints (Monday is 0).
required: true
selector:
object:
create_presence_override:
name: Create presence override
description: Create a date-range presence override for a child.
fields:
child_id:
name: Child ID
description: The child the override applies to.
required: true
selector:
number:
mode: box
start_date:
name: Start date
description: Inclusive start date (YYYY-MM-DD).
required: true
selector:
date:
end_date:
name: End date
description: Inclusive end date (YYYY-MM-DD).
required: true
selector:
date:
is_present:
name: Present
description: Whether the child is present for this range.
required: true
selector:
boolean:
note:
name: Note
description: Optional note stored with the override.
required: false
selector:
text:
delete_presence_override:
name: Delete presence override
description: Delete a presence override by id.
fields:
override_id:
name: Override ID
description: The override to delete.
required: true
selector:
number:
mode: box
export_history_csv:
name: Export history CSV
description: Export completion history as CSV. Not implemented until Feature 13.
manage_child:
name: Manage child
description: Create, edit, activate or reorder child profiles.
fields:
action:
name: Action
description: create, edit, set_active, or reorder.
required: true
selector:
select:
options:
- create
- edit
- set_active
- reorder
child_id:
name: Child ID
description: Required for edit and set_active.
required: false
selector:
number:
mode: box
display_name:
name: Display name
description: Required for create; optional for edit.
required: false
selector:
text:
colour:
name: Colour
description: Optional colour string.
required: false
selector:
text:
avatar_ref:
name: Avatar
description: Optional avatar reference.
required: false
selector:
text:
sort_order:
name: Sort order
description: Optional panel sort order.
required: false
selector:
number:
mode: box
is_active:
name: Active
description: Required for set_active.
required: false
selector:
boolean:
ordered_ids:
name: Ordered IDs
description: Complete permutation of child ids. Required for reorder.
required: false
selector:
object:
text:
tests/test_entity_matrix.py
+25 -8
View File
@@ -22,7 +22,7 @@ from custom_components.nestquest.children import (
list_children,
set_child_active,
)
from custom_components.nestquest.completion import derive_state
from custom_components.nestquest.completion import derive_state, uncomplete_instance
from custom_components.nestquest.const import (
CONF_ADMIN_USER_IDS,
DOMAIN,
@@ -30,7 +30,9 @@ from custom_components.nestquest.const import (
EVENT_QUEST_COMPLETED,
EVENT_QUEST_UNCOMPLETED,
SERVICE_COMPLETE_QUEST,
SERVICE_UNCOMPLETE_QUEST,
)
from custom_components.nestquest.core.events import (
build_quest_uncompleted_events,
)
from custom_components.nestquest.dao_instances import QuestInstancesDao
from custom_components.nestquest.materialize import materialize
@@ -174,14 +176,29 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
)
# --- Reversing one quest fires the uncompleted event exactly once
# and restores the owed counts.
await hass.services.call(
DOMAIN,
SERVICE_UNCOMPLETE_QUEST,
{"instance_id": instances[0].id, "actor": "user"},
context={"user_id": "admin-1"},
# and restores the owed counts. The HA ``uncomplete_quest``
# service is gone (the API service's admin route is the reversal
# path): drive the core completion layer it calls, publish the
# uncompleted transition the way the API does, and deliver it
# through the SSE subscription.
reversal = await uncomplete_instance(
entry.runtime_data.database,
instances[0].id,
actor_source="user",
actor_user_id="admin-1",
)
assert reversal.appended is True
client = coordinator.api_client
client.published_frames.extend(
await build_quest_uncompleted_events(
entry.runtime_data.database, instances[0].id
)
)
await refire_api_transitions(hass, entry, client)
assert len(hass.bus.fired(EVENT_QUEST_UNCOMPLETED)) == 1
# The reversal no longer carries an immediate local refresh: the
# coordinator's next poll picks the restored counts up.
await coordinator.async_refresh()
assert _sensor(hass, ada.id, "all_done").is_on is False
assert _sensor(hass, ada.id, "quests_remaining_today").native_value == 1
tests/test_events.py
+33 -60
View File
@@ -9,7 +9,11 @@ re-fires them. The service-driven tests below therefore deliver the
frames through :func:`conftest.refire_api_transitions` (the
production event path, driven synchronously) before asserting the
bus; the shim-level tests drive :mod:`custom_components.nestquest
.events` directly, the way its remaining callers do.
.events` directly, the way its remaining callers do. The HA
``uncomplete_quest`` service is also gone (the API service's admin
uncomplete route is the reversal path), so the reversal tests drive
the core completion layer the route calls and deliver the published
frame through the same SSE subscription helper.
"""
from __future__ import annotations
@@ -19,6 +23,7 @@ from conftest import refire_api_transitions, wire_entry_to_registry
from custom_components.nestquest import async_setup_entry
from custom_components.nestquest.children import create_child
from custom_components.nestquest.completion import uncomplete_instance
from custom_components.nestquest.const import (
CONF_ADMIN_USER_IDS,
DOMAIN,
@@ -27,10 +32,9 @@ from custom_components.nestquest.const import (
EVENT_QUEST_MISSED,
EVENT_QUEST_UNCOMPLETED,
SERVICE_COMPLETE_QUEST,
SERVICE_CREATE_QUEST_DEFINITION,
SERVICE_MANAGE_CHILD,
SERVICE_REGENERATE,
SERVICE_UNCOMPLETE_QUEST,
)
from custom_components.nestquest.core.events import (
build_quest_uncompleted_events,
)
from custom_components.nestquest.dao_instances import QuestInstancesDao
from custom_components.nestquest.materialize import materialize
@@ -39,7 +43,6 @@ from custom_components.nestquest.quest_definitions import (
)
from custom_components.nestquest.recurrence import ScheduleRule
ADMIN_CTX = {"user_id": "admin-1"}
_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00$")
@@ -61,15 +64,7 @@ async def _setup_and_seed(hass, make_entry, *, windows=("morning",), seed_date=N
)
assert await async_setup_entry(hass, entry) is True
database = entry.runtime_data.database
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
from custom_components.nestquest.children import list_children
child = (await list_children(database))[0]
child = await create_child(database, "Ada")
await create_quest_definition(
database,
"Brush teeth",
@@ -205,32 +200,44 @@ async def test_child_day_complete_does_not_fire_partway(
async def test_uncompleted_fires_on_reversal_only(hass, make_entry) -> None:
"""An actual reversal fires the uncompleted event — but only once,
and only through the API service's transition stream: the HA
``uncomplete_quest`` service is gone, so the reversal is driven
through the core completion layer the API's admin uncomplete route
calls, with the published frame delivered by the SSE subscription."""
entry, child, instances = await _setup_and_seed(hass, make_entry)
instance_id = instances[0].id
database = entry.runtime_data.database
client = entry.runtime_data.coordinator.api_client
await hass.services.call(
DOMAIN,
SERVICE_COMPLETE_QUEST,
{"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
)
# Un-completing an open instance is a no-op and fires nothing.
await hass.services.call(
DOMAIN,
SERVICE_UNCOMPLETE_QUEST,
{"instance_id": instance_id, "actor": "user"},
context=ADMIN_CTX,
# The reversal appends through the core completion layer; the API
# route publishes the transition on its SSE stream and the
# subscription re-fires it on the bus.
result = await uncomplete_instance(
database, instance_id, actor_source="user", actor_user_id="admin-1"
)
assert result.appended is True
client.published_frames.extend(
await build_quest_uncompleted_events(database, instance_id)
)
await refire_api_transitions(hass, entry, client)
fired = hass.bus.fired(EVENT_QUEST_UNCOMPLETED)
assert len(fired) == 1
payload = fired[0]
assert payload["instance_id"] == instance_id
assert payload["quest_title"] == "Brush teeth"
assert _TIMESTAMP.match(payload["occurred_at"])
await hass.services.call(
DOMAIN,
SERVICE_UNCOMPLETE_QUEST,
{"instance_id": instance_id, "actor": "user"},
context=ADMIN_CTX,
# Un-completing an already-open instance is a no-op: nothing is
# appended, nothing is published, and nothing further fires.
no_op = await uncomplete_instance(
database, instance_id, actor_source="user", actor_user_id="admin-1"
)
assert no_op.appended is False
await refire_api_transitions(hass, entry, client)
assert len(hass.bus.fired(EVENT_QUEST_UNCOMPLETED)) == 1
@@ -390,40 +397,6 @@ async def test_completion_updates_entities_immediately(
)
async def test_uncompletion_updates_entities_immediately(
hass, make_entry
) -> None:
entry, child, instances = await _setup_and_seed(hass, make_entry)
coordinator = entry.runtime_data.coordinator
await coordinator.async_refresh()
await hass.services.call(
DOMAIN,
SERVICE_COMPLETE_QUEST,
{
"instance_id": instances[0].id,
"actor": "panel",
"actor_child_id": child.id,
},
)
completed_after_complete = hass.entities[
f"nestquest_child_{child.id}_quests_completed_today"
].native_value
assert completed_after_complete == 1
await hass.services.call(
DOMAIN,
SERVICE_UNCOMPLETE_QUEST,
{"instance_id": instances[0].id, "actor": "user"},
context=ADMIN_CTX,
)
completed_after_uncomplete = hass.entities[
f"nestquest_child_{child.id}_quests_completed_today"
].native_value
assert completed_after_uncomplete == 0, (
"sensor must reflect the reversal without a manual refresh"
)
async def test_forced_refreshes_serialize_no_stale_publish(
hass, make_entry
) -> None:
tests/test_lifecycle.py
+2 -131
View File
@@ -9,22 +9,14 @@ from conftest import make_config_entry, make_hass, wire_entry_to_registry
from custom_components.nestquest import DOMAIN, async_setup_entry, async_unload_entry
from custom_components.nestquest.const import (
CONF_ADMIN_USER_IDS,
CONF_DAY_ROLLOVER_TIME,
CONF_HORIZON_DAYS,
CONF_MORNING_SUMMARY_TIME,
DEFAULT_HORIZON_DAYS,
SERVICE_REGENERATE,
SERVICE_COMPLETE_QUEST,
DOMAIN as DOMAIN_CONST,
)
ADMIN_ID = "admin-1"
ADMIN_CTX = {"user_id": ADMIN_ID}
def _admin_data():
return {CONF_ADMIN_USER_IDS: [ADMIN_ID]}
def _wire(entry, registry):
return wire_entry_to_registry(entry, registry)
@@ -601,127 +593,6 @@ async def test_setup_keeps_horizon_when_sibling_option_is_bad(
assert entry.runtime_data.settings.horizon_days == 5
async def test_regenerate_service_registered_and_materializes(
hass, make_entry
) -> None:
"""``nestquest.regenerate`` is registered and calling it regenerates."""
import datetime
from custom_components.nestquest.dao_instances import QuestInstancesDao
entry = _wire(make_entry(data=_admin_data()), hass.registry)
assert await async_setup_entry(hass, entry) is True
assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
# A definition created AFTER setup has no instances yet (create does not
# materialize); the service must generate them on demand.
database = entry.runtime_data.database
child_id = await _seed_daily_child_and_definition(database, _local_today(hass))
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
today = _local_today(hass)
end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
records = await QuestInstancesDao(database).list_by_date_range(
child_id, today.isoformat(), end
)
assert len(records) == DEFAULT_HORIZON_DAYS + 1
async def test_regenerate_service_uses_configured_horizon_days(
hass, make_entry
) -> None:
"""``nestquest.regenerate`` regenerates only the configured horizon."""
import datetime
from custom_components.nestquest.dao_instances import QuestInstancesDao
entry = _wire(
make_entry(options={CONF_HORIZON_DAYS: 5}, data=_admin_data()),
hass.registry,
)
assert await async_setup_entry(hass, entry) is True
database = entry.runtime_data.database
child_id = await _seed_daily_child_and_definition(database, _local_today(hass))
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
today = _local_today(hass)
end = (today + datetime.timedelta(days=5)).isoformat()
records = await QuestInstancesDao(database).list_by_date_range(
child_id, today.isoformat(), end
)
assert len(records) == 6
beyond = (today + datetime.timedelta(days=6)).isoformat()
beyond_end = (
today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
).isoformat()
assert await QuestInstancesDao(database).list_by_date_range(
child_id, beyond, beyond_end
) == []
async def test_unload_deregisters_regenerate_service(hass, make_entry) -> None:
"""Unload removes the ``regenerate`` service so no stale entry leaks."""
entry = _wire(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
assert await async_unload_entry(hass, entry) is True
assert not hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
async def test_regenerate_service_is_domain_scoped_across_entries(
hass, make_entry
) -> None:
"""The ``regenerate`` service is domain-global, not per-entry.
With two loaded entries the service must be registered and functional;
unloading one entry must keep it registered and functional for the
remaining entry; unloading the final entry must remove it.
"""
import datetime
from custom_components.nestquest.dao_instances import QuestInstancesDao
entry_a = _wire(
make_entry(entry_id="entry_a", data=_admin_data()), hass.registry
)
entry_b = _wire(
make_entry(entry_id="entry_b", data=_admin_data()), hass.registry
)
assert await async_setup_entry(hass, entry_a) is True
assert await async_setup_entry(hass, entry_b) is True
assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
# The domain-global handler resolves a live database from any entry.
database_b = entry_b.runtime_data.database
child_id = await _seed_daily_child_and_definition(
database_b, _local_today(hass)
)
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
today = _local_today(hass)
end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
records = await QuestInstancesDao(database_b).list_by_date_range(
child_id, today.isoformat(), end
)
assert len(records) == DEFAULT_HORIZON_DAYS + 1
# Unloading one entry must not tear down the shared service.
assert await async_unload_entry(hass, entry_a) is True
assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
# It still works for the remaining entry.
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
# Unloading the final entry removes the domain-global service.
assert await async_unload_entry(hass, entry_b) is True
assert not hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
async def test_setup_cleans_listeners_when_materialization_raises(
hass, make_entry, monkeypatch
) -> None:
@@ -745,5 +616,5 @@ async def test_setup_cleans_listeners_when_materialization_raises(
assert hass.time_change.size == 0
assert hass.registry.size == 0
assert not hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
assert not hass.services.has_service(DOMAIN, SERVICE_COMPLETE_QUEST)
assert not hasattr(entry, "runtime_data")
tests/test_permission_matrix.py
-238
View File
@@ -1,238 +0,0 @@
"""Feature 09 permission test matrix across every registered service."""
from __future__ import annotations
import datetime
import pytest
from homeassistant.exceptions import Unauthorized
from conftest import wire_entry_to_registry
from custom_components.nestquest import async_setup_entry
from custom_components.nestquest.const import (
CONF_ADMIN_USER_IDS,
DEFAULT_HORIZON_DAYS,
DOMAIN,
DOMAIN_SERVICES,
SERVICE_COMPLETE_QUEST,
SERVICE_CREATE_PRESENCE_OVERRIDE,
SERVICE_CREATE_QUEST_DEFINITION,
SERVICE_DELETE_PRESENCE_OVERRIDE,
SERVICE_EXPORT_HISTORY_CSV,
SERVICE_MANAGE_CHILD,
SERVICE_REGENERATE,
SERVICE_SET_PRESENCE_PATTERN,
SERVICE_SET_QUEST_DEFINITION_ACTIVE,
SERVICE_UNCOMPLETE_QUEST,
SERVICE_UPDATE_QUEST_DEFINITION,
)
from custom_components.nestquest.service_policy import (
ADMIN_ONLY,
OPEN,
SERVICE_POLICY,
)
ADMIN_ID = "admin-1"
ADMIN_CTX = {"user_id": ADMIN_ID}
NON_ADMIN_ID = "non-admin"
NON_ADMIN_CTX = {"user_id": NON_ADMIN_ID}
KIOSK_ID = "kiosk-panel"
KIOSK_CTX = {"user_id": KIOSK_ID}
def test_every_registered_service_is_in_the_operation_registry() -> None:
"""The registry is total: a service outside it could never be gated."""
missing = [
service for service in DOMAIN_SERVICES if service not in SERVICE_POLICY
]
extra = [
service for service in SERVICE_POLICY if service not in DOMAIN_SERVICES
]
assert not missing, f"registered but ungated: {missing}"
assert not extra, f"gated but never registered: {extra}"
assert set(SERVICE_POLICY.values()) == {OPEN, ADMIN_ONLY}
async def _setup(hass, make_entry, *, admin=True):
data = {CONF_ADMIN_USER_IDS: [ADMIN_ID]} if admin else {}
entry = wire_entry_to_registry(make_entry(data=data), hass.registry)
assert await async_setup_entry(hass, entry) is True
return entry
async def _seed(hass, entry) -> dict:
"""Seed one child, one definition, and materialized instances."""
from custom_components.nestquest.children import list_children
from custom_components.nestquest.dao_instances import QuestInstancesDao
database = entry.runtime_data.database
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
await hass.services.call(
DOMAIN,
SERVICE_CREATE_QUEST_DEFINITION,
{
"title": "Brush teeth",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [(await list_children(database))[0].id],
"windows": ["morning"],
},
context=ADMIN_CTX,
)
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
child = (await list_children(database))[0]
today = datetime.date.today().isoformat()
end = (
datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
).isoformat()
instances = await QuestInstancesDao(database).list_by_date_range(
child.id, today, end
)
assert instances
from custom_components.nestquest.dao_presence import (
PresenceOverridesDao,
)
return {
"child_id": child.id,
"instance_id": instances[0].id,
"definition_id": instances[0].definition_id,
"override": await PresenceOverridesDao(database).create(
child.id, "2027-01-01", "2027-01-02", False
),
"today": today,
"database": database,
}
def _admin_success_payload(service: str, ctx: dict) -> dict | None:
"""A schema-valid, end-to-end-successful payload per admin service."""
if service == SERVICE_UNCOMPLETE_QUEST:
return {"instance_id": ctx["instance_id"], "actor": "user"}
if service == SERVICE_CREATE_QUEST_DEFINITION:
return {
"title": "Second quest",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [ctx["child_id"]],
"windows": ["afternoon"],
}
if service == SERVICE_UPDATE_QUEST_DEFINITION:
return {"definition_id": ctx["definition_id"], "title": "Brush teeth 2"}
if service == SERVICE_SET_QUEST_DEFINITION_ACTIVE:
return {"definition_id": ctx["definition_id"], "is_active": False}
if service == SERVICE_SET_PRESENCE_PATTERN:
return {
"child_id": ctx["child_id"],
"cycle_length_weeks": 1,
"anchor_date": ctx["today"],
"pattern": {0: [0, 1, 2, 3, 4, 5, 6]},
}
if service == SERVICE_CREATE_PRESENCE_OVERRIDE:
return {
"child_id": ctx["child_id"],
"start_date": "2027-03-01",
"end_date": "2027-03-02",
"is_present": False,
}
if service == SERVICE_DELETE_PRESENCE_OVERRIDE:
return {"override_id": ctx["override"].id}
if service == SERVICE_EXPORT_HISTORY_CSV:
return None # raises HomeAssistantError (Feature 13) — not Unauthorized
if service == SERVICE_MANAGE_CHILD:
return {"action": "create", "display_name": "Bo"}
if service == SERVICE_REGENERATE:
return {}
raise AssertionError(f"unexpected service {service}")
_ADMIN_ONLY = [
service for service, policy in SERVICE_POLICY.items() if policy == ADMIN_ONLY
]
@pytest.mark.parametrize("service", _ADMIN_ONLY)
async def test_admin_succeeds_on_every_admin_only_service(
hass, make_entry, service
) -> None:
from homeassistant.exceptions import HomeAssistantError
entry = await _setup(hass, make_entry)
ctx = await _seed(hass, entry)
if service == SERVICE_EXPORT_HISTORY_CSV:
# The gate passes; the Feature 13 stub raises its "not
# implemented" HomeAssistantError — NOT Unauthorized.
with pytest.raises(HomeAssistantError, match="Feature 13"):
await hass.services.call(
DOMAIN, service, None, context=ADMIN_CTX
)
return
await hass.services.call(
DOMAIN, service, _admin_success_payload(service, ctx), context=ADMIN_CTX
)
@pytest.mark.parametrize("service", _ADMIN_ONLY)
async def test_non_admin_refused_on_every_admin_only_service(
hass, make_entry, service
) -> None:
entry = await _setup(hass, make_entry)
ctx = await _seed(hass, entry)
if service != SERVICE_EXPORT_HISTORY_CSV:
await hass.services.call(
DOMAIN,
service,
_admin_success_payload(service, ctx),
context=ADMIN_CTX,
)
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
service,
_admin_success_payload(service, ctx),
context=NON_ADMIN_CTX,
)
async def test_non_admin_allowed_on_complete_quest(hass, make_entry) -> None:
from custom_components.nestquest.completion import instance_state
entry = await _setup(hass, make_entry)
ctx = await _seed(hass, entry)
await hass.services.call(
DOMAIN,
SERVICE_COMPLETE_QUEST,
{
"instance_id": ctx["instance_id"],
"actor": "panel",
"actor_child_id": ctx["child_id"],
},
context=NON_ADMIN_CTX,
)
assert (
await instance_state(ctx["database"], ctx["instance_id"]) == "done"
)
async def test_kiosk_refused_on_uncomplete_and_regenerate(
hass, make_entry
) -> None:
entry = await _setup(hass, make_entry)
ctx = await _seed(hass, entry)
from custom_components.nestquest.admin_allowlist import list_admin_ids
assert KIOSK_ID not in await list_admin_ids(ctx["database"])
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_UNCOMPLETE_QUEST,
{"instance_id": ctx["instance_id"], "actor": "user"},
context=KIOSK_CTX,
)
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN, SERVICE_REGENERATE, context=KIOSK_CTX
)
tests/test_permissions.py
-122
View File
@@ -1,122 +0,0 @@
"""Tests for the Feature 09 permission gate's audit logging."""
from __future__ import annotations
import logging
import re
import pytest
from homeassistant.exceptions import Unauthorized
from conftest import wire_entry_to_registry
from custom_components.nestquest import async_setup_entry
from custom_components.nestquest.const import (
CONF_ADMIN_USER_IDS,
DOMAIN,
SERVICE_MANAGE_CHILD,
)
ADMIN_ID = "admin-1"
ADMIN_CTX = {"user_id": ADMIN_ID}
_TIMESTAMP = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}")
async def _set_up(hass, make_entry):
entry = wire_entry_to_registry(
make_entry(data={CONF_ADMIN_USER_IDS: [ADMIN_ID]}), hass.registry
)
assert await async_setup_entry(hass, entry) is True
return entry
async def test_denied_call_logs_warning_with_fields(
hass, make_entry, caplog
) -> None:
entry = await _set_up(hass, make_entry)
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context={"user_id": "kiosk-panel"},
)
denials = [
record
for record in caplog.records
if record.levelno == logging.WARNING
and "denied" in record.getMessage()
]
assert len(denials) == 1
message = denials[0].getMessage()
assert SERVICE_MANAGE_CHILD in message
assert "kiosk-panel" in message
assert _TIMESTAMP.search(message), message
async def test_denied_no_context_logs_user_none(
hass, make_entry, caplog
) -> None:
entry = await _set_up(hass, make_entry)
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
)
denials = [
record
for record in caplog.records
if record.levelno == logging.WARNING
and "denied" in record.getMessage()
]
assert len(denials) == 1
message = denials[0].getMessage()
assert "none" in message
assert _TIMESTAMP.search(message), message
async def test_successful_admin_call_logs_debug_with_fields(
hass, make_entry, caplog
) -> None:
caplog.set_level(logging.DEBUG, logger="custom_components.nestquest")
entry = await _set_up(hass, make_entry)
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
allowed = [
record
for record in caplog.records
if record.levelno == logging.DEBUG
and "allowed" in record.getMessage()
]
assert len(allowed) == 1
message = allowed[0].getMessage()
assert SERVICE_MANAGE_CHILD in message
assert ADMIN_ID in message
assert _TIMESTAMP.search(message), message
async def test_denied_log_carries_no_personal_data_beyond_user_id(
hass, make_entry, caplog
) -> None:
"""The log line names the caller's user id only: the payload's
display name must not leak into the audit trail."""
entry = await _set_up(hass, make_entry)
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context={"user_id": "kiosk-panel"},
)
denials = [
record.getMessage()
for record in caplog.records
if record.levelno == logging.WARNING and "denied" in record.getMessage()
]
assert denials, "expected a denial log line"
assert "Ada" not in denials[0]
assert "display_name" not in denials[0]
tests/test_service_policy.py
-32
View File
@@ -1,32 +0,0 @@
"""Tests for the Feature 09 admin-only versus open service registry."""
from __future__ import annotations
from custom_components.nestquest.const import (
DOMAIN_SERVICES,
SERVICE_COMPLETE_QUEST,
)
from custom_components.nestquest.service_policy import (
ADMIN_ONLY,
OPEN,
SERVICE_POLICY,
)
def test_policy_covers_exactly_the_registered_services() -> None:
assert frozenset(SERVICE_POLICY) == frozenset(DOMAIN_SERVICES)
def test_only_complete_quest_is_open() -> None:
open_services = {
name for name, access in SERVICE_POLICY.items() if access == OPEN
}
assert open_services == {SERVICE_COMPLETE_QUEST}
assert all(
access == ADMIN_ONLY
for name, access in SERVICE_POLICY.items()
if name != SERVICE_COMPLETE_QUEST
)
def test_policy_values_are_only_open_or_admin_only() -> None:
assert set(SERVICE_POLICY.values()) <= {OPEN, ADMIN_ONLY}
tests/test_services.py
+40 -472
View File
@@ -1,4 +1,13 @@
"""Tests for NestQuest service declarations and domain-global handlers."""
"""Tests for NestQuest service declarations and domain-global handlers.
Since Feature 18 the integration is a client of the NestQuest API
service: the only HA service left is the panel completion path
(``nestquest.complete_quest``, a proxy to the API's panel complete
route). Every admin service was removed — the admin surface is the
PWA and the API service's admin routes — so these tests pin the
shrunk surface: services.yaml declares nothing else, setup registers
nothing else, and no removed admin service name may come back.
"""
from __future__ import annotations
import re
@@ -6,27 +15,16 @@ from pathlib import Path
import pytest
import voluptuous as vol
from homeassistant.exceptions import HomeAssistantError, Unauthorized
from homeassistant.exceptions import HomeAssistantError
from conftest import wire_entry_to_registry
from custom_components.nestquest import async_setup_entry, async_unload_entry
from custom_components.nestquest.const import (
CONF_ADMIN_USER_IDS,
DOMAIN,
DOMAIN_SERVICES,
SERVICE_COMPLETE_QUEST,
SERVICE_CREATE_PRESENCE_OVERRIDE,
SERVICE_CREATE_QUEST_DEFINITION,
SERVICE_DELETE_PRESENCE_OVERRIDE,
SERVICE_EXPORT_HISTORY_CSV,
SERVICE_MANAGE_CHILD,
SERVICE_REGENERATE,
SERVICE_SET_PRESENCE_PATTERN,
SERVICE_SET_QUEST_DEFINITION_ACTIVE,
SERVICE_UNCOMPLETE_QUEST,
)
from custom_components.nestquest.service_policy import ADMIN_ONLY, SERVICE_POLICY
SERVICES_PATH = (
Path(__file__).parent.parent
@@ -35,21 +33,27 @@ SERVICES_PATH = (
/ "services.yaml"
)
#: The admin services removed in Feature 18. None of these names may
#: be registered as an HA service again — the admin surface is the PWA
#: and the API service's admin routes.
REMOVED_ADMIN_SERVICES = (
"uncomplete_quest",
"create_quest_definition",
"update_quest_definition",
"set_quest_definition_active",
"set_presence_pattern",
"create_presence_override",
"delete_presence_override",
"export_history_csv",
"manage_child",
"regenerate",
)
def _wire(entry, registry):
return wire_entry_to_registry(entry, registry)
ADMIN_ID = "admin-1"
ADMIN_CTX = {"user_id": ADMIN_ID}
def _make_admin_entry(make_entry):
from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
return make_entry(data={CONF_ADMIN_USER_IDS: [ADMIN_ID]})
def _service_section(text: str, name: str) -> str:
match = re.search(rf"^{name}:\s*$", text, re.MULTILINE)
assert match is not None, f"services.yaml must declare {name!r}"
@@ -77,42 +81,33 @@ def test_services_yaml_declares_every_canonical_service() -> None:
)
def test_services_yaml_declares_regenerate() -> None:
def test_services_yaml_no_longer_declares_removed_admin_services() -> None:
"""The admin services are gone from the HA service surface."""
text = SERVICES_PATH.read_text(encoding="utf-8")
assert re.search(r"^regenerate:\s*$", text, re.MULTILINE), (
"services.yaml must declare the 'regenerate' service"
)
for name in REMOVED_ADMIN_SERVICES:
assert not re.search(rf"^{name}:\s*$", text, re.MULTILINE), (
f"removed admin service {name!r} must not be declared"
)
def test_services_yaml_regenerate_has_description_and_no_fields() -> None:
"""The regenerate service is documented with a clear description and no
required fields (there is no ``fields`` block under it)."""
text = SERVICES_PATH.read_text(encoding="utf-8")
section = _service_section(text, "regenerate")
assert re.search(r"description:", section), (
"regenerate must carry a description"
)
assert "fields:" not in section, (
"regenerate declares no required fields"
)
def test_services_yaml_other_services_declare_fields() -> None:
def test_services_yaml_canonical_services_declare_fields() -> None:
text = SERVICES_PATH.read_text(encoding="utf-8")
for name in DOMAIN_SERVICES:
if name in (SERVICE_REGENERATE, SERVICE_EXPORT_HISTORY_CSV):
continue
section = _service_section(text, name)
assert "fields:" in section, f"{name} must declare fields"
async def test_all_canonical_services_registered_after_setup(
async def test_only_complete_quest_registered_after_setup(
hass, make_entry
) -> None:
"""Setup registers exactly the panel completion service — and no
removed admin service under any name."""
entry = _wire(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
for service in DOMAIN_SERVICES:
assert hass.services.has_service(DOMAIN, service), service
for service in REMOVED_ADMIN_SERVICES:
assert not hass.services.has_service(DOMAIN, service), service
async def test_unload_last_entry_removes_all_services(hass, make_entry) -> None:
@@ -138,271 +133,6 @@ async def test_unload_one_of_two_entries_keeps_services(
assert not hass.services.has_service(DOMAIN, service), service
async def test_manage_child_create_persists_child(hass, make_entry) -> None:
from custom_components.nestquest.children import list_children
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
children = await list_children(entry.runtime_data.database)
assert [child.display_name for child in children] == ["Ada"]
async def test_create_and_set_quest_definition_active(
hass, make_entry
) -> None:
from custom_components.nestquest.children import list_children
from custom_components.nestquest.quest_definitions import (
list_active_definitions,
)
from custom_components.nestquest.dao_rules import QuestDefinitionsDao
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
child = (await list_children(entry.runtime_data.database))[0]
await hass.services.call(
DOMAIN,
SERVICE_CREATE_QUEST_DEFINITION,
{
"title": "Brush teeth",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [child.id],
"windows": ["morning"],
},
context=ADMIN_CTX,
)
active = await list_active_definitions(entry.runtime_data.database)
assert len(active) == 1
definition_id = active[0].definition.id
await hass.services.call(
DOMAIN,
SERVICE_SET_QUEST_DEFINITION_ACTIVE,
{"definition_id": definition_id, "is_active": False},
context=ADMIN_CTX,
)
assert await list_active_definitions(entry.runtime_data.database) == []
stored = await QuestDefinitionsDao(entry.runtime_data.database).get(
definition_id
)
assert stored is not None
assert stored.is_active is False
async def test_complete_and_uncomplete_quest_call_business_layer(
hass, make_entry
) -> None:
import datetime
from custom_components.nestquest.children import list_children
from custom_components.nestquest.completion import instance_state
from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
from custom_components.nestquest.dao_instances import QuestInstancesDao
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
child = (await list_children(entry.runtime_data.database))[0]
await hass.services.call(
DOMAIN,
SERVICE_CREATE_QUEST_DEFINITION,
{
"title": "Brush teeth",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [child.id],
"windows": ["morning"],
},
context=ADMIN_CTX,
)
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
today = datetime.date.today().isoformat()
end = (
datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
).isoformat()
records = await QuestInstancesDao(
entry.runtime_data.database
).list_by_date_range(child.id, today, end)
assert records
instance_id = records[0].id
await hass.services.call(
DOMAIN,
SERVICE_COMPLETE_QUEST,
{
"instance_id": instance_id,
"actor": "panel",
"actor_child_id": child.id,
},
)
assert (
await instance_state(entry.runtime_data.database, instance_id)
== "done"
)
await hass.services.call(
DOMAIN,
SERVICE_UNCOMPLETE_QUEST,
{
"instance_id": instance_id,
"actor": "user",
"actor_user_id": ADMIN_ID,
},
context=ADMIN_CTX,
)
assert (
await instance_state(entry.runtime_data.database, instance_id)
== "open"
)
async def test_set_presence_pattern_regenerates_child_instances(
hass, make_entry
) -> None:
import datetime
from custom_components.nestquest.children import list_children
from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
from custom_components.nestquest.dao_instances import QuestInstancesDao
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
child = (await list_children(entry.runtime_data.database))[0]
await hass.services.call(
DOMAIN,
SERVICE_CREATE_QUEST_DEFINITION,
{
"title": "Brush teeth",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [child.id],
"windows": ["morning"],
},
context=ADMIN_CTX,
)
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
today = datetime.date.today().isoformat()
end = (
datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
).isoformat()
dao = QuestInstancesDao(entry.runtime_data.database)
assert await dao.list_by_date_range(child.id, today, end)
await hass.services.call(
DOMAIN,
SERVICE_SET_PRESENCE_PATTERN,
{
"child_id": child.id,
"cycle_length_weeks": 1,
"anchor_date": today,
"pattern": {0: []},
},
context=ADMIN_CTX,
)
assert await dao.list_by_date_range(child.id, today, end) == []
async def test_presence_override_create_and_delete_regenerate(
hass, make_entry
) -> None:
import datetime
from custom_components.nestquest.children import list_children
from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
from custom_components.nestquest.dao_instances import QuestInstancesDao
from custom_components.nestquest.dao_presence import PresenceOverridesDao
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
child = (await list_children(entry.runtime_data.database))[0]
await hass.services.call(
DOMAIN,
SERVICE_CREATE_QUEST_DEFINITION,
{
"title": "Brush teeth",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [child.id],
"windows": ["morning"],
},
context=ADMIN_CTX,
)
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
today = datetime.date.today()
end = today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
dao = QuestInstancesDao(entry.runtime_data.database)
before = await dao.list_by_date_range(
child.id, today.isoformat(), end.isoformat()
)
assert before
await hass.services.call(
DOMAIN,
SERVICE_CREATE_PRESENCE_OVERRIDE,
{
"child_id": child.id,
"start_date": today.isoformat(),
"end_date": end.isoformat(),
"is_present": False,
},
context=ADMIN_CTX,
)
assert (
await dao.list_by_date_range(
child.id, today.isoformat(), end.isoformat()
)
== []
)
override = (
await PresenceOverridesDao(
entry.runtime_data.database
).list_by_child_and_range(
child.id, today.isoformat(), end.isoformat()
)
)[0]
await hass.services.call(
DOMAIN,
SERVICE_DELETE_PRESENCE_OVERRIDE,
{"override_id": override.id},
context=ADMIN_CTX,
)
after = await dao.list_by_date_range(
child.id, today.isoformat(), end.isoformat()
)
assert len(after) == len(before)
async def test_export_history_csv_raises_home_assistant_error(
hass, make_entry
) -> None:
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
with pytest.raises(HomeAssistantError, match="Feature 13"):
await hass.services.call(
DOMAIN, SERVICE_EXPORT_HISTORY_CSV, context=ADMIN_CTX
)
async def test_invalid_schema_data_surfaces_as_error(hass, make_entry) -> None:
entry = _wire(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
@@ -414,168 +144,6 @@ async def test_invalid_schema_data_surfaces_as_error(hass, make_entry) -> None:
SERVICE_COMPLETE_QUEST,
{"instance_id": True, "actor": "panel", "actor_child_id": 1},
)
with pytest.raises((vol.Invalid, ValueError, HomeAssistantError)):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create"},
)
async def test_admin_only_service_denies_non_admin(hass, make_entry) -> None:
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context={"user_id": "kiosk-panel"},
)
async def test_admin_only_service_denies_missing_user_context(
hass, make_entry
) -> None:
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
)
async def test_admin_only_service_denies_admin_on_empty_allowlist(
hass, make_entry
) -> None:
"""Even the allowlisted id is denied when the allowlist is empty."""
entry = _wire(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
async def test_unauthorized_is_raised_before_handler_runs(
hass, make_entry
) -> None:
from custom_components.nestquest.children import list_children
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context={"user_id": "kiosk-panel"},
)
assert await list_children(entry.runtime_data.database) == []
_MINIMAL_PAYLOADS = {
"uncomplete_quest": {"instance_id": 1, "actor": "user"},
"create_quest_definition": {
"title": "T",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [1],
"windows": ["morning"],
},
"update_quest_definition": {"definition_id": 1},
"set_quest_definition_active": {"definition_id": 1, "is_active": True},
"set_presence_pattern": {
"child_id": 1,
"cycle_length_weeks": 1,
"anchor_date": "2026-09-19",
"pattern": {0: [0]},
},
"create_presence_override": {
"child_id": 1,
"start_date": "2026-09-19",
"end_date": "2026-09-19",
"is_present": False,
},
"delete_presence_override": {"override_id": 1},
"export_history_csv": {},
"manage_child": {"action": "create", "display_name": "Ada"},
"regenerate": {},
}
_ADMIN_ONLY_SERVICES = [
name for name, policy in SERVICE_POLICY.items() if policy == ADMIN_ONLY
]
@pytest.mark.parametrize("service", _ADMIN_ONLY_SERVICES)
async def test_admin_only_service_fails_closed_without_user_context(
hass, make_entry, service
) -> None:
"""A call with no user_id is denied on every admin-only service."""
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
with pytest.raises(Unauthorized):
await hass.services.call(
DOMAIN, service, dict(_MINIMAL_PAYLOADS[service])
)
async def test_complete_quest_proceeds_without_user_context(
hass, make_entry
) -> None:
"""The open service needs no user context: the kiosk panel taps a
child profile and the call goes through with no HA user at all."""
import datetime
from custom_components.nestquest.children import list_children
from custom_components.nestquest.completion import instance_state
from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
from custom_components.nestquest.dao_instances import QuestInstancesDao
entry = _wire(_make_admin_entry(make_entry), hass.registry)
assert await async_setup_entry(hass, entry) is True
await hass.services.call(
DOMAIN,
SERVICE_MANAGE_CHILD,
{"action": "create", "display_name": "Ada"},
context=ADMIN_CTX,
)
child = (await list_children(entry.runtime_data.database))[0]
await hass.services.call(
DOMAIN,
SERVICE_CREATE_QUEST_DEFINITION,
{
"title": "Brush teeth",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [child.id],
"windows": ["morning"],
},
context=ADMIN_CTX,
)
await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
today = datetime.date.today().isoformat()
end = (
datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
).isoformat()
instance_id = (
await QuestInstancesDao(
entry.runtime_data.database
).list_by_date_range(child.id, today, end)
)[0].id
await hass.services.call(
DOMAIN,
SERVICE_COMPLETE_QUEST,
{"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
)
assert (
await instance_state(entry.runtime_data.database, instance_id)
== "done"
)
async def test_scheduled_internals_never_call_the_service_layer(
@@ -590,4 +158,4 @@ async def test_scheduled_internals_never_call_the_service_layer(
assert "hass.services.call" not in source, (
"scheduled internals must call the materialization directly, "
"never through hass.services.call"
)
)