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

Merge pull request 'Admin allowlist storage seeded with the setup user (Feature 03, task 4)' (#54) from task/288a953c-admin-allowlist into feature/4f802471-1de1-4c35-844f-eebf3bea0de5-children-registry-admin-allowlist

Browse Source
This commit was merged in pull request #54.
This commit is contained in:
dev_agent
2026-09-15 17:33:02 -04:00
parent b7f7e092a0 913d3658b8
commit 7a1cef1e54
11 changed files with 1145 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/nestquest/__init__.py
+74 -2
View File
@@ -13,10 +13,11 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import DOMAIN, LOGGER
from .const import CONF_ADMIN_USER_IDS, DOMAIN, LOGGER
from .db import NestQuestDatabase
from .migrations import apply_migrations
from .store import async_get_db_path
from .admin_allowlist import seed_setup_admin
_LOGGER = LOGGER
@@ -225,6 +226,28 @@ class NestQuestRuntimeData:
remove_update_listener: Callable[[], Any]
async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
"""Return every Home Assistant owner-account user id.
The last-resort admin seed: when no persisted allowlist copy and no
flow-context user survive to first setup, the HA owner accounts are
seeded so the household is never left with zero admins (an empty
allowlist fails closed, which would lock the household out of its
own integration). Owner accounts can already do everything in HA,
so granting NestQuest admin is not a privilege escalation.
"""
auth = getattr(hass, "auth", None)
if auth is None:
return []
users = await auth.async_get_users()
owners = [
user.id
for user in users
if getattr(user, "is_owner", False) and getattr(user, "id", None)
]
return owners
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
"""Set up the NestQuest integration. YAML configuration is not used, returns True."""
return True
@@ -241,10 +264,36 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# record without a live connection (a previous setup's
# close-on-failure, or a hand-replaced file) must get the
# missing/valid/corrupt classification too, never a raw
# sqlite3 error and never skipped migrations.
# sqlite3 error and never skipped migrations — and the
# same allowlist seeding, so a reopened fresh file is
# never ownerless.
db = await _async_open_database(
hass, await async_get_db_path(hass)
)
try:
stored_admin_ids = (
entry.options.get(CONF_ADMIN_USER_IDS)
or entry.data.get(CONF_ADMIN_USER_IDS)
)
context = getattr(entry, "context", None)
context_user_id = (
context.get("user_id")
if isinstance(context, dict)
else None
)
await seed_setup_admin(
db,
stored_admin_ids=stored_admin_ids,
context_user_id=context_user_id,
owner_ids=await _async_owner_user_ids(hass),
)
except BaseException:
# A seeding failure must not leak the just-opened
# connection: close it before the error propagates, so
# HA's setup retry starts from a clean handle (the
# first-setup path closes symmetrically).
await db.close()
raise
existing.database = db
entry.runtime_data = existing
return True
@@ -257,6 +306,29 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
database = await _async_open_database(hass, await async_get_db_path(hass))
try:
# Seed the admin allowlist on first setup so the owner is never
# locked out: an empty database takes the persisted admin copy
# — the options flow's saved list first (it is the current
# one), else the config flow's user from entry data — and only
# then the HA user still present on the entry context. No-op
# on a non-empty list (fail-closed narrowing is deliberate).
stored_admin_ids = (
entry.options.get(CONF_ADMIN_USER_IDS)
or entry.data.get(CONF_ADMIN_USER_IDS)
)
context = getattr(entry, "context", None)
context_user_id = (
context.get("user_id")
if isinstance(context, dict)
else None
)
owner_ids = await _async_owner_user_ids(hass)
await seed_setup_admin(
database,
stored_admin_ids=stored_admin_ids,
context_user_id=context_user_id,
owner_ids=owner_ids,
)
remove_update_listener = entry.add_update_listener(_async_update_listener)
except BaseException:
await database.close()
custom_components/nestquest/admin_allowlist.py
+136
View File
@@ -0,0 +1,136 @@
"""Business layer for the admin allowlist (Feature 03).
Sits between the options flow / Feature 09 service gate and the typed
:class:`~.dao_children.AdminUsersDao`. The allowlist stores Home
Assistant user IDs — never usernames or display names (feature
guardrail) — and an empty allowlist must fail closed: every function
here refuses to write an empty allowlist, so the owner can never be
accidentally locked out of admin operations by a bad save.
Shape policy: a user id is a non-empty trimmed string. Duplicate ids
in one write are rejected up front — they would silently map to one
row and hide a caller bug. Writes replace the allowlist atomically:
the diff (additions and removals) is computed and applied inside one
transaction under the connection-scoped lock shared with the DAOs, so
a concurrent reader never observes a half-applied sync.
Seed policy (:func:`seed_setup_admin`): the allowlist is seeded only
when it is EMPTY, so re-running setup never resurrects a deliberately
narrowed list. The setup user (the HA user who completed the config
flow, via the entry context) is added automatically so the owner is
never locked out on first setup; a stored entry-data copy of the list
re-seeds a lost database.
"""
from __future__ import annotations
import datetime
import logging
from .dao_children import AdminUsersDao, _connection_lock
from .db import NestQuestDatabase
LOGGER = logging.getLogger(__name__)
def _now_stamp() -> str:
"""UTC second-precision stamp, matching the registry's policy."""
return datetime.datetime.now(datetime.timezone.utc).isoformat(
timespec="seconds"
)
def _validate_user_id(value: object) -> str:
"""Return ``value`` trimmed; raise unless a non-empty string."""
if not isinstance(value, str):
raise ValueError(
f"admin user id must be a string, got {value!r}"
)
trimmed = value.strip()
if not trimmed:
raise ValueError("admin user id must not be empty")
return trimmed
async def list_admin_ids(database: NestQuestDatabase) -> list[str]:
"""Return the allowlisted HA user IDs in storage order."""
return [row.ha_user_id for row in await AdminUsersDao(database).list()]
async def set_admin_ids(
database: NestQuestDatabase, user_ids: list[str]
) -> None:
"""Replace the allowlist with exactly ``user_ids`` (atomic sync).
Refuses an empty list — an empty allowlist must fail closed, and
the only way to reach one through this API would be removing the
last admin, so that removal is refused here with a clear error.
Refuses non-string, empty, or duplicate ids up front. The diff
against the current allowlist (additions and removals) applies
inside one transaction under the connection lock.
"""
if not isinstance(user_ids, list):
raise ValueError("user_ids must be a list of HA user id strings")
normalized: list[str] = []
for value in user_ids:
normalized.append(_validate_user_id(value))
if len(set(normalized)) != len(normalized):
raise ValueError("user_ids must not contain duplicates")
if not normalized:
raise ValueError(
"refusing to empty the admin allowlist: at least one admin "
"must remain (an empty allowlist fails closed)"
)
dao = AdminUsersDao(database)
async with _connection_lock(database):
async with database.transaction():
current = {row.ha_user_id for row in await dao.list()}
target = set(normalized)
for user_id in sorted(current - target):
await dao.remove(user_id)
for user_id in normalized:
await dao.add(user_id, _now_stamp())
async def seed_setup_admin(
database: NestQuestDatabase,
*,
stored_admin_ids: list[str] | None = None,
context_user_id: str | None = None,
owner_ids: list[str] | None = None,
) -> list[str]:
"""Seed an empty allowlist so the owner is never locked out.
Only acts when the allowlist is EMPTY. Preference order: the
stored persisted copy of the list (the options flow's current
selection, then the config flow's original — disaster recovery for
a lost database), then the HA user who completed the config flow
(``context_user_id``), then — when no caller identity survived at
all — every Home Assistant OWNER account (``owner_ids``), so a
first setup whose flow context lost the user id still seeds a
working admin. Returns the allowlist after seeding. Non-empty
allowlists are returned untouched — re-running setup must never
resurrect a deliberately narrowed list.
"""
dao = AdminUsersDao(database)
async with _connection_lock(database):
current = [row.ha_user_id for row in await dao.list()]
if current:
return current
candidates: list[str] = []
if stored_admin_ids is not None:
if not isinstance(stored_admin_ids, list):
raise ValueError(
"stored_admin_ids must be a list of HA user id strings"
)
candidates = [_validate_user_id(value) for value in stored_admin_ids]
elif context_user_id is not None:
candidates = [_validate_user_id(context_user_id)]
elif owner_ids:
if not isinstance(owner_ids, list):
raise ValueError(
"owner_ids must be a list of HA user id strings"
)
candidates = [_validate_user_id(value) for value in owner_ids]
for user_id in candidates:
await dao.add(user_id, _now_stamp())
return [row.ha_user_id for row in await dao.list()]
custom_components/nestquest/config_flow.py
+18 -3
View File
@@ -7,7 +7,7 @@ from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN
from .const import CONF_ADMIN_USER_IDS, DOMAIN
class NestQuestConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@@ -28,7 +28,14 @@ class NestQuestConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user step of the config flow."""
"""Handle the user step of the config flow.
The HA user completing the flow is persisted into the entry
data as the initial admin allowlist copy: real HA does not
guarantee the flow context survives to setup, and entry data is
the persisted record first setup seeds from — the owner is
never locked out.
"""
await self.async_set_unique_id(DOMAIN, raise_on_progress=False)
if self._async_in_progress() or self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
@@ -36,4 +43,12 @@ class NestQuestConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is None:
return self.async_show_form(step_id="user")
return self.async_create_entry(title="NestQuest", data=user_input)
data = dict(user_input)
flow_user_id = (
self.context.get("user_id")
if isinstance(self.context, dict)
else None
)
if flow_user_id:
data[CONF_ADMIN_USER_IDS] = [flow_user_id]
return self.async_create_entry(title="NestQuest", data=data)
custom_components/nestquest/const.py
+6
View File
@@ -21,6 +21,12 @@ DEFAULT_DAY_ROLLOVER_TIME = "00:00"
CONF_PANEL_IDLE_TIMEOUT = "panel_idle_timeout"
DEFAULT_PANEL_IDLE_TIMEOUT = 300
# Admin allowlist (Feature 03): HA user IDs permitted to call admin-only
# services. Stored as IDs, never usernames (feature guardrail). The
# database allowlist is authoritative; the config entry carries the same
# list as a disaster-recovery copy so a lost database can be re-seeded.
CONF_ADMIN_USER_IDS = "admin_user_ids"
# Quest windows (D-008): a definition may declare several day windows;
# each produces its own instance per day and groups the panel's Quest
# Log into three columns. The clock ranges are stored here so they stay
custom_components/nestquest/options_flow.py
+146 -25
View File
@@ -7,15 +7,19 @@ from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv
from .const import (
CONF_ADMIN_USER_IDS,
CONF_DAY_ROLLOVER_TIME,
CONF_HORIZON_DAYS,
CONF_PANEL_IDLE_TIMEOUT,
DEFAULT_DAY_ROLLOVER_TIME,
DEFAULT_HORIZON_DAYS,
DEFAULT_PANEL_IDLE_TIMEOUT,
DOMAIN,
)
from .admin_allowlist import list_admin_ids, set_admin_ids
MIN_HORIZON_DAYS = 1
MIN_PANEL_IDLE_TIMEOUT = 30
@@ -55,51 +59,168 @@ def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
return errors
def _build_schema(current: dict[str, Any]) -> vol.Schema:
"""Build the form schema, pre-filled with the current values."""
return vol.Schema(
{
vol.Required(
CONF_HORIZON_DAYS, default=current[CONF_HORIZON_DAYS]
): int,
vol.Required(
CONF_DAY_ROLLOVER_TIME, default=current[CONF_DAY_ROLLOVER_TIME]
): str,
vol.Required(
CONF_PANEL_IDLE_TIMEOUT, default=current[CONF_PANEL_IDLE_TIMEOUT]
): int,
}
)
def _build_schema(
current: dict[str, Any],
*,
admin_choices: dict[str, str] | None = None,
admin_default: list[str] | None = None,
) -> vol.Schema:
"""Build the form schema, pre-filled with the current values.
The admin picker is a multi-select of EXISTING Home Assistant users
(never a free-text field — the allowlist stores user IDs); its
default is the current database allowlist, so an untouched submit
re-saves exactly who is already allowed.
"""
fields: dict[Any, Any] = {
vol.Required(
CONF_HORIZON_DAYS, default=current[CONF_HORIZON_DAYS]
): int,
vol.Required(
CONF_DAY_ROLLOVER_TIME, default=current[CONF_DAY_ROLLOVER_TIME]
): str,
vol.Required(
CONF_PANEL_IDLE_TIMEOUT, default=current[CONF_PANEL_IDLE_TIMEOUT]
): int,
}
if admin_choices is not None:
# HA's supported multi-select validator: the frontend can
# serialize the field as a real picker of the given choices.
fields[
vol.Optional(
CONF_ADMIN_USER_IDS, default=list(admin_default or [])
)
] = cv.multi_select(admin_choices)
return vol.Schema(fields)
class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
"""Handle the options flow for NestQuest."""
def _async_database(self):
"""Return the runtime database for this entry, or None.
Real HA only opens the options flow for an existing entry, so
the runtime record (and its database) exists; the None branch
covers direct construction in tests and degraded states.
"""
hass_data = getattr(self.hass, "data", None)
if not isinstance(hass_data, dict):
return None
record = hass_data.get(DOMAIN, {}).get(self.config_entry.entry_id)
database = getattr(record, "database", None)
return database if getattr(database, "connected", False) else None
async def _async_admin_choices(self) -> dict[str, str]:
"""Map HA user id -> display name for the picker.
Real Home Assistant always provides ``hass.auth``; a missing
one (direct construction in tests, degraded harness) degrades
to an empty picker rather than crashing the form.
"""
auth = getattr(self.hass, "auth", None)
if auth is None:
return {}
users = await auth.async_get_users()
return {
user.id: user.name
for user in users
if getattr(user, "id", None)
}
async def _async_current_admin_ids(self) -> list[str]:
"""Return the current allowlist, or [] without a database."""
database = self._async_database()
if database is None:
return []
return await list_admin_ids(database)
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the NestQuest options."""
admin_choices = await self._async_admin_choices()
admin_default = await self._async_current_admin_ids()
if user_input is None:
return self.async_show_form(
step_id="init", data_schema=_build_schema(self._current_values())
step_id="init",
data_schema=_build_schema(
self._current_values(),
admin_choices=admin_choices,
admin_default=admin_default,
),
)
errors = _async_validate(user_input)
# The admin ids are validated only when the key is present:
# schema-validated submits always carry it (vol.Optional's
# default fills from the current allowlist), while raw callers
# omitting the key keep the field untouched.
admin_ids_present = CONF_ADMIN_USER_IDS in user_input
admin_ids = user_input.get(CONF_ADMIN_USER_IDS)
if errors:
return self.async_show_form(
step_id="init",
data_schema=_build_schema(self._current_values()),
data_schema=_build_schema(
self._current_values(),
admin_choices=admin_choices,
admin_default=admin_default,
),
errors=errors,
)
if admin_ids_present and (
not isinstance(admin_ids, list)
or any(not isinstance(value, str) for value in admin_ids)
):
errors[CONF_ADMIN_USER_IDS] = "invalid_admin"
elif admin_ids_present and not admin_ids:
# An empty allowlist fails closed: saving zero admins is
# refused, never silently applied.
errors[CONF_ADMIN_USER_IDS] = "no_admins"
elif admin_ids_present and any(
value not in admin_choices for value in admin_ids
):
# The schema enforces membership for real HA submits; this
# backs the raw-input path up so an unknown id can never
# reach the allowlist.
errors[CONF_ADMIN_USER_IDS] = "invalid_admin"
elif admin_ids_present and len(set(admin_ids)) != len(admin_ids):
# set_admin_ids refuses duplicates with ValueError; a flow
# submission must surface that as a field error instead of
# an unhandled crash.
errors[CONF_ADMIN_USER_IDS] = "invalid_admin"
if errors:
return self.async_show_form(
step_id="init",
data_schema=_build_schema(
self._current_values(),
admin_choices=admin_choices,
admin_default=admin_default,
),
errors=errors,
)
return self.async_create_entry(
title="NestQuest",
data={
CONF_HORIZON_DAYS: user_input[CONF_HORIZON_DAYS],
CONF_DAY_ROLLOVER_TIME: user_input[CONF_DAY_ROLLOVER_TIME],
CONF_PANEL_IDLE_TIMEOUT: user_input[CONF_PANEL_IDLE_TIMEOUT],
},
)
# Apply the picker's selection to the allowlist when the
# runtime database is reachable; the selection is always
# stored on the entry as the disaster-recovery copy.
database = self._async_database()
if database is not None and admin_ids_present:
await set_admin_ids(database, admin_ids)
data = {
CONF_HORIZON_DAYS: user_input[CONF_HORIZON_DAYS],
CONF_DAY_ROLLOVER_TIME: user_input[CONF_DAY_ROLLOVER_TIME],
CONF_PANEL_IDLE_TIMEOUT: user_input[CONF_PANEL_IDLE_TIMEOUT],
}
if admin_ids_present:
data[CONF_ADMIN_USER_IDS] = admin_ids
elif admin_default:
# HA replaces entry.options wholesale, so an omitted picker
# field must carry the CURRENT allowlist forward: dropping
# it would leave only the stale config-flow copy in entry
# data, and a later database loss would resurrect admins
# the owner has since removed.
data[CONF_ADMIN_USER_IDS] = list(admin_default)
return self.async_create_entry(title="NestQuest", data=data)
def _current_values(self) -> dict[str, Any]:
"""Resolve current values from entry options, then entry data, then defaults."""
custom_components/nestquest/strings.json
+6 -3
View File
@@ -22,13 +22,16 @@
"data": {
"horizon_days": "Generation horizon (days)",
"day_rollover_time": "Day rollover time (HH:MM)",
"panel_idle_timeout": "Panel idle timeout (seconds)"
"panel_idle_timeout": "Panel idle timeout (seconds)",
"admin_user_ids": "Admin users (parents)"
}
}
},
"errors": {
"invalid": "Invalid value.",
"invalid_time": "Time must use 24-hour HH:MM format."
"invalid_time": "Time must use 24-hour HH:MM format.",
"no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
"invalid_admin": "Invalid admin selection."
}
}
}
}
custom_components/nestquest/translations/en.json
+6 -3
View File
@@ -22,13 +22,16 @@
"data": {
"horizon_days": "Generation horizon (days)",
"day_rollover_time": "Day rollover time (HH:MM)",
"panel_idle_timeout": "Panel idle timeout (seconds)"
"panel_idle_timeout": "Panel idle timeout (seconds)",
"admin_user_ids": "Admin users (parents)"
}
}
},
"errors": {
"invalid": "Invalid value.",
"invalid_time": "Time must use 24-hour HH:MM format."
"invalid_time": "Time must use 24-hour HH:MM format.",
"no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
"invalid_admin": "Invalid admin selection."
}
}
}
}
tests/conftest.py
+49 -4
View File
@@ -29,7 +29,7 @@ import tempfile
import weakref
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
from unittest.mock import AsyncMock, MagicMock
import pytest
@@ -42,6 +42,8 @@ _HA_MODULES = (
"homeassistant.config_entries",
"homeassistant.data_entry_flow",
"homeassistant.exceptions",
"homeassistant.helpers",
"homeassistant.helpers.config_validation",
)
def _find_spec(name: str):
@@ -79,6 +81,8 @@ for _parent, _child in (
("homeassistant", "data_entry_flow"),
("homeassistant", "core"),
("homeassistant", "exceptions"),
("homeassistant", "helpers"),
("homeassistant.helpers", "config_validation"),
):
setattr(sys.modules[_parent], _child, sys.modules[f"{_parent}.{_child}"])
@@ -181,6 +185,30 @@ class ConfigEntryNotReady(Exception):
_exceptions_mock.ConfigEntryNotReady = ConfigEntryNotReady
import voluptuous as vol
def _multi_select(choices: dict):
"""Mirror homeassistant.helpers.config_validation.multi_select.
The options flow's admin picker uses HA's supported multi-select
validator (schema-serializable by HA's frontend); the mock provides
the same contract: a list whose members are keys of ``choices``.
"""
def _validate(selected):
if not isinstance(selected, list):
raise vol.Invalid("Not a list")
for value in selected:
if value not in choices:
raise vol.Invalid(f"{value} is not a valid option")
return selected
return _validate
_ha_mock("homeassistant.helpers.config_validation").multi_select = _multi_select
# ---------------------------------------------------------------------------
# Shared helpers for the behavior tests.
@@ -218,12 +246,18 @@ def make_config_entry(
entry_id: str = "test_entry",
options: dict | None = None,
data: dict | None = None,
context: dict | None = None,
):
"""Create a stub config entry shaped like HA's ConfigEntry for tests."""
"""Create a stub config entry shaped like HA's ConfigEntry for tests.
``context`` mirrors HA's entry context (the flow that created the
entry carries the acting HA user under ``context["user_id"]``).
"""
entry = SimpleNamespace(
entry_id=entry_id,
options=dict(options or {}),
data=dict(data or {}),
context=dict(context or {}),
)
entry.add_update_listener = None
return entry
@@ -247,6 +281,10 @@ def make_hass() -> tuple:
config=MagicMock(),
config_entries=SimpleNamespace(async_reload=None),
)
# hass.auth mirrors the real surface the integration reads: an
# async_get_users list the tests can repoint at will (the setup
# path resolves HA owner accounts as the last-resort admin seed).
hass.auth = SimpleNamespace(async_get_users=AsyncMock(return_value=[]))
def _cleanup_config_dir():
import shutil
@@ -302,8 +340,15 @@ async def hass():
def make_entry():
"""Provide the config-entry factory as a fixture."""
def _factory(entry_id: str = "test_entry", options: dict | None = None, data: dict | None = None):
return make_config_entry(entry_id=entry_id, options=options, data=data)
def _factory(
entry_id: str = "test_entry",
options: dict | None = None,
data: dict | None = None,
context: dict | None = None,
):
return make_config_entry(
entry_id=entry_id, options=options, data=data, context=context
)
return _factory
tests/test_admin_allowlist.py
+417
View File
@@ -0,0 +1,417 @@
"""Tests for admin_allowlist.py: the Feature 03 allowlist business layer."""
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from custom_components.nestquest import async_setup_entry, async_unload_entry
from custom_components.nestquest.admin_allowlist import (
list_admin_ids,
seed_setup_admin,
set_admin_ids,
)
from custom_components.nestquest.dao_children import AdminUsersDao
from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import apply_migrations
def _run(coro):
return asyncio.new_event_loop().run_until_complete(coro)
def _make_hass_mock():
hass = MagicMock()
hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
return hass
def _with_db(tmp_path, name):
def _run_test(body):
async def _main():
database = NestQuestDatabase(_make_hass_mock())
await database.open(tmp_path / name)
try:
await apply_migrations(database)
return await body(database)
finally:
await database.close()
return _run(_main())
return _run_test
# ---------------------------------------------------------------------------
# set_admin_ids: read/write with the fail-closed guarantee
# ---------------------------------------------------------------------------
def test_set_admin_ids_round_trip(tmp_path) -> None:
async def _body(database):
await set_admin_ids(database, ["user-owner", "user-parent"])
# Membership is the contract; AdminUsersDao.list orders by id.
assert sorted(await list_admin_ids(database)) == [
"user-owner",
"user-parent",
]
# Replace: the removed id is gone, the new one present.
await set_admin_ids(database, ["user-parent", "user-grandma"])
assert sorted(await list_admin_ids(database)) == [
"user-grandma",
"user-parent",
]
return await list_admin_ids(database)
_with_db(tmp_path, "set-round-trip.db")(_body)
def test_set_admin_ids_rejects_empty_allowlist(tmp_path) -> None:
"""Removing the last admin (or saving an empty selection) is
refused with a clear error: an empty allowlist fails closed, so
the owner must never be able to reach one through this API."""
async def _body(database):
await set_admin_ids(database, ["user-owner"])
with pytest.raises(ValueError, match="at least one admin"):
await set_admin_ids(database, [])
# The refused write left the allowlist intact.
assert await list_admin_ids(database) == ["user-owner"]
return None
_with_db(tmp_path, "set-empty-refused.db")(_body)
@pytest.mark.parametrize(
"bad", [None, "notalist", [42], [None], [""], [" "], 5]
)
def test_set_admin_ids_rejects_malformed_input(tmp_path, bad) -> None:
async def _body(database):
with pytest.raises(ValueError):
await set_admin_ids(database, bad)
assert await list_admin_ids(database) == []
return None
_with_db(tmp_path, "set-malformed.db")(_body)
def test_set_admin_ids_rejects_duplicates(tmp_path) -> None:
async def _body(database):
with pytest.raises(ValueError, match="duplicates"):
await set_admin_ids(database, ["user-a", "user-a"])
assert await list_admin_ids(database) == []
return None
_with_db(tmp_path, "set-duplicates.db")(_body)
def test_set_admin_ids_trims_and_replaces_atomically(tmp_path) -> None:
async def _body(database):
await set_admin_ids(database, [" user-a "])
assert await list_admin_ids(database) == ["user-a"]
# Sync that both adds and removes in one write.
await set_admin_ids(database, ["user-b", "user-c"])
assert await list_admin_ids(database) == ["user-b", "user-c"]
return None
_with_db(tmp_path, "set-trim.db")(_body)
# ---------------------------------------------------------------------------
# seed_setup_admin: the owner is never locked out on first setup
# ---------------------------------------------------------------------------
def test_seed_adds_context_user_on_first_setup(tmp_path) -> None:
async def _body(database):
seeded = await seed_setup_admin(
database, stored_admin_ids=None, context_user_id="owner-1"
)
assert seeded == ["owner-1"]
assert await list_admin_ids(database) == ["owner-1"]
return seeded
_with_db(tmp_path, "seed-context.db")(_body)
def test_seed_prefers_stored_entry_ids_over_context(tmp_path) -> None:
async def _body(database):
seeded = await seed_setup_admin(
database,
stored_admin_ids=["user-a", "user-b"],
context_user_id="owner-1",
)
assert seeded == ["user-a", "user-b"]
return seeded
_with_db(tmp_path, "seed-stored.db")(_body)
def test_seed_leaves_nonempty_allowlist_untouched(tmp_path) -> None:
"""Re-running setup must never resurrect a deliberately narrowed
list: only an EMPTY allowlist is seeded."""
async def _body(database):
await set_admin_ids(database, ["user-kept"])
seeded = await seed_setup_admin(
database,
stored_admin_ids=["user-a", "user-b"],
context_user_id="owner-1",
)
assert seeded == ["user-kept"]
assert await list_admin_ids(database) == ["user-kept"]
return seeded
_with_db(tmp_path, "seed-narrowed.db")(_body)
def test_seed_with_nothing_available_leaves_empty(tmp_path) -> None:
async def _body(database):
seeded = await seed_setup_admin(
database, stored_admin_ids=None, context_user_id=None
)
assert seeded == []
return seeded
_with_db(tmp_path, "seed-nothing.db")(_body)
def test_seed_rejects_malformed_candidates(tmp_path) -> None:
async def _body(database):
with pytest.raises(ValueError):
await seed_setup_admin(
database, stored_admin_ids=[42], context_user_id=None
)
with pytest.raises(ValueError):
await seed_setup_admin(
database, stored_admin_ids=None, context_user_id=" "
)
assert await list_admin_ids(database) == []
return None
_with_db(tmp_path, "seed-malformed.db")(_body)
# ---------------------------------------------------------------------------
# Setup wiring: async_setup_entry seeds automatically
# ---------------------------------------------------------------------------
async def _setup_entry(hass, entry, registry):
from tests.test_lifecycle import _wire
entry = _wire(entry, registry)
await async_setup_entry(hass, entry)
return entry
async def test_setup_seeds_allowlist_from_entry_context(
hass, make_entry
) -> None:
entry = await _setup_entry(
hass,
make_entry(context={"user_id": "ha-owner"}),
hass.registry,
)
database = entry.runtime_data.database
try:
assert await list_admin_ids(database) == ["ha-owner"]
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
async def test_setup_seeds_allowlist_from_stored_entry_data(
hass, make_entry
) -> None:
"""A lost database re-seeded from the entry's stored admin copy."""
from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
entry = await _setup_entry(
hass,
make_entry(data={CONF_ADMIN_USER_IDS: ["user-a", "user-b"]}),
hass.registry,
)
database = entry.runtime_data.database
try:
assert await list_admin_ids(database) == ["user-a", "user-b"]
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
async def test_setup_without_context_leaves_allowlist_empty(
hass, make_entry
) -> None:
"""Fail closed: no context user and no stored copy means an empty
allowlist — nobody is admin until the owner configures it."""
entry = await _setup_entry(hass, make_entry(), hass.registry)
database = entry.runtime_data.database
try:
assert await list_admin_ids(database) == []
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
async def test_setup_does_not_reseed_narrowed_allowlist(
hass, make_entry
) -> None:
from custom_components.nestquest import async_unload_entry
from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
entry = await _setup_entry(
hass, make_entry(context={"user_id": "ha-owner"}), hass.registry
)
database = entry.runtime_data.database
await set_admin_ids(database, ["user-parent"])
await database.close()
assert await async_unload_entry(hass, entry) is True
# Reload with the SAME context: the narrowed list survives.
entry_reloaded = await _setup_entry(
hass, make_entry(context={"user_id": "ha-owner"}), hass.registry
)
database_b = entry_reloaded.runtime_data.database
try:
assert await list_admin_ids(database_b) == ["user-parent"]
finally:
await database.close()
assert await async_unload_entry(hass, entry_reloaded) is True
async def test_setup_seed_prefers_options_over_entry_data(
hass, make_entry
) -> None:
"""Seed precedence on a fresh database: the options flow's saved
list (the current one) beats the config flow's original copy."""
from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
entry = await _setup_entry(
hass,
make_entry(
options={CONF_ADMIN_USER_IDS: ["user-current"]},
data={CONF_ADMIN_USER_IDS: ["ha-owner"]},
context={"user_id": "ha-owner"},
),
hass.registry,
)
database = entry.runtime_data.database
try:
assert await list_admin_ids(database) == ["user-current"]
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
async def test_setup_owner_fallback_seeds_owner_accounts(
hass, make_entry
) -> None:
"""Last-resort seed: with no persisted copy and no flow-context
user, HA owner accounts are seeded so first setup is never
ownerless (the real flow may lose the context; owners can already
do everything in HA)."""
from unittest.mock import AsyncMock
hass.auth.async_get_users = AsyncMock(
return_value=[
SimpleNamespace(id="owner-1", is_owner=True),
SimpleNamespace(id="member-1", is_owner=False),
SimpleNamespace(id="owner-2", is_owner=True),
]
)
entry = await _setup_entry(hass, make_entry(), hass.registry)
database = entry.runtime_data.database
try:
assert sorted(await list_admin_ids(database)) == ["owner-1", "owner-2"]
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
async def test_setup_owner_fallback_not_used_when_context_present(
hass, make_entry
) -> None:
"""The context user (flow caller) wins over the owner fallback:
a non-owner parent configuring the integration becomes the admin."""
from unittest.mock import AsyncMock
hass.auth.async_get_users = AsyncMock(
return_value=[SimpleNamespace(id="owner-1", is_owner=True)]
)
entry = await _setup_entry(
hass,
make_entry(context={"user_id": "parent-configured"}),
hass.registry,
)
database = entry.runtime_data.database
try:
assert await list_admin_ids(database) == ["parent-configured"]
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
def test_seed_owner_ids_argument(tmp_path) -> None:
async def _body(database):
# Malformed owner lists are rejected like any other candidate —
# while the allowlist is still empty, so the seed proceeds far
# enough to validate.
with pytest.raises(ValueError):
await seed_setup_admin(
database,
stored_admin_ids=None,
context_user_id=None,
owner_ids=[7],
)
assert await list_admin_ids(database) == []
seeded = await seed_setup_admin(
database,
stored_admin_ids=None,
context_user_id=None,
owner_ids=["owner-1"],
)
assert seeded == ["owner-1"]
return seeded
_with_db(tmp_path, "seed-owner-arg.db")(_body)
async def test_reopened_setup_failure_closes_database(
hass, make_entry, monkeypatch
) -> None:
"""A seeding failure on the reopened-record path closes the just-
opened connection before re-raising (no leaked handles across HA's
setup retries)."""
import pytest as pytest_module
from unittest.mock import AsyncMock
import custom_components.nestquest as nestquest_module
from custom_components.nestquest.db import NestQuestDatabase
# First: a successful setup to create the runtime record and file.
entry = await _setup_entry(hass, make_entry(), hass.registry)
await async_unload_entry(hass, entry)
# Now a re-setup whose seeding fails after the reopen.
async def _failing_seed(database, **kwargs):
raise RuntimeError("seed boom")
monkeypatch.setattr(
nestquest_module, "seed_setup_admin", _failing_seed, raising=True
)
close_calls: list[bool] = []
original_close = NestQuestDatabase.close
async def _spy_close(self):
close_calls.append(True)
return await original_close(self)
monkeypatch.setattr(NestQuestDatabase, "close", _spy_close)
with pytest_module.raises(RuntimeError, match="seed boom"):
await _setup_entry(hass, entry, hass.registry)
# The reopen path opened AND closed a connection; the failure is
# the last thing observed.
assert close_calls[-1] is True
tests/test_config_flow.py
+22 -1
View File
@@ -154,4 +154,25 @@ def test_no_hardcoded_domain_or_db_filename_in_new_modules() -> None:
)
assert node.value != "nestquest.db", (
f"Found hard-coded database filename in {py_file.name}:{node.lineno}"
)
)
async def test_flow_persists_flow_user_as_initial_admin(hass, make_flow) -> None:
"""The HA user completing the flow lands in entry data as the
initial admin copy: entry.context is not persisted by real HA, so
the durable record must carry the owner."""
from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
flow = make_flow(hass)
flow.context = {"user_id": "ha-owner"}
result = await flow.async_step_user({})
assert result["type"] == "create_entry"
assert result["data"] == {CONF_ADMIN_USER_IDS: ["ha-owner"]}
async def test_flow_without_context_user_keeps_data_untouched(
hass, make_flow
) -> None:
flow = make_flow(hass)
result = await flow.async_step_user({"horizon_days": 7})
assert result["type"] == "create_entry"
assert result["data"] == {"horizon_days": 7}
tests/test_options_flow.py
+265 -4
View File
@@ -14,6 +14,7 @@ from conftest import make_config_entry, make_hass, options_flow_base as _options
from custom_components.nestquest import config_flow
from custom_components.nestquest import options_flow
from custom_components.nestquest.const import (
CONF_ADMIN_USER_IDS,
CONF_DAY_ROLLOVER_TIME,
CONF_HORIZON_DAYS,
CONF_PANEL_IDLE_TIMEOUT,
@@ -46,13 +47,48 @@ def _make_entry(entry_id: str = "test_entry", options: dict | None = None, data:
return make_config_entry(entry_id=entry_id, options=options, data=data)
def _make_flow(entry) -> options_flow.NestQuestOptionsFlow:
"""Create an options flow instance bound to a stubbed config entry."""
def _make_flow(entry, *, users=None, database=None) -> options_flow.NestQuestOptionsFlow:
"""Create an options flow instance bound to a stubbed config entry.
``users`` populates a stubbed hass.auth for the admin picker;
``database`` wires the runtime record so the flow can read and
write the allowlist like real HA's options flow does.
"""
from unittest.mock import AsyncMock
flow = options_flow.NestQuestOptionsFlow(config_entry=entry)
flow.hass = SimpleNamespace()
flow.hass.auth = SimpleNamespace(
async_get_users=AsyncMock(return_value=users or [])
)
if database is not None:
from custom_components.nestquest.const import DOMAIN
flow.hass.data = {
DOMAIN: {entry.entry_id: SimpleNamespace(database=database)}
}
return flow
def _user(user_id: str, name: str):
return SimpleNamespace(id=user_id, name=name)
def _open_allowlist_db(tmp_path, name):
"""A real migrated temp database for the flow to read and write."""
from unittest.mock import AsyncMock, MagicMock
from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import apply_migrations
hass = MagicMock()
hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
database = NestQuestDatabase(hass)
_run(database.open(tmp_path / name))
_run(apply_migrations(database))
return database
def _schema_values(schema) -> dict:
"""Extract the per-field default values from a real voluptuous schema."""
return {
@@ -161,7 +197,12 @@ def test_options_flow_falls_back_to_entry_data() -> None:
def test_options_flow_schema_applies_stored_defaults() -> None:
"""A real voluptuous validate({}) fills every field from the stored defaults."""
"""A real voluptuous validate({}) fills every field from the stored defaults.
Without a runtime database (and no auth in the bare harness), the
admin picker's default is the empty selection and its choice set is
empty — the three legacy settings are unaffected.
"""
entry = _make_entry(
options={
CONF_HORIZON_DAYS: 21,
@@ -174,6 +215,7 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
CONF_HORIZON_DAYS: 21,
CONF_DAY_ROLLOVER_TIME: "05:45",
CONF_PANEL_IDLE_TIMEOUT: 120,
CONF_ADMIN_USER_IDS: [],
}
@@ -373,4 +415,223 @@ def test_options_flow_no_hardcoded_domain_or_db_filename() -> None:
)
assert node.value != "nestquest.db", (
f"Hard-coded database filename in {py_file.name}:{node.lineno}"
)
)
# ---------------------------------------------------------------------------
# Admin allowlist picker (Feature 03)
# ---------------------------------------------------------------------------
def test_options_flow_picker_lists_existing_users(tmp_path) -> None:
"""The picker offers EXISTING HA users (id -> name), never free text."""
import voluptuous as vol
database = _open_allowlist_db(tmp_path, "picker-users.db")
try:
from custom_components.nestquest.admin_allowlist import set_admin_ids
_run(set_admin_ids(database, ["u1"]))
entry = _make_entry()
flow = _make_flow(
entry,
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
database=database,
)
result = _run(flow.async_step_init(None))
schema = result["data_schema"]
marker = next(
m for m in schema.schema
if getattr(m, "schema", None) == CONF_ADMIN_USER_IDS
)
# Default is the CURRENT database allowlist.
values = _schema_values(schema)
assert values[CONF_ADMIN_USER_IDS] == ["u1"]
# Both users are selectable; an unknown id is rejected by the
# schema itself (this is what real HA submits go through).
validated = schema(
{
CONF_HORIZON_DAYS: 7,
CONF_DAY_ROLLOVER_TIME: "03:30",
CONF_PANEL_IDLE_TIMEOUT: 60,
CONF_ADMIN_USER_IDS: ["u1", "u2"],
}
)
assert validated[CONF_ADMIN_USER_IDS] == ["u1", "u2"]
with pytest.raises(vol.Invalid):
schema(
{
CONF_HORIZON_DAYS: 7,
CONF_DAY_ROLLOVER_TIME: "03:30",
CONF_PANEL_IDLE_TIMEOUT: 60,
CONF_ADMIN_USER_IDS: ["impostor"],
}
)
finally:
_run(database.close())
def test_options_flow_submit_applies_selection_to_database(tmp_path) -> None:
database = _open_allowlist_db(tmp_path, "picker-apply.db")
try:
from custom_components.nestquest.admin_allowlist import (
list_admin_ids,
set_admin_ids,
)
_run(set_admin_ids(database, ["u1"]))
entry = _make_entry()
flow = _make_flow(
entry,
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
database=database,
)
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_ADMIN_USER_IDS: ["u2"]})
)
assert result["type"] == "create_entry"
assert result["data"][CONF_ADMIN_USER_IDS] == ["u2"]
# The allowlist (and the entry's recovery copy) carry the pick.
assert await_gather(list_admin_ids(database)) == ["u2"]
finally:
_run(database.close())
def await_gather(coro):
return _run(coro)
def test_options_flow_empty_selection_refused(tmp_path) -> None:
"""Saving zero admins is refused: an empty allowlist fails closed."""
database = _open_allowlist_db(tmp_path, "picker-empty.db")
try:
from custom_components.nestquest.admin_allowlist import (
list_admin_ids,
set_admin_ids,
)
_run(set_admin_ids(database, ["u1"]))
entry = _make_entry()
flow = _make_flow(
entry,
users=[_user("u1", "Joshua")],
database=database,
)
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_ADMIN_USER_IDS: []})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_ADMIN_USER_IDS: "no_admins"}
assert await_gather(list_admin_ids(database)) == ["u1"]
finally:
_run(database.close())
def test_options_flow_unknown_selection_refused(tmp_path) -> None:
database = _open_allowlist_db(tmp_path, "picker-unknown.db")
try:
from custom_components.nestquest.admin_allowlist import (
list_admin_ids,
set_admin_ids,
)
_run(set_admin_ids(database, ["u1"]))
entry = _make_entry()
flow = _make_flow(
entry,
users=[_user("u1", "Joshua")],
database=database,
)
result = _run(
flow.async_step_init(
{**VALID_INPUT, CONF_ADMIN_USER_IDS: ["impostor"]}
)
)
assert result["type"] == "form"
assert result["errors"] == {CONF_ADMIN_USER_IDS: "invalid_admin"}
assert await_gather(list_admin_ids(database)) == ["u1"]
finally:
_run(database.close())
def test_options_flow_submit_without_admin_key_keeps_allowlist(tmp_path) -> None:
"""Raw callers omitting the key (pre-picker payloads) do not
touch the allowlist; the CURRENT allowlist is carried forward on
the options copy so no removed admin can be resurrected later."""
database = _open_allowlist_db(tmp_path, "picker-legacy.db")
try:
from custom_components.nestquest.admin_allowlist import (
list_admin_ids,
set_admin_ids,
)
_run(set_admin_ids(database, ["u1"]))
entry = _make_entry()
flow = _make_flow(
entry,
users=[_user("u1", "Joshua")],
database=database,
)
result = _run(flow.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
assert result["data"][CONF_ADMIN_USER_IDS] == ["u1"]
assert await_gather(list_admin_ids(database)) == ["u1"]
finally:
_run(database.close())
def test_options_flow_duplicate_selection_refused(tmp_path) -> None:
"""Duplicate ids surface as a field error, not an unhandled
ValueError from the business layer."""
database = _open_allowlist_db(tmp_path, "picker-dupe.db")
try:
from custom_components.nestquest.admin_allowlist import (
list_admin_ids,
set_admin_ids,
)
_run(set_admin_ids(database, ["u1"]))
entry = _make_entry()
flow = _make_flow(
entry,
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
database=database,
)
result = _run(
flow.async_step_init(
{**VALID_INPUT, CONF_ADMIN_USER_IDS: ["u2", "u2"]}
)
)
assert result["type"] == "form"
assert result["errors"] == {CONF_ADMIN_USER_IDS: "invalid_admin"}
assert await_gather(list_admin_ids(database)) == ["u1"]
finally:
_run(database.close())
def test_options_flow_omitted_picker_carries_allowlist_forward(
tmp_path,
) -> None:
"""An omitted picker field must not drop the allowlist copy from
the returned options: HA replaces entry.options wholesale, and a
later database loss must not resurrect removed admins from the
stale config-flow copy in entry.data."""
database = _open_allowlist_db(tmp_path, "picker-carry.db")
try:
from custom_components.nestquest.admin_allowlist import (
set_admin_ids,
)
_run(set_admin_ids(database, ["u2"]))
entry = _make_entry()
flow = _make_flow(
entry,
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
database=database,
)
result = _run(flow.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
# The current allowlist rides along on the legacy three-key path.
assert result["data"][CONF_ADMIN_USER_IDS] == ["u2"]
assert result["data"][CONF_HORIZON_DAYS] == VALID_INPUT[CONF_HORIZON_DAYS]
finally:
_run(database.close())