Compare commits

...
Author SHA1 Message Date
overseer 98e89f8b29 fix: reject invalid ScheduleRule storage shapes (review R66-001, R66-002) 2026-09-17 08:09:09 -04:00
overseer e92c79b211 feat: map ScheduleRule models to and from schedule_rules storage 2026-09-17 07:55:03 -04:00
orchestrator-agent 58b11b59cf Merge pull request 'Feature 05: Presence and Custody Schedules -> dev' (#64) from feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules into dev 2026-09-15 18:33:54 -04:00
dev_agent 9d21f29eea Merge pull request 'Year-boundary and leap-year regression test for custody rotation (Feature 05, task 7)' (#63) from task/5841f120-parity-regression into feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules 2026-09-15 18:32:45 -04:00
Joshua King fec607f039 Review fixes: correct 2015 anchor weekday and ISO rollover facts
Review round 2 (Codex gpt-5.6-terra, 2 LOW): 2015-01-06 is a Tuesday,
not a Monday — the anchor is now the computed first Monday of January
(asserted weekday 0); the docstring names the ISO-week rollover
precisely (the Monday on or after January 4th — January 4th in both
walked years, which end ISO week 53) instead of 'January 1st'. Suite:
1005 passed.
2026-09-15 18:31:38 -04:00
Joshua King a18c0060e4 Review fix: extend the parity walk through the following January
Review round 1 (Codex gpt-5.6-terra, MEDIUM): the walk stopped at
December 31st, so the boundary block's next-January days went
unchecked and a year-resetting implementation would pass. The walk
now runs through January 14th of the following year, asserting block
uniformity across the December 31st -> January 1st transition
(including the post-53-week jump), and the walked-year day count is
verified independently of the walk. Suite: 1005 passed.
2026-09-15 18:28:39 -04:00
Joshua King 20d7efb04b Year-boundary and leap-year regression guard (Feature 05, task 7)
The week-parity regression test (D-004): walks a two-week alternating
rotation day by day across a FULL 366-day leap year (2020) and a full
365-day year with 53 ISO weeks (2015), asserting the present-day
sequence never breaks or inverts, including January 1st.  The
expectation is derived from the anchor's 7-day blocks — each block
uniform, consecutive blocks alternating — not from the engine's own
index formula, so a parity-based implementation that resets on
January 1st breaks block uniformity or alternation (most visibly in
the block spanning the boundary).  Year facts asserted: 2020 = 366
days and ISO week 53; 2015 = 365 days and ISO week 53.  The test
docstring names itself as the regression guard.  Suite: 1004 passed.
2026-09-15 18:25:32 -04:00
dev_agent 6551a87bd6 Merge pull request 'Provide a next-N-present-days preview function (Feature 05, task 6)' (#62) from task/7f0190c4-present-preview into feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules 2026-09-15 18:23:35 -04:00
Joshua King e9dc0a1445 Test fix: week 0 runs Mon Jan 5 through Sun Jan 11
The six-day preview expectation missed that Jan 10-11 are still in
the present week (the away week starts Jan 12). Suite green before
commit: 1003 passed.
2026-09-15 18:20:42 -04:00
Joshua King 1bb0617471 Next-N-present-days preview (Feature 05, task 6)
PresenceEngine.next_present_dates(child_id, start_date, count):
walks forward from the INCLUSIVE start date through is_present, so
overrides and the pattern are honoured identically — this is the
preview the admin Schedule tab renders before a parent saves a
custody pattern. Count must be a positive int; the scan is capped at
two years (MAX_PREVIEW_SCAN_DAYS) and an always-absent child raises
a clear ValueError instead of looping forever. Tests: known two-week
pattern (straight five-day week, mid-week start skipping the absent
week), inclusive start, override blocking the preview start,
schedule-free child, input validation, always-absent error. Suite:
1002 passed.
2026-09-15 18:19:53 -04:00
dev_agent 221ab2c20b Merge pull request 'Make overrides take precedence over the repeating pattern (Feature 05, task 5)' (#61) from task/6a1e14aa-override-precedence into feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules 2026-09-15 18:17:40 -04:00
Joshua King bdf3513609 Test fix: build the frozen-snapshot engine on the caller's list
Review round 3 (Codex gpt-5.6-terra, LOW): the helper copied the list,
so the mutation could never have affected even a shallow snapshot.
The engine is now constructed directly on the caller's list, making
the freeze regression-proof. Suite green before commit: 997 passed.
2026-09-15 18:16:40 -04:00
Joshua King 830d663a45 Test fix: tuples raise AttributeError for append, not TypeError
The frozen-snapshot test expected TypeError from tuple.append; tuples
have no append at all. Suite green BEFORE commit: 997 passed.
2026-09-15 18:15:13 -04:00
Joshua King bff2f4ff5e Review fix: freeze the engine's overrides snapshot
Review round 1 (Codex gpt-5.6-terra, MEDIUM): the overrides mapping
was shallow-copied, leaving caller-owned lists mutable through
_overrides — a later append would change is_present results. Each
validated list is copied into an immutable tuple before wrapping;
tests prove post-construction appends change nothing and the snapshot
refuses mutation. Suite: 996 passed.
2026-09-15 18:13:48 -04:00
Joshua King 8f04d93363 Overrides take precedence over the pattern (Feature 05, task 5)
PresenceEngine gains an overrides snapshot (child_id -> list of
PresenceOverride): is_present checks covering overrides BEFORE the
pattern and returns the override's flag as the answer — an absent
override on a normally-present day returns False, a present one on a
normally-absent day returns True, and an override on a child with no
schedule still applies (holidays work for the always-present sibling).
Snapshot validation mirrors the schedules': dict of lists keyed by the
override's own child id, exposed read-only. The DAO layer guarantees
no overlapping overrides per child; a hand-built snapshot that carries
them resolves deterministically to the LATEST start date (documented).
Tests: all three precedence cases, span edges (outside days stay on
the pattern), snapshot rejection matrix, deterministic overlap
tie-break. Suite: 995 passed.
2026-09-15 18:12:16 -04:00
dev_agent d9326deeb5 Merge pull request 'Implement presence overrides for holidays and swapped weeks (Feature 05, task 4)' (#60) from task/46a867a4-override-overlap into feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules 2026-09-15 18:08:07 -04:00
Joshua King 94f54523f9 Review fix: gate the override race test inside A's transaction
Review round 1 (Codex gpt-5.6-terra, LOW): the racing test completed
synchronously and proved nothing. The executor now gates task A's
INSERT — inside its open transaction, after its overlap probe — so
task B queues on the connection lock, re-runs the probe against A's
committed row, and is rejected; exactly one lands. Suite: 990 passed.
2026-09-15 18:06:44 -04:00
Joshua King 51b1388233 Fix: inverted ranges skip the overlap probe and hit the schema CHECK
The matrix test's end-before-start probe (expecting the schema's
end>=start IntegrityError) hit the new overlap query first, since an
inverted range lexically 'overlaps' the seeded row. The overlap
check now runs only for well-ordered ranges; invalid ones fall
through to the INSERT and keep the IntegrityError contract. Suite:
990 passed.
2026-09-15 18:03:49 -04:00
Joshua King 744a40dcb5 Presence override overlap rejection naming the conflict (Feature 05, task 4)
PresenceOverridesDao.create now refuses an override that overlaps any
existing override of the SAME child (inclusive boundaries): the check
runs inside the same transaction as the INSERT under the connection
lock, so a racing create cannot slip a conflict between check and
write — a deterministic race test proves exactly one of two concurrent
overlapping creates lands. The ValueError names the conflicting
override's id, its range, and its flag. Round-trip, per-child scope,
and remove-then-recreate tests added. Suite: 990 passed.
2026-09-15 18:03:00 -04:00
dev_agent e15648232f Merge pull request 'Implement is_present with always-present default (Feature 05, task 3)' (#59) from task/5059d16c-is-present into feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules 2026-09-15 18:00:06 -04:00
Joshua King 6d48f433b7 Review fix: validate engine keys as child ids; every-day three-child walk
Review round 1 (Codex gpt-5.6-terra, 2 LOW): engine snapshot keys now
pass _validate_child_id (True/1.0 compare equal to 1 and would be
accepted then behave differently on lookup); the always-present test
now walks all 28 days asserting all three children's expected verdicts
instead of sampling every third day. Suite: 985 passed.
2026-09-15 17:58:51 -04:00
Joshua King 540f59a353 PresenceEngine with is_present and the always-present default (Feature 05, task 3)
PresenceEngine is an immutable, database-free snapshot of the
children's schedules answering is_present(child_id, date): a child
with NO schedule entry is present every day (the household's
always-present sibling); a scheduled child is present exactly when
the date's weekday is in the pattern for that date's cycle week
(anchor-date arithmetic via cycle_week_index — never ISO parity).
Construction validates the snapshot shape (dict keyed by the
schedule's own child_id, values are PresenceSchedules) and exposes it
read-only (MappingProxyType + __setattr__ refusal). is_present
rejects non-int child ids and non-plain dates.

Tests: the three-child fixture (two complementary rotation children,
one schedule-free) asserted for every day of a four-week span plus
spot-checked swap days; weekend/weekday-set behaviour across weeks;
construction/input rejection matrix; engine immutability. Suite:
985 passed.
2026-09-15 17:57:02 -04:00
dev_agent ee994a3938 Merge pull request 'Implement anchor-date cycle position arithmetic (Feature 05, task 2)' (#58) from task/06cc4782-cycle-arithmetic into feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules 2026-09-15 17:55:21 -04:00
dev_agent 7ae4018e59 Merge pull request 'Define the PresenceSchedule model with validation (Feature 05, task 1)' (#57) from task/ede1dc76-presence-model into feature/167e3565-704d-41c7-adf6-3b88ed266f78-presence-custody-schedules 2026-09-15 17:48:35 -04:00
6 changed files with 1264 additions and 1 deletions
@@ -186,6 +186,15 @@ class PresenceOverridesDao:
The child must exist (validated atomically under the The child must exist (validated atomically under the
connection lock). Date shape is validated strictly, and the connection lock). Date shape is validated strictly, and the
schema CHECK rejects end < start. schema CHECK rejects end < start.
Overlap policy (Feature 05): an override for a child may not
overlap another override of the SAME child — two different
present/absent answers for one date are unresolvable. The
overlap check (inclusive boundaries) runs inside the same
transaction as the INSERT, so a racing create cannot slip a
conflicting override between the check and the write; a
conflict raises ValueError naming the conflicting override's
id and range.
""" """
_validate_date(start_date, "start_date") _validate_date(start_date, "start_date")
_validate_date(end_date, "end_date") _validate_date(end_date, "end_date")
@@ -196,6 +205,29 @@ class PresenceOverridesDao:
) )
if child is None: if child is None:
raise ValueError(f"child {child_id} does not exist") raise ValueError(f"child {child_id} does not exist")
# Only a well-ordered range has overlap semantics:
# inverted ranges fall through to the INSERT and hit
# the schema's end>=start CHECK, preserving the
# IntegrityError contract the matrix tests rely on.
if end_date >= start_date:
conflicting = await self._database.fetch_one(
f"SELECT {_OVERRIDE_COLUMNS} "
"FROM presence_overrides "
"WHERE child_id = ? AND end_date >= ? "
"AND start_date <= ? ORDER BY id LIMIT 1",
(child_id, start_date, end_date),
)
else:
conflicting = None
if conflicting is not None:
conflict = _override_from_row(conflicting)
raise ValueError(
f"override for child {child_id} "
f"[{start_date}, {end_date}] overlaps existing "
f"override {conflict.id} "
f"[{conflict.start_date}, {conflict.end_date}] "
f"(is_present={conflict.is_present})"
)
result = await self._database.execute( result = await self._database.execute(
"INSERT INTO presence_overrides (child_id, " "INSERT INTO presence_overrides (child_id, "
"start_date, end_date, is_present, note) " "start_date, end_date, is_present, note) "
+212
View File
@@ -40,6 +40,7 @@ from .const import QUEST_WINDOWS
from .dao_children import ChildRecord, _child_from_row, _connection_lock from .dao_children import ChildRecord, _child_from_row, _connection_lock
from .dao_children import _CHILD_COLUMNS from .dao_children import _CHILD_COLUMNS
from .db import NestQuestDatabase from .db import NestQuestDatabase
from .recurrence import RuleType, RuleValidationError, ScheduleRule
#: Sentinel distinguishing "argument omitted" from "explicit SQL NULL" #: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
#: in update methods: passing ``None`` must mean clearing a nullable #: in update methods: passing ``None`` must mean clearing a nullable
@@ -119,6 +120,26 @@ class ScheduleRuleRecord:
end_date: str | None end_date: str | None
@dataclass(frozen=True)
class ScheduleRuleStorage:
"""The eight value columns a ScheduleRule maps to (no id column).
``id`` is assigned by the database on insert, so the model<->storage
mapping concerns only the value columns. ``nth_weekday_weekday`` is
deliberately absent here: for MONTHLY_WEEKDAY its value is folded
into ``weekday_set`` as that column's single CSV element.
"""
rule_type: str
interval: int
weekday_set: str | None
day_of_month: int | None
nth_weekday: int | None
month: int | None
start_date: str
end_date: str | None
@dataclass(frozen=True) @dataclass(frozen=True)
class QuestDefinitionRecord: class QuestDefinitionRecord:
"""One row of ``quest_definitions`` (without its assignees).""" """One row of ``quest_definitions`` (without its assignees)."""
@@ -188,6 +209,197 @@ def _window_from_row(row: tuple) -> QuestDefinitionWindowRecord:
) )
#: Storage columns that are forbidden for each rule type. A populated
#: forbidden column means the row's shape is invalid and must raise on
#: read, never be silently dropped. ``weekday_set`` is absent for
#: DAILY/MONTHLY_DAY/YEARLY; MONTHLY_WEEKDAY is the only shape allowed a
#: non-NULL ``nth_weekday``; YEARLY is the only shape allowed ``month``.
_FORBIDDEN_STORAGE_FIELDS: dict[str, tuple[RuleType, ...]] = {
"weekday_set": (RuleType.DAILY, RuleType.MONTHLY_DAY, RuleType.YEARLY),
"day_of_month": (
RuleType.DAILY,
RuleType.WEEKLY,
RuleType.CUSTOM_DAYS,
RuleType.MONTHLY_WEEKDAY,
),
"nth_weekday": (
RuleType.DAILY,
RuleType.WEEKLY,
RuleType.CUSTOM_DAYS,
RuleType.MONTHLY_DAY,
RuleType.YEARLY,
),
"month": (
RuleType.DAILY,
RuleType.WEEKLY,
RuleType.CUSTOM_DAYS,
RuleType.MONTHLY_DAY,
RuleType.MONTHLY_WEEKDAY,
),
}
def schedule_rule_to_storage(rule: ScheduleRule) -> ScheduleRuleStorage:
"""Map a validated ScheduleRule onto its storage columns.
``rule_type`` stores as ``RuleType.storage_value`` (MONTHLY_DAY and
MONTHLY_WEEKDAY both store ``'monthly'``); ``weekday_set`` stores as
a sorted CSV list. MONTHLY_WEEKDAY folds its model-only
``nth_weekday_weekday`` into ``weekday_set`` as that list's single
element; every other type leaves the column as its own set
(WEEKLY/CUSTOM_DAYS) or NULL.
"""
shape = rule.rule_type
if shape is RuleType.MONTHLY_WEEKDAY:
weekday_set = str(rule.nth_weekday_weekday)
elif shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
weekday_set = ",".join(str(w) for w in sorted(rule.weekday_set))
else:
weekday_set = None
return ScheduleRuleStorage(
rule_type=shape.storage_value,
interval=rule.interval,
weekday_set=weekday_set,
day_of_month=rule.day_of_month,
nth_weekday=rule.nth_weekday,
month=rule.month,
start_date=rule.start_date,
end_date=rule.end_date,
)
def schedule_rule_from_storage(storage: ScheduleRuleStorage) -> ScheduleRule:
"""Reconstruct a ScheduleRule from storage columns.
Disambiguates MONTHLY_DAY from MONTHLY_WEEKDAY by which field is
populated, and raises RuleValidationError rather than silently
mis-decoding an ambiguous or invalid shape.
"""
rule_type = storage.rule_type
if rule_type == "daily":
shape = RuleType.DAILY
elif rule_type == "weekly":
shape = RuleType.WEEKLY
elif rule_type == "monthly":
shape = _disambiguate_monthly(storage)
elif rule_type == "yearly":
shape = RuleType.YEARLY
elif rule_type == "custom":
shape = RuleType.CUSTOM_DAYS
else:
raise RuleValidationError(
f"unknown storage rule_type {rule_type!r}"
)
_reject_forbidden_storage_fields(storage, shape)
weekday_set = None
nth_weekday_weekday = None
if shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
weekday_set = _weekdays_from_csv(storage.weekday_set)
elif shape is RuleType.MONTHLY_WEEKDAY:
nth_weekday_weekday = _single_weekday_from_csv(storage.weekday_set)
return ScheduleRule(
rule_type=shape,
interval=storage.interval,
weekday_set=weekday_set,
day_of_month=storage.day_of_month,
nth_weekday=storage.nth_weekday,
nth_weekday_weekday=nth_weekday_weekday,
month=storage.month,
start_date=storage.start_date,
end_date=storage.end_date,
)
def _reject_forbidden_storage_fields(
storage: ScheduleRuleStorage, shape: RuleType
) -> None:
"""Raise unless every populated column is permitted for ``shape``.
The model constructor rejects a forbidden ``day_of_month``,
``nth_weekday``, or ``month``, but ``weekday_set`` never reaches it
(it is folded for MONTHLY_WEEKDAY and dropped otherwise), so a daily
row carrying ``weekday_set="0"`` would otherwise decode as a valid
daily rule and silently lose the stored data. This validates every
shape column explicitly so a populated-forbidden column raises
instead of being discarded.
"""
for field, forbidden_shapes in _FORBIDDEN_STORAGE_FIELDS.items():
value = getattr(storage, field)
if value is not None and shape in forbidden_shapes:
raise RuleValidationError(
f"{shape.value} storage rows must not set {field}, "
f"got {value!r}"
)
def _disambiguate_monthly(storage: ScheduleRuleStorage) -> RuleType:
"""Resolve the shared 'monthly' storage value to a model type.
MONTHLY_DAY names ``day_of_month`` only; MONTHLY_WEEKDAY names
``nth_weekday`` only. Neither, or both, is ambiguous and raises.
"""
day_of_month = storage.day_of_month
nth_weekday = storage.nth_weekday
if day_of_month is not None and nth_weekday is None:
return RuleType.MONTHLY_DAY
if nth_weekday is not None and day_of_month is None:
return RuleType.MONTHLY_WEEKDAY
raise RuleValidationError(
"storage rule_type 'monthly' is ambiguous: must name exactly one "
"of day_of_month or nth_weekday, got "
f"day_of_month={day_of_month!r}, nth_weekday={nth_weekday!r}"
)
def _weekdays_from_csv(value: str | None) -> frozenset[int]:
"""Parse a weekday CSV column into a frozenset of ints.
``None`` and ``""`` both yield an empty frozenset (the caller decides
whether empty is legal); a non-integer token raises
RuleValidationError rather than leaking a raw ValueError.
"""
if value is None or value == "":
return frozenset()
try:
entries = [int(token) for token in value.split(",")]
except ValueError:
raise RuleValidationError(
f"weekday_set must be a CSV of integers, got {value!r}"
) from None
return frozenset(entries)
def _single_weekday_from_csv(value: str | None) -> int:
"""The one weekday a MONTHLY_WEEKDAY stores in weekday_set.
Parses the CSV into tokens FIRST and requires exactly one token,
so a duplicate pair like ``"1,1"`` is rejected rather than silently
canonicalized to a single weekday on re-storage. An empty or
multi-element column raises RuleValidationError.
"""
if value is None or value == "":
raise RuleValidationError(
"MONTHLY_WEEKDAY storage weekday_set must hold exactly one "
f"weekday, got {value!r}"
)
tokens = value.split(",")
if len(tokens) != 1:
raise RuleValidationError(
"MONTHLY_WEEKDAY storage weekday_set must hold exactly one "
f"weekday, got {value!r}"
)
try:
return int(tokens[0])
except ValueError:
raise RuleValidationError(
"MONTHLY_WEEKDAY storage weekday_set must be an integer, "
f"got {value!r}"
) from None
class ScheduleRulesDao: class ScheduleRulesDao:
"""Typed async access to the ``schedule_rules`` table.""" """Typed async access to the ``schedule_rules`` table."""
+158
View File
@@ -317,3 +317,161 @@ def cycle_week_index(
target = _parse_date(target_date, "target_date") target = _parse_date(target_date, "target_date")
days = (target - schedule.anchor_date).days days = (target - schedule.anchor_date).days
return (days // 7) % schedule.cycle_length_weeks return (days // 7) % schedule.cycle_length_weeks
#: Search cap for :meth:`PresenceEngine.next_present_dates`: two full
#: years of daily checks. A child absent every day (an all-absent
#: pattern with a blocking override) would otherwise loop forever; the
#: cap bounds the scan and surfaces the situation as an error instead.
MAX_PREVIEW_SCAN_DAYS = 366 * 2
class PresenceEngine:
"""Answers "is this child at this house on this date?".
Pure and side-effect-free: built from an immutable snapshot of the
children's schedules (``child_id -> PresenceSchedule``), it never
touches the database — callers build a snapshot from the DAO layer
and pass it in. A child with NO schedule entry is present every
day (the third household child's case); a child WITH a schedule is
present exactly when the date's weekday is in that schedule's
pattern for the date's cycle week index (anchor-date arithmetic —
:func:`cycle_week_index`, never ISO week parity).
"""
def __init__(
self,
schedules: dict[int, PresenceSchedule],
overrides: dict[int, list[PresenceOverride]] | None = None,
) -> None:
if not isinstance(schedules, dict):
raise ValueError(
f"schedules must map child_id to PresenceSchedule, got "
f"{schedules!r}"
)
for child_id, schedule in schedules.items():
# Validate the KEY as a model child id too: True and 1.0
# compare equal to 1, so {True: schedule(1)} would be
# accepted and then behave differently on lookup.
_validate_child_id(child_id)
if not isinstance(schedule, PresenceSchedule):
raise ValueError(
f"schedules[{child_id!r}] must be a PresenceSchedule, "
f"got {schedule!r}"
)
if schedule.child_id != child_id:
raise ValueError(
f"schedules is keyed {child_id!r} but holds the "
f"schedule of child {schedule.child_id}"
)
if overrides is None:
overrides = {}
if not isinstance(overrides, dict):
raise ValueError(
"overrides must map child_id to a list of "
f"PresenceOverride, got {overrides!r}"
)
for child_id, entries in overrides.items():
_validate_child_id(child_id)
if not isinstance(entries, list):
raise ValueError(
f"overrides[{child_id!r}] must be a list of "
f"PresenceOverride, got {entries!r}"
)
for entry in entries:
if not isinstance(entry, PresenceOverride):
raise ValueError(
f"overrides[{child_id!r}] entries must be "
f"PresenceOverride, got {entry!r}"
)
if entry.child_id != child_id:
raise ValueError(
f"overrides is keyed {child_id!r} but holds an "
f"override of child {entry.child_id}"
)
# Read-only snapshot: reassignment is refused and no mutator
# exists, so a built engine answers consistently for its
# lifetime. The overrides lists are copied to immutable
# tuples — a shallow mapping copy alone would leave caller-
# owned lists reachable through _overrides, and mutating one
# later would change is_present results.
object.__setattr__(self, "_schedules", MappingProxyType(dict(schedules)))
object.__setattr__(
self,
"_overrides",
MappingProxyType(
{child_id: tuple(entries) for child_id, entries in overrides.items()}
),
)
def is_present(self, child_id: object, date: object) -> bool:
"""Return whether ``child_id`` is present on ``date``.
Overrides take precedence over the repeating pattern: when an
override covers the date, its flag IS the answer — including
for a child with no schedule at all. The DAO layer guarantees
a child's overrides never overlap; if a snapshot still carries
two covering the same date, the one with the LATEST start date
wins (deterministic, and matches "the most specific swap wins").
"""
if isinstance(child_id, bool) or not isinstance(child_id, int):
raise ValueError(f"child_id must be an integer, got {child_id!r}")
target = _parse_date(date, "date")
entries = self._overrides.get(child_id) or []
covering = [
entry for entry in entries if entry.covers(target)
]
if covering:
winner = max(covering, key=lambda entry: entry.start_date)
return winner.is_present
schedule = self._schedules.get(child_id)
if schedule is None:
# No schedule at all: present every day (household default).
return True
week = cycle_week_index(schedule, target)
return target.weekday() in schedule.pattern[week]
def next_present_dates(
self,
child_id: object,
start_date: object,
count: int,
) -> list[datetime.date]:
"""Return the next ``count`` dates the child is present.
Walks forward from ``start_date`` (INCLUSIVE — a parent
previewing "starting Monday" expects Monday itself when the
child is present) checking :meth:`is_present` for every day,
so overrides and the pattern are honoured identically. The
admin Schedule tab renders this preview so a parent can confirm
a custody pattern is right before saving it.
A child absent every scanned day raises ValueError after the
two-year search cap — an always-absent child would otherwise
loop forever. ``count`` must be a positive integer.
"""
_validate_child_id(child_id)
start = _parse_date(start_date, "start_date")
if isinstance(count, bool) or not isinstance(count, int):
raise ValueError(f"count must be an integer, got {count!r}")
if count < 1:
raise ValueError(f"count must be at least 1, got {count}")
found: list[datetime.date] = []
day = start
for _ in range(MAX_PREVIEW_SCAN_DAYS):
if self.is_present(child_id, day):
found.append(day)
if len(found) == count:
return found
day += datetime.timedelta(days=1)
raise ValueError(
f"child {child_id} is not present on any of the "
f"{MAX_PREVIEW_SCAN_DAYS} days from {start.isoformat()}; "
f"{count} present day(s) cannot be previewed"
)
def __setattr__(self, name: object, value: object) -> None:
raise AttributeError(
"PresenceEngine is an immutable snapshot; build a new engine "
"instead of mutating this one"
)
+169
View File
@@ -620,3 +620,172 @@ def test_dao_presence_test_file_uses_dao_not_raw_table_sql() -> None:
assert sql_pattern.search(remaining) is None, ( assert sql_pattern.search(remaining) is None, (
"tests must go through the DAO, not raw SQL, for these tables" "tests must go through the DAO, not raw SQL, for these tables"
) )
# ---------------------------------------------------------------------------
# Presence overrides: overlap rejection naming the conflict
# ---------------------------------------------------------------------------
def test_override_create_round_trip_single_and_range(tmp_path) -> None:
async def _body(database, schedules, overrides, children, child):
single = await overrides.create(
child.id, "2026-07-20", "2026-07-20", False
)
assert (single.start_date, single.end_date) == ("2026-07-20",) * 2
span = await overrides.create(
child.id, "2026-08-01", "2026-08-05", True, note="traded week"
)
assert span.is_present is True
assert span.note == "traded week"
# Distinct, non-overlapping ranges coexist.
listed = await overrides.list_by_child_and_range(
child.id, "2026-01-01", "2026-12-31"
)
assert [row.id for row in listed] == [single.id, span.id]
return listed
_with_db(tmp_path, "override-round-trip.db")(_body)
def test_override_overlap_rejected_and_names_the_conflict(tmp_path) -> None:
async def _body(database, schedules, overrides, children, child):
existing = await overrides.create(
child.id, "2026-07-20", "2026-07-24", False, note="holiday"
)
for start, end in (
("2026-07-24", "2026-07-26"), # touches the existing end
("2026-07-18", "2026-07-20"), # touches the existing start
("2026-07-21", "2026-07-22"), # fully inside
("2026-07-01", "2026-07-30"), # fully covering
):
with pytest.raises(ValueError, match="overlaps existing"):
await overrides.create(
child.id, start, end, True, note="swap"
)
# The error NAMES the conflicting override (id + its range).
with pytest.raises(
ValueError,
match=(
rf"override {existing.id} "
rf"\[2026-07-20, 2026-07-24\].*is_present=False"
),
):
await overrides.create(child.id, "2026-07-22", "2026-07-22", True)
# Nothing was written by any rejected attempt.
listed = await overrides.list_by_child_and_range(
child.id, "2026-01-01", "2026-12-31"
)
assert [row.id for row in listed] == [existing.id]
return existing
_with_db(tmp_path, "override-overlap.db")(_body)
def test_override_overlap_is_per_child(tmp_path) -> None:
"""Different children's overrides may cover the same dates: the
conflict rule is per child."""
async def _body(database, schedules, overrides, children, child):
other = await children.create("Bo", NOW)
await overrides.create(child.id, "2026-07-20", "2026-07-24", False)
twin = await overrides.create(
other.id, "2026-07-20", "2026-07-24", True
)
assert twin.child_id == other.id
listed = await overrides.list_by_child_and_range(
other.id, "2026-01-01", "2026-12-31"
)
assert [row.id for row in listed] == [twin.id]
return twin
_with_db(tmp_path, "override-per-child.db")(_body)
def test_override_remove_then_recreate_overlapping(tmp_path) -> None:
async def _body(database, schedules, overrides, children, child):
first = await overrides.create(
child.id, "2026-07-20", "2026-07-24", False
)
assert await overrides.delete(first.id) is True
assert await overrides.delete(first.id) is False
replacement = await overrides.create(
child.id, "2026-07-20", "2026-07-24", True, note="swap back"
)
assert replacement.is_present is True
return replacement
_with_db(tmp_path, "override-recreate.db")(_body)
def test_override_create_racing_conflicts_serialize(tmp_path) -> None:
"""Two racing creates for the same child and dates: exactly one
lands, the other is rejected with the overlap error. DETERMINISTIC
gate: task A is paused at its INSERT — inside its open transaction,
after its overlap probe — so task B is created while A provably
holds the connection lock and must queue; B then re-runs the probe
against A's committed row and is rejected."""
import asyncio as asyncio_module
from unittest.mock import MagicMock
async def _main():
from custom_components.nestquest.dao_children import ChildrenDao
from custom_components.nestquest.dao_presence import (
PresenceOverridesDao,
)
from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import apply_migrations
armed = {"active": False, "gated": False}
gate_open = asyncio_module.Event()
a_started = asyncio_module.Event()
hass = MagicMock()
async def _gated_executor(fn, *args):
if armed["active"]:
# Gate the FIRST _execute after arming: task A's INSERT,
# which runs inside its open transaction (BEGIN already
# executed) after the overlap probe.
if getattr(fn, "__name__", "") == "_execute" and not armed["gated"]:
armed["gated"] = True
a_started.set()
await gate_open.wait()
return fn(*args)
hass.async_add_executor_job = _gated_executor
database = NestQuestDatabase(hass)
await database.open(tmp_path / "override-race.db")
try:
await apply_migrations(database)
children = ChildrenDao(database)
overrides = PresenceOverridesDao(database)
child = await children.create("Ada", NOW)
armed["active"] = True
task_a = asyncio_module.ensure_future(
overrides.create(child.id, "2026-07-20", "2026-07-24", True)
)
await a_started.wait()
armed["active"] = False
task_b = asyncio_module.ensure_future(
overrides.create(child.id, "2026-07-22", "2026-07-26", False)
)
await asyncio_module.sleep(0)
gate_open.set()
record_a, error_b = await asyncio_module.gather(
task_a, task_b, return_exceptions=True
)
assert isinstance(record_a, object) and not isinstance(
record_a, BaseException
), f"A must land: {record_a!r}"
assert isinstance(error_b, ValueError), (
f"B must be rejected with the overlap error: {error_b!r}"
)
assert "overlaps existing" in str(error_b)
listed = await overrides.list_by_child_and_range(
child.id, "2026-01-01", "2026-12-31"
)
assert [row.id for row in listed] == [record_a.id]
finally:
await database.close()
asyncio.new_event_loop().run_until_complete(_main())
+251
View File
@@ -9,13 +9,21 @@ import pytest
from custom_components.nestquest.dao_children import ChildrenDao from custom_components.nestquest.dao_children import ChildrenDao
from custom_components.nestquest.dao_rules import ( from custom_components.nestquest.dao_rules import (
ScheduleRuleRecord, ScheduleRuleRecord,
ScheduleRuleStorage,
ScheduleRulesDao, ScheduleRulesDao,
QuestDefinitionRecord, QuestDefinitionRecord,
QuestDefinitionWindowRecord, QuestDefinitionWindowRecord,
QuestDefinitionsDao, QuestDefinitionsDao,
schedule_rule_from_storage,
schedule_rule_to_storage,
) )
from custom_components.nestquest.db import NestQuestDatabase from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import apply_migrations from custom_components.nestquest.migrations import apply_migrations
from custom_components.nestquest.recurrence import (
RuleType,
RuleValidationError,
ScheduleRule,
)
def _run(coro): def _run(coro):
@@ -423,6 +431,249 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
assert create_won is False assert create_won is False
# ---------------------------------------------------------------------------
# ScheduleRule <-> storage mapping
# ---------------------------------------------------------------------------
_RULE_STORAGE_CASES = [
(
ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01"),
ScheduleRuleStorage(
"daily", 1, None, None, None, None, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.DAILY,
interval=3,
start_date="2026-09-01",
end_date="2026-12-31",
),
ScheduleRuleStorage(
"daily", 3, None, None, None, None, "2026-09-01", "2026-12-31"
),
),
(
ScheduleRule(
rule_type=RuleType.WEEKLY,
weekday_set={2, 0, 4},
start_date="2026-09-01",
),
ScheduleRuleStorage(
"weekly", 1, "0,2,4", None, None, None, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.CUSTOM_DAYS,
weekday_set={1, 3, 5},
interval=2,
start_date="2026-09-01",
),
ScheduleRuleStorage(
"custom", 2, "1,3,5", None, None, None, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.MONTHLY_DAY,
day_of_month=15,
start_date="2026-09-01",
),
ScheduleRuleStorage(
"monthly", 1, None, 15, None, None, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.MONTHLY_DAY,
day_of_month=31,
interval=3,
start_date="2026-09-01",
),
ScheduleRuleStorage(
"monthly", 3, None, 31, None, None, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.MONTHLY_WEEKDAY,
nth_weekday=2,
nth_weekday_weekday=1,
start_date="2026-09-01",
),
ScheduleRuleStorage(
"monthly", 1, "1", None, 2, None, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.MONTHLY_WEEKDAY,
nth_weekday=-1,
nth_weekday_weekday=4,
interval=2,
start_date="2026-09-01",
),
ScheduleRuleStorage(
"monthly", 2, "4", None, -1, None, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.YEARLY, month=6, start_date="2026-09-01"
),
ScheduleRuleStorage(
"yearly", 1, None, None, None, 6, "2026-09-01", None
),
),
(
ScheduleRule(
rule_type=RuleType.YEARLY,
month=2,
day_of_month=29,
start_date="2026-09-01",
),
ScheduleRuleStorage(
"yearly", 1, None, 29, None, 2, "2026-09-01", None
),
),
]
@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
def test_schedule_rule_to_storage_maps_each_type(rule, storage) -> None:
assert schedule_rule_to_storage(rule) == storage
@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
def test_schedule_rule_from_storage_rebuilds_each_type(rule, storage) -> None:
assert schedule_rule_from_storage(storage) == rule
@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
def test_schedule_rule_round_trips_losslessly_both_directions(
rule, storage
) -> None:
assert schedule_rule_from_storage(schedule_rule_to_storage(rule)) == rule
assert schedule_rule_to_storage(schedule_rule_from_storage(storage)) == storage
@pytest.mark.parametrize(
"storage",
[
# monthly with neither day_of_month nor nth_weekday
ScheduleRuleStorage(
"monthly", 1, None, None, None, None, "2026-09-01", None
),
# monthly with both day_of_month and nth_weekday
ScheduleRuleStorage(
"monthly", 1, None, 15, 2, None, "2026-09-01", None
),
# MONTHLY_WEEKDAY whose weekday_set is None
ScheduleRuleStorage(
"monthly", 1, None, None, 2, None, "2026-09-01", None
),
# MONTHLY_WEEKDAY whose weekday_set is empty
ScheduleRuleStorage("monthly", 1, "", None, 2, None, "2026-09-01", None),
# MONTHLY_WEEKDAY whose weekday_set is multi-element
ScheduleRuleStorage(
"monthly", 1, "1,3", None, 2, None, "2026-09-01", None
),
# unknown storage rule_type
ScheduleRuleStorage(
"sometimes", 1, None, None, None, None, "2026-09-01", None
),
],
)
def test_schedule_rule_from_storage_rejects_ambiguous_or_invalid(
storage,
) -> None:
with pytest.raises(RuleValidationError):
schedule_rule_from_storage(storage)
def test_schedule_rule_from_storage_rejects_malformed_weekday_csv() -> None:
with pytest.raises(RuleValidationError):
schedule_rule_from_storage(
ScheduleRuleStorage(
"weekly", 1, "0,2,x", None, None, None, "2026-09-01", None
)
)
with pytest.raises(RuleValidationError):
schedule_rule_from_storage(
ScheduleRuleStorage(
"weekly", 1, "7", None, None, None, "2026-09-01", None
)
)
@pytest.mark.parametrize(
"storage",
[
# DAILY carrying shape columns it must not hold.
ScheduleRuleStorage(
"daily", 1, "0", None, None, None, "2026-09-01", None
),
ScheduleRuleStorage(
"daily", 1, None, 15, None, None, "2026-09-01", None
),
ScheduleRuleStorage(
"daily", 1, None, None, 2, None, "2026-09-01", None
),
ScheduleRuleStorage(
"daily", 1, None, None, None, 6, "2026-09-01", None
),
# WEEKLY carrying shape columns it must not hold.
ScheduleRuleStorage(
"weekly", 1, "0,2", 15, None, None, "2026-09-01", None
),
ScheduleRuleStorage(
"weekly", 1, "0,2", None, None, 6, "2026-09-01", None
),
# CUSTOM_DAYS carrying shape columns it must not hold.
ScheduleRuleStorage(
"custom", 1, "0,2", 15, None, None, "2026-09-01", None
),
ScheduleRuleStorage(
"custom", 1, "0,2", None, None, 6, "2026-09-01", None
),
# MONTHLY_DAY carrying weekday_set (folded-only column) or month.
ScheduleRuleStorage(
"monthly", 1, "1", 15, None, None, "2026-09-01", None
),
ScheduleRuleStorage(
"monthly", 1, None, 15, None, 6, "2026-09-01", None
),
# MONTHLY_WEEKDAY carrying a forbidden month.
ScheduleRuleStorage(
"monthly", 1, "1", None, 2, 6, "2026-09-01", None
),
# YEARLY carrying weekday_set or nth_weekday.
ScheduleRuleStorage(
"yearly", 1, "0", None, None, 6, "2026-09-01", None
),
ScheduleRuleStorage(
"yearly", 1, None, None, 2, 6, "2026-09-01", None
),
],
)
def test_schedule_rule_from_storage_rejects_forbidden_fields(storage) -> None:
with pytest.raises(RuleValidationError):
schedule_rule_from_storage(storage)
def test_schedule_rule_from_storage_rejects_duplicate_weekday_token() -> None:
"""A MONTHLY_WEEKDAY with a duplicate CSV token ("1,1") must raise,
never silently canonicalize to a single weekday on re-storage."""
with pytest.raises(RuleValidationError):
schedule_rule_from_storage(
ScheduleRuleStorage(
"monthly", 1, "1,1", None, 2, None, "2026-09-01", None
)
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# quest_definitions: create / get # quest_definitions: create / get
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+441
View File
@@ -7,6 +7,7 @@ import pytest
from custom_components.nestquest.presence import ( from custom_components.nestquest.presence import (
MAX_CYCLE_LENGTH_WEEKS, MAX_CYCLE_LENGTH_WEEKS,
PresenceEngine,
PresenceOverride, PresenceOverride,
PresenceSchedule, PresenceSchedule,
cycle_week_index, cycle_week_index,
@@ -371,3 +372,443 @@ def test_module_uses_no_iso_week_functions() -> None:
).read_text() ).read_text()
assert "isocalendar" not in source assert "isocalendar" not in source
assert "isoweek" not in source assert "isoweek" not in source
# ---------------------------------------------------------------------------
# PresenceEngine: is_present with the always-present default
# ---------------------------------------------------------------------------
def _three_child_engine() -> PresenceEngine:
"""Two children on opposite weeks of one two-week rotation, one
with no schedule at all (the always-present sibling)."""
anchor = datetime.date(2026, 1, 5) # Monday, week 0
declan = PresenceSchedule(
1, 2, anchor, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
)
jordyn = PresenceSchedule(
2, 2, anchor, {0: set(), 1: {0, 1, 2, 3, 4, 5, 6}}
)
return PresenceEngine({1: declan, 2: jordyn})
def test_engine_child_without_schedule_is_always_present() -> None:
"""Every day of the four-week span, all three children get the
expected verdict: Chloe (no schedule) always present; the two
rotation children complementary per the anchor pattern."""
engine = _three_child_engine()
anchor = datetime.date(2026, 1, 5)
for offset in range(28):
day = anchor + datetime.timedelta(days=offset)
week = ((day - anchor).days // 7) % 2
assert engine.is_present(3, day) is True, f"offset {offset}: Chloe"
assert engine.is_present(1, day) is (week == 0), (
f"offset {offset}: Declan should follow week {week}"
)
assert engine.is_present(2, day) is (week == 1), (
f"offset {offset}: Jordyn should follow week {week}"
)
def test_engine_alternating_children_are_complementary_for_four_weeks() -> None:
"""Every day of a four-week span: the two rotation children swap
presence exactly per the anchor pattern — when one is home the
other is away."""
engine = _three_child_engine()
anchor = datetime.date(2026, 1, 5)
for offset in range(28):
day = anchor + datetime.timedelta(days=offset)
d = engine.is_present(1, day)
j = engine.is_present(2, day)
assert d == (not j), f"offset {offset}: {d}/{j} not complementary"
expected_week = ((day - anchor).days // 7) % 2
expected_present = day.weekday() in {0, 1, 2, 3, 4, 5, 6}
assert engine.is_present(1, day) == (expected_week == 0), (
f"offset {offset}: Declan week {expected_week}"
)
# Spot-check the swap days.
assert engine.is_present(1, anchor) is True
assert engine.is_present(2, anchor) is False
assert engine.is_present(1, anchor + datetime.timedelta(days=7)) is False
assert engine.is_present(2, anchor + datetime.timedelta(days=7)) is True
def test_engine_weekend_absence_respects_the_pattern() -> None:
"""A weekday-set pattern excludes days outside the set even in the
child's present week."""
engine = PresenceEngine(
{
1: PresenceSchedule(
1, 2, datetime.date(2026, 1, 5), {0: {1, 3}, 1: {2, 4}}
)
}
)
# 2026-01-06 (Tue) is week 0, weekday 1 -> present.
assert engine.is_present(1, datetime.date(2026, 1, 6)) is True
# 2026-01-07 (Wed) is week 0, weekday 2 -> absent.
assert engine.is_present(1, datetime.date(2026, 1, 7)) is False
# 2026-01-14 (Wed) is week 1, weekday 2 -> present.
assert engine.is_present(1, datetime.date(2026, 1, 14)) is True
def test_engine_rejects_bad_construction_and_inputs() -> None:
schedule = PresenceSchedule(1, 1, ANCHOR, {0: {0}})
with pytest.raises(ValueError, match="must map child_id"):
PresenceEngine([schedule])
with pytest.raises(ValueError, match="must be a PresenceSchedule"):
PresenceEngine({1: "not-a-schedule"})
with pytest.raises(ValueError, match="holds the schedule"):
PresenceEngine({2: schedule})
with pytest.raises(ValueError, match="child_id must be an integer"):
PresenceEngine({True: schedule})
with pytest.raises(ValueError, match="child_id must be an integer"):
PresenceEngine({1.0: PresenceSchedule(1, 1, ANCHOR, {0: {0}})})
engine = PresenceEngine({1: schedule})
with pytest.raises(ValueError, match="child_id must be an integer"):
engine.is_present(True, ANCHOR)
with pytest.raises(ValueError, match="plain calendar date"):
engine.is_present(1, datetime.datetime(2026, 1, 5))
def test_engine_is_immutable() -> None:
engine = _three_child_engine()
with pytest.raises(AttributeError, match="immutable snapshot"):
engine._schedules = {}
with pytest.raises(AttributeError, match="immutable snapshot"):
engine.something_new = 1
# ---------------------------------------------------------------------------
# Overrides take precedence over the pattern
# ---------------------------------------------------------------------------
def _engine_with_overrides(*overrides_for_declan) -> PresenceEngine:
anchor = datetime.date(2026, 1, 5) # Monday
return PresenceEngine(
{
1: PresenceSchedule(
1, 2, anchor, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
)
},
overrides={1: list(overrides_for_declan)} if overrides_for_declan else {},
)
def test_absent_override_on_normally_present_day_wins() -> None:
# 2026-01-06 is Tuesday of Declan's week 0: normally present.
engine = _engine_with_overrides(
PresenceOverride(1, "2026-01-06", "2026-01-06", False, note="away")
)
assert engine.is_present(1, datetime.date(2026, 1, 6)) is False
# The pattern resumes outside the override.
assert engine.is_present(1, datetime.date(2026, 1, 7)) is True
def test_present_override_on_normally_absent_day_wins() -> None:
# 2026-01-13 is in Declan's week 1 (away all week); a present
# override on the Tuesday of week 1 brings him home for that day.
engine = _engine_with_overrides(
PresenceOverride(1, "2026-01-13", "2026-01-13", True, note="trade")
)
assert engine.is_present(1, datetime.date(2026, 1, 13)) is True
assert engine.is_present(1, datetime.date(2026, 1, 14)) is False
def test_override_applies_without_a_schedule() -> None:
"""An override on a schedule-free child still applies: holidays and
swaps must work for the always-present sibling too."""
engine = PresenceEngine(
{},
overrides={
3: [
PresenceOverride(
3, "2026-07-20", "2026-07-24", False, note="holiday"
)
]
},
)
assert engine.is_present(3, datetime.date(2026, 7, 22)) is False
assert engine.is_present(3, datetime.date(2026, 7, 19)) is True
assert engine.is_present(3, datetime.date(2026, 7, 25)) is True
def test_override_range_beats_pattern_for_its_span() -> None:
engine = _engine_with_overrides(
PresenceOverride(1, "2026-01-13", "2026-01-17", True, note="traded week")
)
for offset in range(12, 17):
day = datetime.date(2026, 1, 1) + datetime.timedelta(days=offset)
assert engine.is_present(1, day) is True, f"{day}"
# Days outside the span stay on the pattern (week 1: absent).
assert engine.is_present(1, datetime.date(2026, 1, 12)) is False
assert engine.is_present(1, datetime.date(2026, 1, 18)) is False
def test_engine_override_snapshot_validation() -> None:
schedule = PresenceSchedule(1, 1, ANCHOR, {0: {0}})
entry = PresenceOverride(1, "2026-07-20", "2026-07-20", True)
with pytest.raises(ValueError, match="must map child_id"):
PresenceEngine({1: schedule}, overrides=[entry])
with pytest.raises(ValueError, match="must be a list"):
PresenceEngine({1: schedule}, overrides={1: entry})
with pytest.raises(ValueError, match="must be"):
PresenceEngine({1: schedule}, overrides={1: ["nope"]})
with pytest.raises(ValueError, match="holds an override"):
PresenceEngine({1: schedule}, overrides={2: [entry]})
with pytest.raises(ValueError, match="child_id must be an integer"):
PresenceEngine({1: schedule}, overrides={True: [entry]})
def test_engine_overlapping_snapshot_uses_latest_start() -> None:
"""The DAO layer refuses overlapping overrides per child; if a
snapshot carries them anyway (hand-built), the latest start date
wins deterministically."""
engine = _engine_with_overrides(
PresenceOverride(1, "2026-01-06", "2026-01-10", False),
PresenceOverride(1, "2026-01-08", "2026-01-12", True),
)
# Jan 6-7: only the earlier override covers -> absent.
assert engine.is_present(1, datetime.date(2026, 1, 6)) is False
# Jan 8-10: both cover; the LATER start (present) wins.
assert engine.is_present(1, datetime.date(2026, 1, 9)) is True
# Jan 11-12: only the later override covers -> present.
assert engine.is_present(1, datetime.date(2026, 1, 11)) is True
def test_engine_override_lists_are_frozen() -> None:
"""The overrides snapshot must be truly read-only: the engine is
built directly on the CALLER's list, so mutating that list after
construction would change answers under a shallow copy — under the
frozen snapshot it must not."""
entry = PresenceOverride(1, "2026-01-06", "2026-01-06", False)
entries = [entry]
schedule = PresenceSchedule(
1, 2, ANCHOR, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
)
engine = PresenceEngine(
{1: schedule}, overrides={1: entries} # caller-owned list
)
assert engine.is_present(1, datetime.date(2026, 1, 6)) is False
entries.append(PresenceOverride(1, "2026-01-07", "2026-01-07", False))
# The late append must NOT change the engine's answers...
assert engine.is_present(1, datetime.date(2026, 1, 7)) is True
# Tuples have no append at all (AttributeError), and the mapping
# itself refuses reassignment.
with pytest.raises(AttributeError):
engine._overrides[1].append(entry)
with pytest.raises(AttributeError, match="immutable snapshot"):
engine._overrides = {}
# ---------------------------------------------------------------------------
# next_present_dates: the parent's pattern preview
# ---------------------------------------------------------------------------
def _preview_engine() -> PresenceEngine:
"""Declan: home all of week 0, away all of week 1 (anchor
2026-01-05, a Monday); an away override Tue Jan 13 (his away week)
would matter only for the override test."""
return PresenceEngine(
{
1: PresenceSchedule(
1, 2, datetime.date(2026, 1, 5),
{0: {0, 1, 2, 3, 4, 5, 6}, 1: set()},
)
}
)
def test_preview_returns_the_next_present_days_of_a_known_pattern() -> None:
engine = _preview_engine()
# Starting Monday Jan 5 (present week 0): five straight days.
preview = engine.next_present_dates(1, "2026-01-05", 5)
assert preview == [
datetime.date(2026, 1, 5),
datetime.date(2026, 1, 6),
datetime.date(2026, 1, 7),
datetime.date(2026, 1, 8),
datetime.date(2026, 1, 9),
]
# Starting mid-week: the rest of week 0 first, then the absent
# week 1 is skipped and the next present week picks up.
preview = engine.next_present_dates(1, "2026-01-07", 3)
assert preview == [
datetime.date(2026, 1, 7),
datetime.date(2026, 1, 8),
datetime.date(2026, 1, 9),
]
preview = engine.next_present_dates(1, "2026-01-07", 6)
# Week 0 runs Mon Jan 5 .. Sun Jan 11 (all present); the away week
# (Jan 12-18) is skipped; week 2 resumes Jan 19.
assert preview == [
datetime.date(2026, 1, 7),
datetime.date(2026, 1, 8),
datetime.date(2026, 1, 9),
datetime.date(2026, 1, 10),
datetime.date(2026, 1, 11),
datetime.date(2026, 1, 19),
]
def test_preview_start_date_is_inclusive() -> None:
engine = _preview_engine()
# Jan 12 (Monday of the away week) is absent: the first present day
# is Jan 19 — the start date itself when present.
preview = engine.next_present_dates(1, "2026-01-19", 1)
assert preview == [datetime.date(2026, 1, 19)]
def test_preview_honours_overrides() -> None:
schedule = PresenceSchedule(
1, 2, datetime.date(2026, 1, 5),
{0: {0, 1, 2, 3, 4, 5, 6}, 1: set()},
)
engine = PresenceEngine(
{1: schedule},
overrides={
1: [
PresenceOverride(
1, "2026-01-05", "2026-01-06", False, note="holiday"
)
]
},
)
# Jan 5-6 blocked by the override: the preview starts Jan 7.
preview = engine.next_present_dates(1, "2026-01-04", 3)
assert preview == [
datetime.date(2026, 1, 7),
datetime.date(2026, 1, 8),
datetime.date(2026, 1, 9),
]
def test_preview_schedule_free_child_returns_consecutive_days() -> None:
engine = PresenceEngine({})
preview = engine.next_present_dates(3, "2026-03-01", 3)
assert preview == [
datetime.date(2026, 3, 1),
datetime.date(2026, 3, 2),
datetime.date(2026, 3, 3),
]
def test_preview_rejects_bad_inputs() -> None:
engine = _preview_engine()
with pytest.raises(ValueError, match="count must be an integer"):
engine.next_present_dates(1, "2026-01-05", "3")
with pytest.raises(ValueError, match="count must be at least 1"):
engine.next_present_dates(1, "2026-01-05", 0)
with pytest.raises(ValueError, match="child_id must be an integer"):
engine.next_present_dates(True, "2026-01-05", 1)
with pytest.raises(ValueError, match="plain calendar date"):
engine.next_present_dates(1, datetime.datetime(2026, 1, 5), 1)
def test_preview_always_absent_child_raises_not_loops() -> None:
"""A child absent every scanned day must surface as an error, not
hang the admin preview."""
engine = PresenceEngine(
{
1: PresenceSchedule(
1, 1, datetime.date(2026, 1, 5), {0: set()}
)
}
)
with pytest.raises(ValueError, match="not present on any"):
engine.next_present_dates(1, "2026-01-05", 2)
# ---------------------------------------------------------------------------
# The week-parity regression guard (Feature 05, D-004)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("year", "length", "reason"),
[
(2020, 366, "leap year, 53 ISO weeks"),
(2015, 365, "365 days, 53 ISO weeks"),
],
)
def test_custody_rotation_walks_full_years_without_inversion(
year, length, reason
) -> None:
"""REGRESSION GUARD for the week-parity bug (D-004).
Walks a two-week alternating rotation day by day across a FULL
{length}-day year ({year} — {reason}) plus the following January,
asserting the present/absent sequence never breaks or inverts at
any point, including January 1st. ISO week parity silently flips
at the ISO-week rollover (the Monday on or after January 4th —
January 4th itself in both of these years, which end ISO week 53)
and would invert every custody schedule there; anchor-date
arithmetic must not.
""".format(length=length, year=year, reason=reason)
if year == 2020:
assert datetime.date(2020, 12, 31).isocalendar()[1] == 53
assert (datetime.date(2021, 1, 1) - datetime.date(2020, 1, 1)).days == 366
if year == 2015:
assert datetime.date(2015, 12, 31).isocalendar()[1] == 53
assert (datetime.date(2016, 1, 1) - datetime.date(2015, 1, 1)).days == 365
# The first Monday of January (2020: Jan 6; 2015: Jan 5) — an
# on-week anchor, whatever the year.
anchor = datetime.date(year, 1, 1) + datetime.timedelta(
days=(7 - datetime.date(year, 1, 1).weekday()) % 7
)
assert anchor.weekday() == 0
rotation = PresenceSchedule(
1, 2, anchor, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
)
engine = PresenceEngine({1: rotation})
# Walk EVERY day of the year. The expectation is derived from the
# anchor's 7-day BLOCKS, not from the engine: each block of seven
# days starting at the anchor is uniformly present (even block
# index) or uniformly absent (odd block index), and consecutive
# blocks alternate. A parity-based implementation resets on
# January 1st and breaks either the uniformity of a block or the
# alternation between them — most visibly in the block spanning
# December 31st -> January 1st of a 53-ISO-week year.
# Walk the FULL year AND the first two weeks of the following
# January: the block spanning December 31st -> January 1st must
# stay uniform ACROSS the boundary, so an implementation that
# resets when the target year differs from the anchor's fails
# here (the post-boundary days of its final block disagree).
first_day = datetime.date(year, 1, 1)
last_day = datetime.date(year + 1, 1, 14)
days_seen = 0
day = first_day
while day <= last_day:
days_since_anchor = (day - anchor).days
block = days_since_anchor // 7 # which 7-day block of the cycle
block_start = anchor + datetime.timedelta(days=block * 7)
# Uniformity within the block: every day of this 7-day block
# must answer exactly like the block's first day.
expected = ((block % 2) == 0)
assert engine.is_present(1, day) is expected, (
f"{day} (block {block}, day offset "
f"{(day - block_start).days}): present="
f"{engine.is_present(1, day)}, expected {expected}"
)
# The block's OTHER days must agree — including the days on
# the far side of January 1st.
for offset in range(7):
block_day = block_start + datetime.timedelta(days=offset)
if first_day <= block_day <= last_day:
assert engine.is_present(1, block_day) is expected, (
f"{block_day} disagrees with its block {block}"
)
days_seen += 1
day += datetime.timedelta(days=1)
# The walked year itself contributed exactly its full length.
year_days = sum(
1
for offset in range((last_day - first_day).days + 1)
if (first_day + datetime.timedelta(days=offset)).year == year
)
assert year_days == length, (
f"walked {year_days} days of {year}, expected {length}"
)