Test the permission gate's audit logging

Warning on denial and debug on success carry the service, the caller
user id (or none), and a UTC stamp; nothing beyond the user id leaks.

tests/test_permissions.py

@@ -0,0 +1,122 @@
+"""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_services.py

@@ -26,6 +26,7 @@ from custom_components.nestquest.const import (
     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
@@ -476,3 +477,117 @@ async def test_unauthorized_is_raised_before_handler_runs(
             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(
+    hass, make_entry
+) -> None:
+    """The daily rollover and startup backfill run the materialization
+    as direct function calls; they never widen the service gate."""
+    import custom_components.nestquest as nestquest
+
+    assert not hasattr(nestquest, "_run_horizon_materialization_service")
+    source = Path(nestquest.__file__).read_text(encoding="utf-8")
+    assert "hass.services.call" not in source, (
+        "scheduled internals must call the materialization directly, "
+        "never through hass.services.call"
+    )