Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d09ac91ade | ||
|
|
e7daecf970 |
@@ -907,6 +907,66 @@ class QuestDefinitionsDao:
|
||||
)
|
||||
return [_definition_from_row(row) for row in rows]
|
||||
|
||||
async def _snapshots_for(
|
||||
self, definitions: list[QuestDefinitionRecord]
|
||||
) -> list[QuestDefinitionSnapshot]:
|
||||
"""Read each definition's rule, assignees and windows.
|
||||
|
||||
MUST be called inside the connection lock (and ideally a
|
||||
transaction) so the definitions handed in and the rows this
|
||||
reads back cannot be raced by a concurrent edit between list
|
||||
and bundle. Each rule is read through
|
||||
:class:`ScheduleRulesDao` on the same connection, so the
|
||||
snapshot's rule record is the one the definition references AT
|
||||
the moment of the locked read.
|
||||
"""
|
||||
rules_dao = ScheduleRulesDao(self._database)
|
||||
snapshots: list[QuestDefinitionSnapshot] = []
|
||||
for definition in definitions:
|
||||
rule_record = await rules_dao.get(definition.schedule_rule_id)
|
||||
assert rule_record is not None
|
||||
snapshots.append(
|
||||
QuestDefinitionSnapshot(
|
||||
definition=definition,
|
||||
rule=rule_record,
|
||||
assignees=await self.list_assignees(definition.id),
|
||||
windows=await self.list_windows(definition.id),
|
||||
)
|
||||
)
|
||||
return snapshots
|
||||
|
||||
async def list_snapshots_active(self) -> list[QuestDefinitionSnapshot]:
|
||||
"""Return every active definition with its rule, assignees and
|
||||
windows, all read inside one locked transaction.
|
||||
|
||||
The definitive list and each definition's linked rows are read
|
||||
in a single BEGIN..COMMIT span under the connection lock, so a
|
||||
concurrent atomic edit cannot interleave between the definition
|
||||
list and the bundle read: the caller receives a coherent
|
||||
snapshot (each definition's rule, assignees and windows as they
|
||||
were at one instant), never a mixed old-rule/new-windows state.
|
||||
"""
|
||||
async with _connection_lock(self._database):
|
||||
async with self._database.transaction():
|
||||
definitions = await self.list_active()
|
||||
return await self._snapshots_for(definitions)
|
||||
|
||||
async def list_snapshots_by_child(
|
||||
self, child_id: int
|
||||
) -> list[QuestDefinitionSnapshot]:
|
||||
"""Return each definition assigned to ``child_id`` with its rule,
|
||||
assignees and windows, all read inside one locked transaction.
|
||||
|
||||
Mirrors :meth:`list_snapshots_active` but selects by assignment
|
||||
(``list_by_child`` order: newest first). Locks read-and-bundle
|
||||
into one transaction so the bundle never mixes one edit's
|
||||
definition with another's windows.
|
||||
"""
|
||||
async with _connection_lock(self._database):
|
||||
async with self._database.transaction():
|
||||
definitions = await self.list_by_child(child_id)
|
||||
return await self._snapshots_for(definitions)
|
||||
|
||||
async def update(
|
||||
self,
|
||||
definition_id: int,
|
||||
|
||||
@@ -44,6 +44,7 @@ from dataclasses import dataclass
|
||||
from .dao_children import ChildRecord, _connection_lock
|
||||
from .dao_rules import (
|
||||
QuestDefinitionRecord,
|
||||
QuestDefinitionSnapshot,
|
||||
QuestDefinitionWindowRecord,
|
||||
QuestDefinitionsDao,
|
||||
ScheduleRuleStorage,
|
||||
@@ -318,14 +319,23 @@ async def edit_quest_definition(
|
||||
window_specs = _normalize_windows(windows)
|
||||
|
||||
dao = QuestDefinitionsDao(database)
|
||||
snapshot = await dao.edit_definition(
|
||||
definition_id,
|
||||
title=title_value,
|
||||
description=description_value,
|
||||
icon=icon_value,
|
||||
rule=rule_storage,
|
||||
windows=window_specs,
|
||||
)
|
||||
try:
|
||||
snapshot = await dao.edit_definition(
|
||||
definition_id,
|
||||
title=title_value,
|
||||
description=description_value,
|
||||
icon=icon_value,
|
||||
rule=rule_storage,
|
||||
windows=window_specs,
|
||||
)
|
||||
except ValueError as error:
|
||||
# The DAO's missing-definition error names only the id; re-raise
|
||||
# so the public contract names the field the caller passed (matching
|
||||
# assign/unassign/deactivate).
|
||||
message = str(error)
|
||||
if message.startswith("quest definition "):
|
||||
raise ValueError(f"definition_id: {message}") from error
|
||||
raise
|
||||
|
||||
decoded_rule = schedule_rule_from_storage(
|
||||
schedule_rule_storage_from_record(snapshot.rule)
|
||||
@@ -455,33 +465,25 @@ async def set_quest_definition_active(
|
||||
)
|
||||
|
||||
|
||||
async def _bundle_definition(
|
||||
database: NestQuestDatabase,
|
||||
dao: QuestDefinitionsDao,
|
||||
definition: QuestDefinitionRecord,
|
||||
def _bundle_snapshot(
|
||||
snapshot: QuestDefinitionSnapshot,
|
||||
) -> CreatedQuestDefinition:
|
||||
"""Assemble a definition's rich view: decoded rule + assignees + windows.
|
||||
"""Decode a DAO snapshot into the layer's rich view.
|
||||
|
||||
Reads the schedule rule through :class:`~.dao_rules.ScheduleRulesDao`
|
||||
and the assignees and windows through the DAO, then wraps them in the
|
||||
same :class:`CreatedQuestDefinition` shape the create/edit/activate
|
||||
paths return, so every query helper here reuses one snapshot type
|
||||
rather than inventing a second one.
|
||||
The rule record carried by the snapshot is the one the definition
|
||||
referenced at read time, so decoding it here (rather than re-reading
|
||||
through :class:`~.dao_rules.ScheduleRulesDao`) preserves the
|
||||
coherence the DAO's locked snapshot guarantees: the rule, assignees
|
||||
and windows all come from one atomic read, never three separate ones.
|
||||
"""
|
||||
rule_record = await ScheduleRulesDao(database).get(
|
||||
definition.schedule_rule_id
|
||||
)
|
||||
assert rule_record is not None
|
||||
rule = schedule_rule_from_storage(
|
||||
schedule_rule_storage_from_record(rule_record)
|
||||
schedule_rule_storage_from_record(snapshot.rule)
|
||||
)
|
||||
assignees = await dao.list_assignees(definition.id)
|
||||
windows = await dao.list_windows(definition.id)
|
||||
return CreatedQuestDefinition(
|
||||
definition=definition,
|
||||
definition=snapshot.definition,
|
||||
rule=rule,
|
||||
assignees=assignees,
|
||||
windows=windows,
|
||||
assignees=snapshot.assignees,
|
||||
windows=snapshot.windows,
|
||||
)
|
||||
|
||||
|
||||
@@ -494,14 +496,14 @@ async def list_active_definitions(
|
||||
order, so callers get a stable, repeatable sequence. Each entry is
|
||||
the same rich :class:`CreatedQuestDefinition` snapshot as the rest of
|
||||
the layer: the stored row plus its decoded
|
||||
:class:`~.recurrence.ScheduleRule`, assignees and windows.
|
||||
:class:`~.recurrence.ScheduleRule`, assignees and windows. The
|
||||
definitions and their links are read by the DAO inside one locked
|
||||
transaction, so each bundle is a coherent snapshot that a concurrent
|
||||
edit cannot interleave.
|
||||
"""
|
||||
dao = QuestDefinitionsDao(database)
|
||||
definitions = await dao.list_active()
|
||||
return [
|
||||
await _bundle_definition(database, dao, definition)
|
||||
for definition in definitions
|
||||
]
|
||||
snapshots = await dao.list_snapshots_active()
|
||||
return [_bundle_snapshot(snapshot) for snapshot in snapshots]
|
||||
|
||||
|
||||
async def list_definitions_for_child(
|
||||
@@ -515,15 +517,12 @@ async def list_definitions_for_child(
|
||||
rising definition id, matching ``list_active_definitions`` and
|
||||
``list_definitions_firing_on``. Each entry is a
|
||||
:class:`CreatedQuestDefinition` snapshot bundling the definition with
|
||||
its decoded rule, assignees and windows.
|
||||
its decoded rule, assignees and windows, read atomically by the DAO.
|
||||
"""
|
||||
_validate_child_id(child_id)
|
||||
dao = QuestDefinitionsDao(database)
|
||||
definitions = await dao.list_by_child(child_id)
|
||||
bundles = [
|
||||
await _bundle_definition(database, dao, definition)
|
||||
for definition in definitions
|
||||
]
|
||||
snapshots = await dao.list_snapshots_by_child(child_id)
|
||||
bundles = [_bundle_snapshot(snapshot) for snapshot in snapshots]
|
||||
return sorted(bundles, key=lambda bundle: bundle.definition.id)
|
||||
|
||||
|
||||
@@ -540,13 +539,13 @@ async def list_definitions_firing_on(
|
||||
definitions whose decoded rule fires on the date are returned, in
|
||||
rising definition-id order, each as the same rich
|
||||
:class:`CreatedQuestDefinition` snapshot as the rest of the layer.
|
||||
The definitions and their rules are read inside one locked
|
||||
transaction, so a bundle can never filter on one rule while
|
||||
presenting another state's windows.
|
||||
"""
|
||||
_validate_date(date, "date")
|
||||
target = datetime.datetime.strptime(date, "%Y-%m-%d").date()
|
||||
dao = QuestDefinitionsDao(database)
|
||||
definitions = await dao.list_active()
|
||||
bundles = [
|
||||
await _bundle_definition(database, dao, definition)
|
||||
for definition in definitions
|
||||
]
|
||||
snapshots = await dao.list_snapshots_active()
|
||||
bundles = [_bundle_snapshot(snapshot) for snapshot in snapshots]
|
||||
return [bundle for bundle in bundles if occurs_on(bundle.rule, target)]
|
||||
|
||||
@@ -786,6 +786,24 @@ def test_edit_rejects_nonexistent_definition(tmp_path) -> None:
|
||||
_with_db(tmp_path, "edit-missing-definition.db")(_body)
|
||||
|
||||
|
||||
def test_edit_nonexistent_definition_names_field(tmp_path) -> None:
|
||||
"""A missing definition's error names the ``definition_id`` field.
|
||||
|
||||
The DAO raises a bare "quest definition <id> does not exist"; the
|
||||
business layer must re-raise it prefixed with the field the caller
|
||||
passed, matching assign/unassign/deactivate.
|
||||
"""
|
||||
async def _body(database, children, rules, definitions, child):
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match="definition_id: quest definition 999 does not exist",
|
||||
):
|
||||
await edit_quest_definition(database, 999, title="X")
|
||||
return None
|
||||
|
||||
_with_db(tmp_path, "edit-missing-definition-field.db")(_body)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", [True, 1.5, "1", None])
|
||||
def test_edit_rejects_non_int_definition_id(tmp_path, bad) -> None:
|
||||
async def _body(database, children, rules, definitions, child):
|
||||
@@ -1623,6 +1641,201 @@ def test_list_definitions_firing_on_rejects_malformed_date(tmp_path, bad) -> Non
|
||||
_with_db(tmp_path, "list-firing-on-bad-date.db")(_body)
|
||||
|
||||
|
||||
def test_list_query_snapshot_coherent_under_racing_edit(tmp_path) -> None:
|
||||
"""A racing edit cannot mix states into a query helper's snapshot.
|
||||
|
||||
The query helper pauses mid-snapshot (inside its locked transaction)
|
||||
while an edit that changes BOTH the rule and the windows is queued
|
||||
behind the connection lock. Because the helper reads definitions,
|
||||
rules and windows in one locked transaction, its returned bundle is
|
||||
either the whole old state or the whole new state — never the old
|
||||
rule paired with the new windows (the exact interleaving the unlocked
|
||||
reads used to allow). The pause hook holds
|
||||
``list_windows`` mid-flight so the edit is deterministically queued
|
||||
against the helper's lock.
|
||||
"""
|
||||
async def _body(database, children, rules, definitions, child):
|
||||
created = await create_quest_definition(
|
||||
database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
|
||||
)
|
||||
weekly = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY,
|
||||
weekday_set={2},
|
||||
start_date="2026-09-14",
|
||||
)
|
||||
|
||||
import custom_components.nestquest.dao_rules as dao_rules
|
||||
|
||||
original_list = dao_rules.QuestDefinitionsDao.list_windows
|
||||
started = asyncio.Event()
|
||||
release = asyncio.Event()
|
||||
|
||||
async def _pausing_list(self, definition_id):
|
||||
if not started.is_set():
|
||||
started.set()
|
||||
await release.wait()
|
||||
return await original_list(self, definition_id)
|
||||
|
||||
dao_rules.QuestDefinitionsDao.list_windows = _pausing_list
|
||||
try:
|
||||
read_task = asyncio.ensure_future(
|
||||
list_definitions_firing_on(database, "2026-09-16")
|
||||
)
|
||||
await started.wait()
|
||||
edit_task = asyncio.ensure_future(
|
||||
edit_quest_definition(
|
||||
database,
|
||||
created.definition.id,
|
||||
rule=weekly,
|
||||
windows=["evening"],
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(0)
|
||||
release.set()
|
||||
result, edited = await asyncio.gather(read_task, edit_task)
|
||||
finally:
|
||||
dao_rules.QuestDefinitionsDao.list_windows = original_list
|
||||
|
||||
# The helper finished before the queued edit, so its snapshot is
|
||||
# the whole OLD state: the daily rule with its morning window.
|
||||
assert [b.definition.id for b in result] == [created.definition.id]
|
||||
assert result[0].rule == _daily_rule()
|
||||
assert [w.window for w in result[0].windows] == ["morning"]
|
||||
# The edit then reported the whole NEW state, not a mix either.
|
||||
assert edited.rule == weekly
|
||||
assert [w.window for w in edited.windows] == ["evening"]
|
||||
return result
|
||||
|
||||
_with_db(tmp_path, "list-snapshot-concurrent.db")(_body)
|
||||
|
||||
|
||||
def test_list_definitions_for_child_snapshot_coherent_under_racing_unassign(
|
||||
tmp_path,
|
||||
) -> None:
|
||||
"""A racing unassign cannot strip the child from the child-scoped
|
||||
snapshot that ``list_definitions_for_child`` returns.
|
||||
|
||||
The helper pauses mid-snapshot (inside its locked transaction) while
|
||||
an unassign of the very child it is listing is queued behind the
|
||||
connection lock. Because the definition set and its assignees are
|
||||
read in one locked transaction, the returned bundle still shows the
|
||||
child assigned — the coherent pre-unassign snapshot — rather than the
|
||||
mixed state (definition returned "for child" but with the child
|
||||
already gone from its roster) that the unlocked reads used to allow.
|
||||
The pause hook holds ``list_assignees`` mid-flight so the unassign is
|
||||
deterministically queued against the helper's lock.
|
||||
"""
|
||||
async def _body(database, children, rules, definitions, child):
|
||||
created = await create_quest_definition(
|
||||
database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
|
||||
)
|
||||
|
||||
import custom_components.nestquest.dao_rules as dao_rules
|
||||
|
||||
original_list = dao_rules.QuestDefinitionsDao.list_assignees
|
||||
started = asyncio.Event()
|
||||
release = asyncio.Event()
|
||||
|
||||
async def _pausing_list(self, definition_id):
|
||||
if not started.is_set():
|
||||
started.set()
|
||||
await release.wait()
|
||||
return await original_list(self, definition_id)
|
||||
|
||||
dao_rules.QuestDefinitionsDao.list_assignees = _pausing_list
|
||||
try:
|
||||
read_task = asyncio.ensure_future(
|
||||
list_definitions_for_child(database, child.id)
|
||||
)
|
||||
await started.wait()
|
||||
unassign_task = asyncio.ensure_future(
|
||||
unassign_child(database, created.definition.id, child.id)
|
||||
)
|
||||
for _ in range(5):
|
||||
await asyncio.sleep(0)
|
||||
release.set()
|
||||
result, removed = await asyncio.gather(read_task, unassign_task)
|
||||
finally:
|
||||
dao_rules.QuestDefinitionsDao.list_assignees = original_list
|
||||
|
||||
# The helper finished before the queued unassign, so its snapshot
|
||||
# is the whole OLD state: the child still assigned to the
|
||||
# definition it returned for that child.
|
||||
assert [b.definition.id for b in result] == [created.definition.id]
|
||||
assert [c.id for c in result[0].assignees] == [child.id]
|
||||
assert removed is True
|
||||
return result
|
||||
|
||||
_with_db(tmp_path, "list-for-child-snapshot-concurrent.db")(_body)
|
||||
|
||||
|
||||
def test_list_active_definitions_snapshot_coherent_under_racing_edit(
|
||||
tmp_path,
|
||||
) -> None:
|
||||
"""A racing edit cannot mix states into ``list_active_definitions``.
|
||||
|
||||
The helper pauses mid-snapshot (inside its locked transaction) while
|
||||
an edit that changes BOTH the rule and the windows is queued behind
|
||||
the connection lock. Because definitions, rules and windows are read
|
||||
in one locked transaction, the returned bundle is the whole old state
|
||||
— the daily rule with its morning window — never the old rule paired
|
||||
with the new windows. The pause hook holds ``list_windows``
|
||||
mid-flight so the edit is deterministically queued against the
|
||||
helper's lock.
|
||||
"""
|
||||
async def _body(database, children, rules, definitions, child):
|
||||
created = await create_quest_definition(
|
||||
database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
|
||||
)
|
||||
weekly = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY,
|
||||
weekday_set={2},
|
||||
start_date="2026-09-14",
|
||||
)
|
||||
|
||||
import custom_components.nestquest.dao_rules as dao_rules
|
||||
|
||||
original_list = dao_rules.QuestDefinitionsDao.list_windows
|
||||
started = asyncio.Event()
|
||||
release = asyncio.Event()
|
||||
|
||||
async def _pausing_list(self, definition_id):
|
||||
if not started.is_set():
|
||||
started.set()
|
||||
await release.wait()
|
||||
return await original_list(self, definition_id)
|
||||
|
||||
dao_rules.QuestDefinitionsDao.list_windows = _pausing_list
|
||||
try:
|
||||
read_task = asyncio.ensure_future(list_active_definitions(database))
|
||||
await started.wait()
|
||||
edit_task = asyncio.ensure_future(
|
||||
edit_quest_definition(
|
||||
database,
|
||||
created.definition.id,
|
||||
rule=weekly,
|
||||
windows=["evening"],
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(0)
|
||||
release.set()
|
||||
result, edited = await asyncio.gather(read_task, edit_task)
|
||||
finally:
|
||||
dao_rules.QuestDefinitionsDao.list_windows = original_list
|
||||
|
||||
# The helper finished before the queued edit, so its snapshot is
|
||||
# the whole OLD state: the daily rule with its morning window.
|
||||
assert [b.definition.id for b in result] == [created.definition.id]
|
||||
assert result[0].rule == _daily_rule()
|
||||
assert [w.window for w in result[0].windows] == ["morning"]
|
||||
# The edit then reported the whole NEW state, not a mix either.
|
||||
assert edited.rule == weekly
|
||||
assert [w.window for w in edited.windows] == ["evening"]
|
||||
return result
|
||||
|
||||
_with_db(tmp_path, "list-active-snapshot-concurrent.db")(_body)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# full-lifecycle integration: one definition through every operation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user