Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
98e89f8b29 | ||
|
|
e92c79b211 | ||
|
|
58b11b59cf | ||
|
|
9d21f29eea | ||
|
|
fec607f039 | ||
|
|
a18c0060e4 | ||
|
|
20d7efb04b | ||
|
|
6551a87bd6 | ||
|
|
e9dc0a1445 | ||
|
|
1bb0617471 | ||
|
|
221ab2c20b | ||
|
|
bdf3513609 | ||
|
|
830d663a45 | ||
|
|
bff2f4ff5e | ||
|
|
8f04d93363 | ||
|
|
d9326deeb5 | ||
|
|
94f54523f9 | ||
|
|
51b1388233 | ||
|
|
744a40dcb5 | ||
|
|
e15648232f | ||
|
|
6d48f433b7 | ||
|
|
540f59a353 | ||
|
|
ee994a3938 | ||
|
|
7ae4018e59 |
@@ -186,6 +186,15 @@ class PresenceOverridesDao:
|
||||
The child must exist (validated atomically under the
|
||||
connection lock). Date shape is validated strictly, and the
|
||||
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(end_date, "end_date")
|
||||
@@ -196,6 +205,29 @@ class PresenceOverridesDao:
|
||||
)
|
||||
if child is None:
|
||||
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(
|
||||
"INSERT INTO presence_overrides (child_id, "
|
||||
"start_date, end_date, is_present, note) "
|
||||
|
||||
@@ -40,6 +40,7 @@ from .const import QUEST_WINDOWS
|
||||
from .dao_children import ChildRecord, _child_from_row, _connection_lock
|
||||
from .dao_children import _CHILD_COLUMNS
|
||||
from .db import NestQuestDatabase
|
||||
from .recurrence import RuleType, RuleValidationError, ScheduleRule
|
||||
|
||||
#: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
|
||||
#: in update methods: passing ``None`` must mean clearing a nullable
|
||||
@@ -119,6 +120,26 @@ class ScheduleRuleRecord:
|
||||
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)
|
||||
class QuestDefinitionRecord:
|
||||
"""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:
|
||||
"""Typed async access to the ``schedule_rules`` table."""
|
||||
|
||||
|
||||
@@ -317,3 +317,161 @@ def cycle_week_index(
|
||||
target = _parse_date(target_date, "target_date")
|
||||
days = (target - schedule.anchor_date).days
|
||||
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"
|
||||
)
|
||||
|
||||
+170
-1
@@ -619,4 +619,173 @@ def test_dao_presence_test_file_uses_dao_not_raw_table_sql() -> None:
|
||||
)
|
||||
assert sql_pattern.search(remaining) is None, (
|
||||
"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())
|
||||
|
||||
@@ -9,13 +9,21 @@ import pytest
|
||||
from custom_components.nestquest.dao_children import ChildrenDao
|
||||
from custom_components.nestquest.dao_rules import (
|
||||
ScheduleRuleRecord,
|
||||
ScheduleRuleStorage,
|
||||
ScheduleRulesDao,
|
||||
QuestDefinitionRecord,
|
||||
QuestDefinitionWindowRecord,
|
||||
QuestDefinitionsDao,
|
||||
schedule_rule_from_storage,
|
||||
schedule_rule_to_storage,
|
||||
)
|
||||
from custom_components.nestquest.db import NestQuestDatabase
|
||||
from custom_components.nestquest.migrations import apply_migrations
|
||||
from custom_components.nestquest.recurrence import (
|
||||
RuleType,
|
||||
RuleValidationError,
|
||||
ScheduleRule,
|
||||
)
|
||||
|
||||
|
||||
def _run(coro):
|
||||
@@ -423,6 +431,249 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
|
||||
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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -7,6 +7,7 @@ import pytest
|
||||
|
||||
from custom_components.nestquest.presence import (
|
||||
MAX_CYCLE_LENGTH_WEEKS,
|
||||
PresenceEngine,
|
||||
PresenceOverride,
|
||||
PresenceSchedule,
|
||||
cycle_week_index,
|
||||
@@ -371,3 +372,443 @@ def test_module_uses_no_iso_week_functions() -> None:
|
||||
).read_text()
|
||||
assert "isocalendar" 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}"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user