Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
76d0d50297 |
@@ -1621,3 +1621,113 @@ def test_list_definitions_firing_on_rejects_malformed_date(tmp_path, bad) -> Non
|
||||
return None
|
||||
|
||||
_with_db(tmp_path, "list-firing-on-bad-date.db")(_body)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# full-lifecycle integration: one definition through every operation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_full_lifecycle_never_touches_instances_or_history(tmp_path) -> None:
|
||||
"""End-to-end lifecycle on one definition writes only the four
|
||||
definition tables; ``quest_instances`` and ``completion_events`` are
|
||||
left exactly as they were at every stage.
|
||||
|
||||
A real instance and completion event are seeded before the lifecycle
|
||||
begins, so each stage is checked against pre-existing data being
|
||||
mutated (not an empty table staying empty): create -> edit
|
||||
rule/windows -> assign a second child -> unassign the original child
|
||||
-> deactivate -> reactivate.
|
||||
"""
|
||||
async def _body(database, children, rules, definitions, child):
|
||||
created = await create_quest_definition(
|
||||
database,
|
||||
"Brush teeth",
|
||||
_daily_rule(),
|
||||
[child.id],
|
||||
[("morning", "07:00")],
|
||||
)
|
||||
definition_id = created.definition.id
|
||||
|
||||
instances = QuestInstancesDao(database)
|
||||
events = CompletionEventsDao(database)
|
||||
instance = await instances.upsert(
|
||||
definition_id, child.id, _today_iso(), NOW, window="morning"
|
||||
)
|
||||
event = await events.append(
|
||||
instance.id,
|
||||
child.id,
|
||||
"completed",
|
||||
"user",
|
||||
NOW,
|
||||
True,
|
||||
actor_user_id="admin-1",
|
||||
)
|
||||
instance_count = await database.fetch_one(
|
||||
"SELECT COUNT(*) FROM quest_instances"
|
||||
)
|
||||
event_count = await database.fetch_one(
|
||||
"SELECT COUNT(*) FROM completion_events"
|
||||
)
|
||||
|
||||
async def _assert_history_untouched():
|
||||
assert await instances.get_by_id(instance.id) == instance
|
||||
assert await events.list_by_instance(instance.id) == [event]
|
||||
assert await database.fetch_one(
|
||||
"SELECT COUNT(*) FROM quest_instances"
|
||||
) == instance_count
|
||||
assert await database.fetch_one(
|
||||
"SELECT COUNT(*) FROM completion_events"
|
||||
) == event_count
|
||||
|
||||
await _assert_history_untouched()
|
||||
|
||||
# Edit rule and windows together; the definition's rule row is
|
||||
# rewritten in place and its window set replaced.
|
||||
edited = await edit_quest_definition(
|
||||
database,
|
||||
definition_id,
|
||||
rule=ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY,
|
||||
weekday_set={1, 3},
|
||||
start_date="2026-09-14",
|
||||
),
|
||||
windows=[("morning", "08:00"), ("evening", "19:30")],
|
||||
)
|
||||
assert edited.rule == ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY,
|
||||
weekday_set={1, 3},
|
||||
start_date="2026-09-14",
|
||||
)
|
||||
by_name = {w.window: w.due_time for w in edited.windows}
|
||||
assert by_name == {"morning": "08:00", "evening": "19:30"}
|
||||
await _assert_history_untouched()
|
||||
|
||||
# Assign a second child: multi-assignee, both present.
|
||||
bo = await children.create("Bo", NOW)
|
||||
roster = await assign_child(database, definition_id, bo.id)
|
||||
assert sorted(c.id for c in roster) == sorted([child.id, bo.id])
|
||||
await _assert_history_untouched()
|
||||
|
||||
# Unassign the original child: only the link drops.
|
||||
assert await unassign_child(database, definition_id, child.id) is True
|
||||
assert [
|
||||
c.id for c in await definitions.list_assignees(definition_id)
|
||||
] == [bo.id]
|
||||
await _assert_history_untouched()
|
||||
|
||||
# Deactivate, then reactivate: the definition survives both.
|
||||
deactivated = await set_quest_definition_active(
|
||||
database, definition_id, False
|
||||
)
|
||||
assert deactivated.definition.is_active is False
|
||||
await _assert_history_untouched()
|
||||
reactivated = await set_quest_definition_active(
|
||||
database, definition_id, True
|
||||
)
|
||||
assert reactivated.definition.is_active is True
|
||||
await _assert_history_untouched()
|
||||
|
||||
return reactivated
|
||||
|
||||
_with_db(tmp_path, "full-lifecycle.db")(_body)
|
||||
|
||||
Reference in New Issue
Block a user