Joshua King
f00ec03c78
- AdminUsersDao.add: single INSERT ON CONFLICT DO NOTHING (rowcount-derived result), race-safe; concurrent-add test with 5 racing coroutines - ChildrenDao.reorder: rejects duplicate ids (ValueError); per-instance asyncio lock serializes concurrent reorders (the wrapper rejects concurrent transactions); rollback test now forces a real finalize-batch failure and asserts original order survives - SQL-leak guard: scans package + tests with FROM/INTO/UPDATE/DELETE FROM/ JOIN pattern; narrowly justified exclusions (schema DDL, migrations, their tests); this test file verified to contain no direct table SQL
521 lines
17 KiB
Python
521 lines
17 KiB
Python
"""Tests for dao_children.py: typed DAO for children and admin_users."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from custom_components.nestquest.dao_children import (
|
|
AdminUserRecord,
|
|
AdminUsersDao,
|
|
ChildRecord,
|
|
ChildrenDao,
|
|
)
|
|
from custom_components.nestquest.db import NestQuestDatabase
|
|
from custom_components.nestquest.schema import SCHEMA_V1_STATEMENTS
|
|
from custom_components.nestquest.migrations import apply_migrations
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.new_event_loop().run_until_complete(coro)
|
|
|
|
|
|
def _make_hass_mock():
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
hass = MagicMock()
|
|
hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
|
|
return hass
|
|
|
|
|
|
def _open_db(path) -> NestQuestDatabase:
|
|
database = NestQuestDatabase(_make_hass_mock())
|
|
_run(database.open(path))
|
|
_run(apply_migrations(database))
|
|
return database
|
|
|
|
|
|
def _daos(database) -> tuple[ChildrenDao, AdminUsersDao]:
|
|
return ChildrenDao(database), AdminUsersDao(database)
|
|
|
|
|
|
NOW = "2026-09-14T11:00:00+00:00"
|
|
|
|
|
|
def _make_child(dao, name="Ada", **kwargs) -> ChildRecord:
|
|
return _run(dao.create(name, NOW, **kwargs))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# children: create and get
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_returns_typed_record(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "create.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
child = _make_child(dao, colour="#ff0000", avatar_ref="avatars/1")
|
|
assert isinstance(child, ChildRecord)
|
|
assert child.id == 1
|
|
assert child.display_name == "Ada"
|
|
assert child.colour == "#ff0000"
|
|
assert child.avatar_ref == "avatars/1"
|
|
assert child.sort_order == 0
|
|
assert child.is_active is True
|
|
assert child.created_at == NOW
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_get_returns_none_for_missing_child(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "get-missing.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
assert _run(dao.get(999)) is None
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_create_duplicate_name_is_allowed(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "dup-name.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
first = _make_child(dao, "Ada")
|
|
second = _make_child(dao, "Ada")
|
|
assert first.id != second.id
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# children: list active / list all
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_active_excludes_inactive_and_sorts(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "list-active.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
_make_child(dao, "Bo", sort_order=2)
|
|
ada = _make_child(dao, "Ada", sort_order=1)
|
|
cy = _make_child(dao, "Cy", sort_order=1)
|
|
_make_child(dao, "Dee", sort_order=0)
|
|
_run(dao.set_active(ada.id, False))
|
|
active = _run(dao.list_active())
|
|
assert [c.display_name for c in active] == ["Dee", "Cy", "Bo"]
|
|
assert all(c.is_active for c in active)
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_list_all_includes_inactive(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "list-all.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
ada = _make_child(dao, "Ada", sort_order=1)
|
|
_make_child(dao, "Bo", sort_order=0)
|
|
_run(dao.set_active(ada.id, False))
|
|
all_children = _run(dao.list_all())
|
|
assert [c.display_name for c in all_children] == ["Bo", "Ada"]
|
|
assert all_children[1].is_active is False
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# children: update, set_active, reorder
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_update_fields_round_trip(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "update.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
child = _make_child(dao, "Ada", colour="#111111")
|
|
updated = _run(
|
|
dao.update(
|
|
child.id,
|
|
display_name="Ada L.",
|
|
colour="#222222",
|
|
avatar_ref="avatars/x.png",
|
|
)
|
|
)
|
|
assert updated == 1
|
|
fetched = _run(dao.get(child.id))
|
|
assert fetched is not None
|
|
assert fetched.display_name == "Ada L."
|
|
assert fetched.colour == "#222222"
|
|
assert fetched.avatar_ref == "avatars/x.png"
|
|
assert fetched.sort_order == 0
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_update_omitted_fields_untouched(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "update-partial.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
child = _make_child(dao, "Ada", colour="#111111")
|
|
_run(dao.update(child.id, display_name="Ada L."))
|
|
fetched = _run(dao.get(child.id))
|
|
assert fetched.colour == "#111111"
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_update_no_fields_returns_zero_and_changes_nothing(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "update-empty.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
child = _make_child(dao, "Ada")
|
|
assert _run(dao.update(child.id)) == 0
|
|
fetched = _run(dao.get(child.id))
|
|
assert fetched is not None and fetched.display_name == "Ada"
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_update_missing_child_returns_zero(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "update-missing.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
assert _run(dao.update(999, display_name="X")) == 0
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_set_active_missing_child_returns_zero(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "set-active-missing.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
assert _run(dao.set_active(999, False)) == 0
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_reorder_assigns_positions_in_order(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "reorder.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
a = _make_child(dao, "Ada")
|
|
b = _make_child(dao, "Bo")
|
|
c = _make_child(dao, "Cy")
|
|
_run(dao.reorder([c.id, a.id, b.id]))
|
|
children = _run(dao.list_all())
|
|
assert [x.display_name for x in children] == ["Cy", "Ada", "Bo"]
|
|
assert [x.sort_order for x in children] == [0, 1, 2]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_reorder_is_transactional_and_idempotent(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "reorder-idempotent.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
a = _make_child(dao, "Ada")
|
|
b = _make_child(dao, "Bo")
|
|
_run(dao.reorder([b.id, a.id]))
|
|
_run(dao.reorder([b.id, a.id]))
|
|
children = _run(dao.list_all())
|
|
assert [(x.display_name, x.sort_order) for x in children] == [
|
|
("Bo", 0),
|
|
("Ada", 1),
|
|
]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_reorder_empty_list_is_noop(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "reorder-empty.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
_make_child(dao, "Ada")
|
|
_run(dao.reorder([]))
|
|
children = _run(dao.list_all())
|
|
assert children[0].sort_order == 0
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_reorder_unknown_id_leaves_it_untouched(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "reorder-unknown.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
a = _make_child(dao, "Ada")
|
|
_run(dao.reorder([a.id, 999]))
|
|
children = _run(dao.list_all())
|
|
assert children[0].sort_order == 0
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_reorder_rolls_back_to_original_order_on_failure(tmp_path) -> None:
|
|
"""A failing finalize batch rolls the whole reorder back.
|
|
|
|
The two-phase staging is only meaningful if a crash between the
|
|
batches leaves the original order intact: monkeypatch execute_many
|
|
so the finalize (non-negative) batch raises, then assert every
|
|
sort_order is unchanged.
|
|
"""
|
|
database = _open_db(tmp_path / "reorder-rollback.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
a = _make_child(dao, "Ada", sort_order=0)
|
|
b = _make_child(dao, "Bo", sort_order=1)
|
|
|
|
original_execute_many = database.execute_many
|
|
|
|
async def _failing_execute_many(sql, parameters):
|
|
# The staging batch writes negative positions; the finalize
|
|
# batch writes 0..n-1. Fail the finalize batch.
|
|
if all(position >= 0 for position, _ in parameters):
|
|
raise sqlite3.OperationalError("boom")
|
|
return await original_execute_many(sql, parameters)
|
|
|
|
database.execute_many = _failing_execute_many
|
|
with pytest.raises(sqlite3.OperationalError, match="boom"):
|
|
_run(dao.reorder([b.id, a.id]))
|
|
children = _run(dao.list_all())
|
|
assert [(c.display_name, c.sort_order) for c in children] == [
|
|
("Ada", 0),
|
|
("Bo", 1),
|
|
], "failed reorder must leave original sort orders intact"
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_reorder_rejects_duplicate_ids(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "reorder-duplicate.db")
|
|
try:
|
|
dao, _ = _daos(database)
|
|
a = _make_child(dao, "Ada")
|
|
with pytest.raises(ValueError, match="duplicate"):
|
|
_run(dao.reorder([a.id, a.id]))
|
|
fetched = _run(dao.get(a.id))
|
|
assert fetched is not None and fetched.sort_order == 0
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_reorder_concurrent_calls_serialize(tmp_path) -> None:
|
|
"""Two concurrent reorders queue on the transaction, both succeed.
|
|
|
|
The second call sees the first call's final positions as its
|
|
starting state; both end with a consistent 0..n-1 assignment.
|
|
"""
|
|
async def _main() -> None:
|
|
database = NestQuestDatabase(_make_hass_mock())
|
|
await database.open(tmp_path / "reorder-concurrent.db")
|
|
try:
|
|
await apply_migrations(database)
|
|
dao, _ = _daos(database)
|
|
a = await dao.create("Ada", NOW)
|
|
b = await dao.create("Bo", NOW)
|
|
task_1 = asyncio.ensure_future(dao.reorder([b.id, a.id]))
|
|
await asyncio.sleep(0)
|
|
task_2 = asyncio.ensure_future(dao.reorder([a.id, b.id]))
|
|
await asyncio.gather(task_1, task_2)
|
|
children = await dao.list_all()
|
|
orders = {c.display_name: c.sort_order for c in children}
|
|
assert sorted(orders.values()) == [0, 1]
|
|
finally:
|
|
await database.close()
|
|
|
|
_run(_main())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# admin_users: add, remove, list, exists
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_admin_add_then_exists(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "admin-add.db")
|
|
try:
|
|
_, dao = _daos(database)
|
|
assert _run(dao.add("user-1", NOW)) is True
|
|
assert _run(dao.exists("user-1")) is True
|
|
assert _run(dao.exists("user-2")) is False
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_admin_add_is_idempotent(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "admin-idempotent.db")
|
|
try:
|
|
_, dao = _daos(database)
|
|
assert _run(dao.add("user-1", NOW)) is True
|
|
assert _run(dao.add("user-1", NOW)) is False
|
|
records = _run(dao.list())
|
|
assert len(records) == 1
|
|
assert records[0].added_at == NOW
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_admin_add_concurrent_calls_are_idempotent(tmp_path) -> None:
|
|
"""Concurrent add() calls of the same user: one inserts, rest False.
|
|
|
|
The single-statement ON CONFLICT DO NOTHING makes the decision and
|
|
the write atomic per statement, so racing coroutines cannot both
|
|
'see no row' and one of them raise IntegrityError.
|
|
"""
|
|
async def _main() -> None:
|
|
database = NestQuestDatabase(_make_hass_mock())
|
|
await database.open(tmp_path / "admin-race.db")
|
|
try:
|
|
await apply_migrations(database)
|
|
_, dao = _daos(database)
|
|
tasks = [
|
|
asyncio.ensure_future(dao.add("user-1", NOW))
|
|
for _ in range(5)
|
|
]
|
|
await asyncio.sleep(0)
|
|
results = list(await asyncio.gather(*tasks))
|
|
assert sorted(results) == [False, False, False, False, True]
|
|
records = await dao.list()
|
|
assert len(records) == 1
|
|
assert records[0].added_at == NOW
|
|
finally:
|
|
await database.close()
|
|
|
|
_run(_main())
|
|
|
|
|
|
def test_admin_add_preserves_original_added_at(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "admin-added-at.db")
|
|
try:
|
|
_, dao = _daos(database)
|
|
_run(dao.add("user-1", NOW))
|
|
_run(dao.add("user-1", "2026-09-15T00:00:00+00:00"))
|
|
records = _run(dao.list())
|
|
assert records[0].added_at == NOW
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_admin_remove_returns_true_then_false(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "admin-remove.db")
|
|
try:
|
|
_, dao = _daos(database)
|
|
_run(dao.add("user-1", NOW))
|
|
assert _run(dao.remove("user-1")) is True
|
|
assert _run(dao.remove("user-1")) is False
|
|
assert _run(dao.exists("user-1")) is False
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_admin_list_returns_typed_records_ordered(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "admin-list.db")
|
|
try:
|
|
_, dao = _daos(database)
|
|
_run(dao.add("user-2", "2026-09-14T10:00:00+00:00"))
|
|
_run(dao.add("user-1", "2026-09-14T09:00:00+00:00"))
|
|
records = _run(dao.list())
|
|
assert isinstance(records[0], AdminUserRecord)
|
|
assert [r.ha_user_id for r in records] == ["user-1", "user-2"]
|
|
assert records[0].added_at == "2026-09-14T09:00:00+00:00"
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_admin_exists_empty_allowlist_fails_closed(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "admin-empty.db")
|
|
try:
|
|
_, dao = _daos(database)
|
|
assert _run(dao.exists("user-1")) is False
|
|
assert _run(dao.list()) == []
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_admin_records_are_typed_not_raw_rows(tmp_path) -> None:
|
|
database = _open_db(tmp_path / "admin-typed.db")
|
|
try:
|
|
_, dao = _daos(database)
|
|
_run(dao.add("user-1", NOW))
|
|
records = _run(dao.list())
|
|
assert records == [AdminUserRecord("user-1", NOW)]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# No SQL for these tables outside the DAO module
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
|
|
"""Guardrail: children/admin_users SQL may appear only in the DAO
|
|
(and the schema DDL declarations). Scans the integration package
|
|
AND the repo's test files, matching FROM/INTO/UPDATE/DELETE FROM/
|
|
JOIN against either table — including DELETE FROM children, the
|
|
hard-delete path the guardrail forbids.
|
|
|
|
Exclusions are narrowly justified: schema.py declares the tables'
|
|
DDL; migrations.py applies that DDL; test_schema.py and
|
|
test_migrations.py test exactly those DDL/migration layers. None
|
|
of them touch the tables' data paths — that is the DAO's job.
|
|
"""
|
|
import re
|
|
from pathlib import Path
|
|
|
|
package = Path(
|
|
__import__(
|
|
"custom_components.nestquest", fromlist=["__file__"]
|
|
).__file__
|
|
).parent
|
|
repo_root = package.parent.parent
|
|
scan_roots = [package, repo_root / "tests"]
|
|
|
|
allowed = {
|
|
"dao_children.py", # the DAO itself
|
|
"schema.py", # declares the DDL
|
|
"migrations.py", # applies the DDL
|
|
"test_schema.py", # tests the DDL
|
|
"test_migrations.py", # tests migration application of the DDL
|
|
"test_dao_children.py", # this file, scanned separately below
|
|
}
|
|
sql_pattern = re.compile(
|
|
r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+"
|
|
r'["\`]*(children|admin_users)["\`]*\b',
|
|
re.IGNORECASE,
|
|
)
|
|
offenders: list[str] = []
|
|
for root in scan_roots:
|
|
for py in sorted(root.glob("*.py")):
|
|
if py.name in allowed:
|
|
continue
|
|
if sql_pattern.search(py.read_text()):
|
|
offenders.append(py.name)
|
|
assert offenders == [], (
|
|
f"SQL touching children/admin_users leaked into: {offenders}"
|
|
)
|
|
|
|
|
|
def test_dao_test_file_uses_dao_not_raw_table_sql() -> None:
|
|
"""This test module must exercise the DAO, not raw SQL, for these
|
|
tables — except inside this guard itself. Any SQL statement naming
|
|
children or admin_users in a string literal elsewhere here is a
|
|
violation.
|
|
"""
|
|
import re
|
|
from pathlib import Path
|
|
|
|
text = Path(__file__).read_text()
|
|
# Cut everything from the guard function to the end of file: the
|
|
# guard's own patterns mention the keywords in regex strings, not
|
|
# executable SQL.
|
|
head = text[: text.index("def test_children_and_admin_sql_lives")]
|
|
sql_pattern = re.compile(
|
|
r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM)\s+"
|
|
r'["\`]*(children|admin_users)\b',
|
|
re.IGNORECASE,
|
|
)
|
|
assert sql_pattern.search(head) is None, (
|
|
"tests must go through the DAO, not raw SQL, for these tables"
|
|
)
|