overseer
c5ed4250b6
The HA integration is now purely a client of the NestQuest API service (Feature 18): the API owns the SQLite file, the migrations, and every business rule, so keeping a second copy of that ownership inside the integration meant two sources of truth for household state, dead wiring (the startup backfill, the day-rollover materialization and sweep listeners, the corrupt-DB handling, the admin-allowlist seeding) and settings keys nothing consumes any more. Removed the integration's re-export/DB shim modules (db, store, migrations, schema, the five DAOs, completion, children, quest_definitions, admin_allowlist, presence, presence_management, materialize, recurrence, events, sweep); the bundled domain package at custom_components/nestquest/core/ is untouched and remains the single home of those engines. Setup now builds only the API client, the snapshot coordinator, and the SSE subscription, registers the one complete_quest proxy, and forwards the platforms. The options flow keeps only the keys the client itself consumes (API base URL + panel token, update interval, staleness, panel idle timeout); the horizon, rollover, notification and admin-allowlist settings moved out with the database (the API's settings store owns horizon/rollover, and the automations take their behaviour from blueprint inputs). The integration-plane tests now script the panel route's payloads and SSE frames instead of seeding a local database; the core-plane tests keep their coverage with imports pointed at core directly.
269 lines
9.2 KiB
Python
269 lines
9.2 KiB
Python
"""Tests for the Feature 03 admin allowlist business layer (core.admin_allowlist)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from custom_components.nestquest.core.admin_allowlist import (
|
|
list_admin_ids,
|
|
seed_setup_admin,
|
|
set_admin_ids,
|
|
)
|
|
from custom_components.nestquest.core.db import NestQuestDatabase
|
|
from custom_components.nestquest.core.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().async_add_executor_job)
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
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)
|
|
|
|
|
|
# is_admin: the fail-closed resolver (Feature 09's verdict function)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_is_admin_verdict_matrix(tmp_path) -> None:
|
|
async def _body(database):
|
|
from custom_components.nestquest.core.admin_allowlist import is_admin
|
|
|
|
await set_admin_ids(database, ["user-owner"])
|
|
assert await is_admin(database, "user-owner") is True
|
|
assert await is_admin(database, "user-other") is False
|
|
# Unknown, malformed, and absent identities all fail closed.
|
|
assert await is_admin(database, "") is False
|
|
assert await is_admin(database, " user-owner ") is False, (
|
|
"ids are stored trimmed; an untrimmed lookup is a different "
|
|
"id and must not match"
|
|
)
|
|
assert await is_admin(database, None) is False
|
|
assert await is_admin(database, 42) is False
|
|
assert await is_admin(database, ["user-owner"]) is False
|
|
assert await is_admin(database, True) is False
|
|
return None
|
|
|
|
_with_db(tmp_path, "resolver-matrix.db")(_body)
|
|
|
|
|
|
def test_is_admin_empty_allowlist_fails_closed(tmp_path) -> None:
|
|
"""An empty allowlist returns False for EVERYONE, not True."""
|
|
async def _body(database):
|
|
from custom_components.nestquest.core.admin_allowlist import is_admin
|
|
|
|
for candidate in ("user-owner", "", None, 1):
|
|
assert await is_admin(database, candidate) is False
|
|
return None
|
|
|
|
_with_db(tmp_path, "resolver-empty.db")(_body)
|
|
|
|
|
|
def test_is_admin_false_after_last_admin_removal_attempt(tmp_path) -> None:
|
|
"""The refusal keeps the verdict sound: a refused removal leaves
|
|
the admin in place, so is_admin still resolves True."""
|
|
async def _body(database):
|
|
from custom_components.nestquest.core.admin_allowlist import is_admin
|
|
|
|
await set_admin_ids(database, ["user-owner"])
|
|
with pytest.raises(ValueError, match="at least one admin"):
|
|
await set_admin_ids(database, [])
|
|
assert await is_admin(database, "user-owner") is True
|
|
return None
|
|
|
|
_with_db(tmp_path, "resolver-refusal.db")(_body)
|