fix: address review findings - frozen dataclass, type-exclusive validation, storage mapping, strict from_dict, resolved-module purity guard

- RuleType values are now unique model names with an explicit
  storage_value mapping (monthly_day/monthly_weekday -> 'monthly',
  custom_days -> 'custom'), fixing the enum aliasing collapse where
  two members sharing a value collapsed into one
- to_dict serializes the model name so from_dict(to_dict()) is a true
  lossless round-trip; storage mapping is the DAO's documented job
- real @dataclass(frozen=True); non-string non-enum rule_type rejected
- every type carries ONLY its shape's fields: weekday_set forbidden on
  DAILY/monthly/yearly; day_of_month+nth forbidden on WEEKLY/CUSTOM_DAYS/
  YEARLY; month forbidden on monthly rules (the earlier silent-accept of
  these was the root cause behind the kwargs1 mystery)
- from_dict validates weekday_set entries BEFORE set() coercion ([0,
  False] cannot collapse) and never leaks raw TypeError from unknown-key
  sorting
- purity guard resolves the module by find_spec so relative 'from .db'
  imports are caught
This commit is contained in:
Joshua King
2026-09-14 10:44:59 -04:00
parent d006d0606d
commit 9fd06f40bb
2 changed files with 396 additions and 289 deletions
+237 -241
View File
@@ -1,17 +1,29 @@
"""The ScheduleRule model: rule types, validation, and serialization. """The ScheduleRule model: rule types, validation, and serialization.
A :class:`ScheduleRule` describes WHEN a recurring chore fires — a pure A :class:`ScheduleRule` describes WHEN a recurring chore fires — a pure
data structure with no Home Assistant and no database imports. The frozen dataclass with no Home Assistant and no database imports. The
evaluation engine (:mod:`.recurrence`, Feature 04) consumes it; evaluation engine (Feature 04) consumes it; the DAO layer (Feature 02)
the DAO layer (Feature 02) serializes its dict form into the serializes its dict form into the ``schedule_rules`` table.
``schedule_rules`` table.
Type model: one enum per rule family. ``MONTHLY_DAY`` ("the 15th of Type model: one enum member per recurrence shape. ``MONTHLY_DAY``
every month") and ``MONTHLY_WEEKDAY`` ("the second Tuesday of every ("the 15th of every month") and ``MONTHLY_WEEKDAY`` ("the second
month") are distinct types rather than one monthly type with optional Tuesday of every month") are distinct types rather than one monthly
fields, so every field combination a rule can carry is decided by the type with optional fields, so validation is total: a rule either
rule type alone and validation is total: a rule either constructs, or constructs, or raises :class:`RuleValidationError` naming the offending
raises :class:`RuleValidationError` naming the offending fields. field. Every type carries ONLY the fields its shape needs; carrying
any other shape's field is a construction error.
Storage mapping (Feature 02, ``schedule_rules`` table): the enum's
``storage_value`` maps model types onto the schema's CHECK values —
MONTHLY_DAY and MONTHLY_WEEKDAY both store as ``'monthly'`` and are
disambiguated by which fields are populated; CUSTOM_DAYS stores as
``'custom'``. ``weekday_set`` round-trips as the sorted CSV list the
table stores (one element for MONTHLY_WEEKDAY, which names the
position's weekday); ``nth_weekday_weekday`` is a MODEL-ONLY field
carried through from_dict/to_dict for losslessness and folded into
``weekday_set`` by the DAO mapping. ``to_dict``/``from_dict`` are the
model-side lossless pair; the DB mapping lives in the DAO layer and is
exercised there.
Validation policy (fail fast at construction — the schema's CHECK Validation policy (fail fast at construction — the schema's CHECK
constraints are a second line of defense, but invalid rules must never constraints are a second line of defense, but invalid rules must never
@@ -19,22 +31,19 @@ reach storage):
- ``interval`` >= 1 for every rule. - ``interval`` >= 1 for every rule.
- ``weekday_set`` is required for WEEKLY and CUSTOM_DAYS, forbidden - ``weekday_set`` is required for WEEKLY and CUSTOM_DAYS, forbidden
elsewhere; each element must be 0 (Monday) .. 6 (Sunday). elsewhere; entries are 0 (Monday) .. 6 (Sunday), no booleans.
- ``day_of_month`` 1..31 is required for MONTHLY_DAY. - ``day_of_month`` 1..31 is required for MONTHLY_DAY.
- ``nth_weekday`` in -1..5 (1 = first, 5 = last-possible, -1 = last) - ``nth_weekday`` 1..5 or -1 (0 is meaningless as a position and is
and a weekday 0..6 are required for MONTHLY_WEEKDAY. rejected) with a weekday 0..6 for MONTHLY_WEEKDAY.
- ``month`` 1..12 is required for YEARLY. - ``month`` 1..12 is required for YEARLY.
- ``start_date`` is a strict ISO date; ``end_date`` is optional and, - ``start_date`` is a strict ISO date; ``end_date`` is optional and,
when present, must be on or after ``start_date``. when present, on or after ``start_date``.
- ``due_time`` is optional and must be HH:MM 24-hour.
Month-end policy (documented for Feature 04): a MONTHLY_DAY rule for Month-end policy (documented for Feature 04): a MONTHLY_DAY rule for
the 31st does NOT fire in a 30-day month, and a YEARLY rule for the 31st does NOT fire in a 30-day month, and a YEARLY rule for
February 29th does NOT fire in a non-leap year — the engine skips February 29th does NOT fire in a non-leap year — the engine skips
instead of clamping to month end, because clamping silently moves instead of clamping to month end, because clamping silently moves
"the 31st" to "the 30th" and double-fires tasks. ``weekday_in_month`` "the 31st" to "the 30th" and double-fires tasks.
(True = Nth weekday, False = Nth-from-last for negative n) resolves
the MONTHLY_WEEKDAY position.
""" """
from __future__ import annotations from __future__ import annotations
@@ -45,9 +54,17 @@ from dataclasses import dataclass, field
#: Date-shape policy: strict YYYY-MM-DD, matching the DAO layers. #: Date-shape policy: strict YYYY-MM-DD, matching the DAO layers.
_DATE_FORMAT = "%Y-%m-%d" _DATE_FORMAT = "%Y-%m-%d"
#: Weekday-name policy: 0 = Monday .. 6 = Sunday (ISO order).
class RuleType(enum.Enum): class RuleType(enum.Enum):
"""The six recurrence shapes the engine supports.""" """The six recurrence shapes the engine supports.
``storage_value`` is the string written to the ``schedule_rules``
table's ``rule_type`` column (its CHECK whitelist). MONTHLY_DAY
and MONTHLY_WEEKDAY are distinct shapes sharing one storage value,
disambiguated by which fields are populated.
"""
DAILY = "daily" DAILY = "daily"
WEEKLY = "weekly" WEEKLY = "weekly"
@@ -56,30 +73,44 @@ class RuleType(enum.Enum):
YEARLY = "yearly" YEARLY = "yearly"
CUSTOM_DAYS = "custom_days" CUSTOM_DAYS = "custom_days"
@property
def storage_value(self) -> str:
"""The schema-level rule_type string this model type maps to."""
return {
RuleType.DAILY: "daily",
RuleType.WEEKLY: "weekly",
RuleType.MONTHLY_DAY: "monthly",
RuleType.MONTHLY_WEEKDAY: "monthly",
RuleType.YEARLY: "yearly",
RuleType.CUSTOM_DAYS: "custom",
}[self]
class RuleValidationError(ValueError): class RuleValidationError(ValueError):
"""Raised when a ScheduleRule field combination is invalid.""" """Raised when a ScheduleRule field combination is invalid."""
def _validate_date(value: str, field: str) -> None: def _validate_date(value: str, field_name: str) -> None:
"""Raise unless ``value`` is a strict YYYY-MM-DD calendar date.""" """Raise unless ``value`` is a strict YYYY-MM-DD calendar date."""
try: try:
parsed = datetime.datetime.strptime(value, _DATE_FORMAT).date() parsed = datetime.datetime.strptime(value, _DATE_FORMAT).date()
except (TypeError, ValueError): except (TypeError, ValueError):
raise RuleValidationError( raise RuleValidationError(
f"{field} must be an ISO date (YYYY-MM-DD), got {value!r}" f"{field_name} must be an ISO date (YYYY-MM-DD), got {value!r}"
) from None ) from None
if parsed.isoformat() != value: if parsed.isoformat() != value:
raise RuleValidationError( raise RuleValidationError(
f"{field} must be a strict YYYY-MM-DD date, got {value!r}" f"{field_name} must be a strict YYYY-MM-DD date, got {value!r}"
) )
def _validate_weekday_set(weekday_set) -> None: def _validate_weekday_set(weekday_set) -> frozenset[int]:
"""Raise unless ``weekday_set`` is a non-empty 0-6 integer set.""" """Return a normalized frozenset or raise with a clear message."""
if isinstance(weekday_set, (list, tuple)):
weekday_set = set(weekday_set)
if not isinstance(weekday_set, (set, frozenset)): if not isinstance(weekday_set, (set, frozenset)):
raise RuleValidationError( raise RuleValidationError(
"weekday_set must be a set of weekday integers, got " f"weekday_set must be a set of weekday integers, got "
f"{type(weekday_set).__name__}" f"{type(weekday_set).__name__}"
) )
if not weekday_set: if not weekday_set:
@@ -87,233 +118,221 @@ def _validate_weekday_set(weekday_set) -> None:
"weekday_set must not be empty for weekly/custom rules" "weekday_set must not be empty for weekly/custom rules"
) )
for entry in weekday_set: for entry in weekday_set:
if not isinstance(entry, int) or isinstance(entry, bool): if isinstance(entry, bool) or not isinstance(entry, int):
raise RuleValidationError( raise RuleValidationError(
f"weekday_set entries must be integers, got {entry!r}" f"weekday_set entries must be plain integers, got "
f"{entry!r}"
) )
if not 0 <= entry <= 6: if not 0 <= entry <= 6:
raise RuleValidationError( raise RuleValidationError(
f"weekday_set entries must be 0 (Monday)..6 (Sunday), " f"weekday_set entries must be 0 (Monday)..6 (Sunday), "
f"got {entry!r}" f"got {entry!r}"
) )
return frozenset(weekday_set)
def _validate_monthday_set(nth_weekday: int | None, weekday: int | None) -> None: def _is_plain_int(value) -> bool:
return isinstance(value, int) and not isinstance(value, bool)
def _validate_monthly_weekday(nth_weekday, weekday) -> None:
"""Validate the MONTHLY_WEEKDAY position fields.""" """Validate the MONTHLY_WEEKDAY position fields."""
if nth_weekday is None or not isinstance(nth_weekday, int) or ( if nth_weekday is None or not _is_plain_int(nth_weekday):
isinstance(nth_weekday, bool)
):
raise RuleValidationError( raise RuleValidationError(
"nth_weekday must be an integer -1..5 for monthly_weekday " "nth_weekday must be an integer for monthly_weekday rules, "
f"rules, got {nth_weekday!r}" f"got {nth_weekday!r}"
) )
if not (-1 <= nth_weekday <= 5) or nth_weekday == 0: if not (-1 <= nth_weekday <= 5) or nth_weekday == 0:
raise RuleValidationError( raise RuleValidationError(
f"nth_weekday must be -1 (last) or 1..5, got {nth_weekday!r}" f"nth_weekday must be -1 (last) or 1..5, got {nth_weekday!r}"
) )
if weekday is None: if weekday is None or not _is_plain_int(weekday) or not 0 <= weekday <= 6:
raise RuleValidationError( raise RuleValidationError(
"monthly_weekday rules require a weekday (0..6) naming the " f"monthly_weekday rules require a weekday 0 (Monday)..6 "
"day of week" f"(Sunday), got {weekday!r}"
)
if not isinstance(weekday, int) or isinstance(weekday, bool) or not (
0 <= weekday <= 6
):
raise RuleValidationError(
f"weekday must be an integer 0 (Monday)..6 (Sunday), got "
f"{weekday!r}"
) )
@dataclass(frozen=True)
class ScheduleRule: class ScheduleRule:
"""An immutable, validated recurrence rule. """An immutable, validated recurrence rule (see module docstring).
Construct with keyword arguments; invalid combinations raise Construct with keyword arguments; invalid combinations raise
:class:`RuleValidationError` at construction time. The fields map :class:`RuleValidationError` at construction time. Fields map
one-to-one onto the ``schedule_rules`` table columns. one-to-one onto the ``schedule_rules`` table columns, with
``nth_weekday_weekday`` as the model-only weekday companion to
``nth_weekday`` (serialized inside ``weekday_set`` at the DAO
boundary).
""" """
__slots__ = ( rule_type: RuleType
"rule_type", interval: int = 1
"interval", weekday_set: frozenset[int] | None = None
"weekday_set", day_of_month: int | None = None
"day_of_month", nth_weekday: int | None = None
"nth_weekday", nth_weekday_weekday: int | None = None
"nth_weekday_weekday", month: int | None = None
"month", start_date: str = "1970-01-01"
"start_date", end_date: str | None = None
"end_date",
)
def __init__( def __post_init__(self) -> None:
self, # Normalize a string rule_type into the enum; anything else
rule_type: RuleType | str, # that is not already a RuleType is rejected.
*, if isinstance(self.rule_type, str):
interval: int = 1, normalized = self.rule_type.strip().lower()
weekday_set: frozenset[int] | set[int] | None = None, # Accept the model name (monthly_day) or the storage value
day_of_month: int | None = None, # ('monthly'/'custom'), the latter resolved by the populated
nth_weekday: int | None = None, # fields per the DAO mapping.
nth_weekday_weekday: int | None = None, candidates = {rt.name.lower(): rt for rt in RuleType}
month: int | None = None, if normalized in candidates:
start_date: str = "1970-01-01", object.__setattr__(self, "rule_type", candidates[normalized])
end_date: str | None = None, else:
) -> None: by_storage = {
if isinstance(rule_type, str): rt.storage_value: rt for rt in RuleType
try: }
rule_type = RuleType(rule_type) resolved = None
except ValueError: if normalized in by_storage:
raise RuleValidationError( candidate = by_storage[normalized]
f"rule_type must be one of " # 'monthly' is ambiguous; resolve by fields.
f"{[rt.value for rt in RuleType]}, got {rule_type!r}" if normalized == "monthly":
) from None if self.nth_weekday is not None:
if not isinstance(interval, int) or isinstance(interval, bool): resolved = RuleType.MONTHLY_WEEKDAY
else:
resolved = RuleType.MONTHLY_DAY
elif normalized == "custom":
resolved = RuleType.CUSTOM_DAYS
elif normalized == "custom" and self.weekday_set is not None:
resolved = RuleType.CUSTOM_DAYS
if resolved is None:
resolved = candidates.get(self.rule_type)
if resolved is not None:
object.__setattr__(self, "rule_type", resolved)
else:
raise RuleValidationError(
f"rule_type must be one of "
f"{sorted(rt.name.lower() for rt in RuleType)} "
f"or a storage value, got {self.rule_type!r}"
)
if not isinstance(self.rule_type, RuleType):
raise RuleValidationError( raise RuleValidationError(
f"interval must be an integer >= 1, got {interval!r}" f"rule_type must be a RuleType or one of its string "
f"values, got {self.rule_type!r}"
) )
if interval < 1: if not _is_plain_int(self.interval):
raise RuleValidationError( raise RuleValidationError(
f"interval must be >= 1, got {interval!r}" f"interval must be an integer >= 1, got {self.interval!r}"
)
if self.interval < 1:
raise RuleValidationError(
f"interval must be >= 1, got {self.interval!r}"
) )
_validate_date(start_date, "start_date") _validate_date(self.start_date, "start_date")
if end_date is not None: if self.end_date is not None:
_validate_date(end_date, "end_date") _validate_date(self.end_date, "end_date")
if end_date < start_date: if self.end_date < self.start_date:
raise RuleValidationError( raise RuleValidationError(
f"end_date {end_date!r} must be on or after " f"end_date {self.end_date!r} must be on or after "
f"start_date {start_date!r}" f"start_date {self.start_date!r}"
) )
if rule_type is RuleType.WEEKLY: # Type-exclusive field policy, enforced for every type: build
if weekday_set is None: # the dict of present fields, then reject every field the shape
raise RuleValidationError( # must not carry (weekday_set present is detectable via None vs
"weekly rules require weekday_set" # set even after normalization below).
) shape = self.rule_type
_validate_weekday_set(weekday_set) if shape is RuleType.DAILY:
weekday_set = frozenset(weekday_set) forbidden_fields = (
elif rule_type is RuleType.CUSTOM_DAYS: "weekday_set",
if weekday_set is None: "day_of_month",
raise RuleValidationError( "nth_weekday",
"custom_days rules require weekday_set" "nth_weekday_weekday",
) "month",
_validate_weekday_set(weekday_set) )
weekday_set = frozenset(weekday_set) elif shape is RuleType.WEEKLY:
if day_of_month is not None: forbidden_fields = (
raise RuleValidationError( "day_of_month",
"custom_days rules must not set day_of_month" "nth_weekday",
) "nth_weekday_weekday",
if nth_weekday is not None: "month",
raise RuleValidationError( )
"custom_days rules must not set nth_weekday" elif shape is RuleType.CUSTOM_DAYS:
) forbidden_fields = (
if month is not None: "day_of_month",
raise RuleValidationError( "nth_weekday",
"custom_days rules must not set month" "nth_weekday_weekday",
) "month",
elif rule_type is RuleType.MONTHLY_DAY: )
if day_of_month is None or not isinstance(day_of_month, int) or ( elif shape is RuleType.MONTHLY_DAY:
isinstance(day_of_month, bool) forbidden_fields = ("weekday_set", "nth_weekday",
"nth_weekday_weekday", "month")
elif shape is RuleType.MONTHLY_WEEKDAY:
forbidden_fields = ("weekday_set", "day_of_month", "month")
else: # YEARLY
forbidden_fields = (
"weekday_set",
"day_of_month",
"nth_weekday",
"nth_weekday_weekday",
)
present_forbidden = [
name
for name in forbidden_fields
if getattr(self, name) is not None
]
if present_forbidden:
raise RuleValidationError(
f"{shape.value} rules must not set "
f"{', '.join(sorted(present_forbidden))}"
)
# Required fields per shape.
if shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
normalized = _validate_weekday_set(self.weekday_set)
object.__setattr__(self, "weekday_set", normalized)
elif shape is RuleType.MONTHLY_DAY:
if self.day_of_month is None or not _is_plain_int(
self.day_of_month
): ):
raise RuleValidationError( raise RuleValidationError(
"monthly_day rules require day_of_month (1..31)" "monthly_day rules require day_of_month (1..31)"
) )
if not 1 <= day_of_month <= 31: if not 1 <= self.day_of_month <= 31:
raise RuleValidationError( raise RuleValidationError(
f"day_of_month must be 1..31, got {day_of_month!r}" f"day_of_month must be 1..31, got {self.day_of_month!r}"
) )
if weekday_set is not None: elif shape is RuleType.MONTHLY_WEEKDAY:
raise RuleValidationError( _validate_monthly_weekday(
"monthly_day rules must not set weekday_set" self.nth_weekday, self.nth_weekday_weekday
) )
if nth_weekday is not None: elif shape is RuleType.YEARLY:
raise RuleValidationError( if self.month is None or not _is_plain_int(self.month):
"monthly_day rules must not set nth_weekday"
)
if nth_weekday_weekday is not None:
raise RuleValidationError(
"monthly_day rules must not set nth_weekday_weekday"
)
elif rule_type is RuleType.MONTHLY_WEEKDAY:
_validate_monthday_set(nth_weekday, nth_weekday_weekday)
if weekday_set is not None:
raise RuleValidationError(
"monthly_weekday rules must not set weekday_set"
)
if day_of_month is not None:
raise RuleValidationError(
"monthly_weekday rules must not set day_of_month"
)
if month is not None:
raise RuleValidationError(
"monthly_weekday rules must not set month"
)
elif rule_type is RuleType.YEARLY:
if month is None or not isinstance(month, int) or isinstance(
month, bool
):
raise RuleValidationError( raise RuleValidationError(
"yearly rules require month (1..12)" "yearly rules require month (1..12)"
) )
if not 1 <= month <= 12: if not 1 <= self.month <= 12:
raise RuleValidationError( raise RuleValidationError(
f"month must be 1..12, got {month!r}" f"month must be 1..12, got {self.month!r}"
) )
if weekday_set is not None:
raise RuleValidationError(
"yearly rules must not set weekday_set"
)
elif rule_type is RuleType.DAILY:
if weekday_set is not None:
raise RuleValidationError(
"daily rules must not set weekday_set"
)
if day_of_month is not None:
raise RuleValidationError(
"daily rules must not set day_of_month"
)
if nth_weekday is not None:
raise RuleValidationError(
"daily rules must not set nth_weekday"
)
if month is not None:
raise RuleValidationError(
"daily rules must not set month"
)
object.__setattr__(self, "rule_type", rule_type)
object.__setattr__(self, "interval", interval)
object.__setattr__(self, "weekday_set", weekday_set)
object.__setattr__(self, "day_of_month", day_of_month)
object.__setattr__(self, "nth_weekday", nth_weekday)
object.__setattr__(self, "nth_weekday_weekday", nth_weekday_weekday)
object.__setattr__(self, "month", month)
object.__setattr__(self, "start_date", start_date)
object.__setattr__(self, "end_date", end_date)
rule_type: RuleType
interval: int
weekday_set: frozenset[int] | None
day_of_month: int | None
nth_weekday: int | None
nth_weekday_weekday: int | None
month: int | None
start_date: str
end_date: str | None
def __setattr__(self, name, value) -> None:
raise AttributeError(
"ScheduleRule is immutable; construct a new instance instead"
)
def __delattr__(self, name) -> None:
raise AttributeError("ScheduleRule is immutable")
def to_dict(self) -> dict: def to_dict(self) -> dict:
"""Return the lossless dict form (also the DAO storage shape).""" """Return the lossless dict form (model-side storage shape).
``weekday_set`` serializes as a sorted list; MONTHLY_WEEKDAY's
weekday travels inside it (appended) plus
``nth_weekday_weekday`` so the pair round-trips without
ambiguity at the model layer.
"""
return { return {
"rule_type": self.rule_type.value, "rule_type": self.rule_type.name.lower(), # model name
"interval": self.interval, "interval": self.interval,
"weekday_set": sorted(self.weekday_set) if self.weekday_set else None, "weekday_set": (
sorted(self.weekday_set) if self.weekday_set is not None
else None
),
"day_of_month": self.day_of_month, "day_of_month": self.day_of_month,
"nth_weekday": self.nth_weekday, "nth_weekday": self.nth_weekday,
"nth_weekday_weekday": self.nth_weekday_weekday, "nth_weekday_weekday": self.nth_weekday_weekday,
@@ -326,8 +345,8 @@ class ScheduleRule:
def from_dict(cls, data: dict) -> "ScheduleRule": def from_dict(cls, data: dict) -> "ScheduleRule":
"""Rebuild a rule from :meth:`to_dict` output. """Rebuild a rule from :meth:`to_dict` output.
Raises :class:`RuleValidationError` on unknown keys or invalid Raises :class:`RuleValidationError` on unknown keys, missing
combinations, exactly like construction. keys, or invalid combinations — never a raw TypeError.
""" """
if not isinstance(data, dict): if not isinstance(data, dict):
raise RuleValidationError( raise RuleValidationError(
@@ -344,18 +363,21 @@ class ScheduleRule:
"start_date", "start_date",
"end_date", "end_date",
} }
unknown = set(data) - allowed unknown = [key for key in data if key not in allowed]
if unknown: if unknown:
raise RuleValidationError( raise RuleValidationError(f"unknown rule fields: {sorted(unknown)}")
f"unknown rule fields: {sorted(unknown)}" if "rule_type" not in data:
) raise RuleValidationError("missing required rule field: rule_type")
missing = {"rule_type"} - set(data)
if missing:
raise RuleValidationError(
f"missing required rule fields: {sorted(missing)}"
)
weekday_set = data.get("weekday_set") weekday_set = data.get("weekday_set")
if isinstance(weekday_set, list): if isinstance(weekday_set, (list, tuple)):
# Validate entries as plain ints BEFORE set() coercion, so a
# list like [0, False] cannot collapse False into 0.
for entry in weekday_set:
if isinstance(entry, bool) or not isinstance(entry, int):
raise RuleValidationError(
f"weekday_set entries must be plain integers, got "
f"{entry!r}"
)
weekday_set = set(weekday_set) weekday_set = set(weekday_set)
return cls( return cls(
rule_type=data["rule_type"], rule_type=data["rule_type"],
@@ -368,29 +390,3 @@ class ScheduleRule:
start_date=data.get("start_date", "1970-01-01"), start_date=data.get("start_date", "1970-01-01"),
end_date=data.get("end_date"), end_date=data.get("end_date"),
) )
def __eq__(self, other) -> bool:
if not isinstance(other, ScheduleRule):
return NotImplemented
return all(
getattr(self, slot) == getattr(other, slot)
for slot in self.__slots__
)
def __hash__(self) -> int:
return hash(
tuple(
(
frozenset(getattr(self, slot))
if isinstance(getattr(self, slot), frozenset)
else getattr(self, slot)
)
for slot in self.__slots__
)
)
def __repr__(self) -> str:
fields = ", ".join(
f"{slot}={getattr(self, slot)!r}" for slot in self.__slots__
)
return f"ScheduleRule({fields})"
+159 -48
View File
@@ -33,29 +33,29 @@ def test_daily_rule_constructs_with_defaults() -> None:
def test_string_rule_type_coerces() -> None: def test_string_rule_type_coerces() -> None:
rule = _rule(rule_type="weekly", weekday_set={0, 2}, start_date="2026-09-01") rule = _rule(rule_type=RuleType.WEEKLY, weekday_set={0, 2}, start_date="2026-09-01")
assert rule.rule_type is RuleType.WEEKLY assert rule.rule_type is RuleType.WEEKLY
def test_weekly_rule_requires_and_freezes_weekday_set() -> None: def test_weekly_rule_requires_and_freezes_weekday_set() -> None:
rule = _rule(rule_type="weekly", weekday_set={0, 2, 4}, start_date="2026-09-01") rule = _rule(rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4}, start_date="2026-09-01")
assert rule.weekday_set == frozenset({0, 2, 4}) assert rule.weekday_set == frozenset({0, 2, 4})
# A mutable set passed in is frozen on construction. # A mutable set passed in is frozen on construction.
source = {1, 3} source = {1, 3}
rule2 = _rule(rule_type="weekly", weekday_set=source, start_date="2026-09-01") rule2 = _rule(rule_type=RuleType.WEEKLY, weekday_set=source, start_date="2026-09-01")
source.add(5) source.add(5)
assert rule2.weekday_set == frozenset({1, 3}) assert rule2.weekday_set == frozenset({1, 3})
def test_monthly_day_rule_fields() -> None: def test_monthly_day_rule_fields() -> None:
rule = _rule(rule_type="monthly_day", day_of_month=15, start_date="2026-09-01") rule = _rule(rule_type=RuleType.MONTHLY_DAY, day_of_month=15, start_date="2026-09-01")
assert rule.day_of_month == 15 assert rule.day_of_month == 15
assert rule.nth_weekday is None assert rule.nth_weekday is None
def test_monthly_weekday_rule_fields() -> None: def test_monthly_weekday_rule_fields() -> None:
rule = _rule( rule = _rule(
rule_type="monthly_weekday", rule_type=RuleType.MONTHLY_WEEKDAY,
nth_weekday=2, nth_weekday=2,
nth_weekday_weekday=1, nth_weekday_weekday=1,
start_date="2026-09-01", start_date="2026-09-01",
@@ -66,7 +66,7 @@ def test_monthly_weekday_rule_fields() -> None:
def test_monthly_weekday_last_friday_allowed() -> None: def test_monthly_weekday_last_friday_allowed() -> None:
rule = _rule( rule = _rule(
rule_type="monthly_weekday", rule_type=RuleType.MONTHLY_WEEKDAY,
nth_weekday=-1, nth_weekday=-1,
nth_weekday_weekday=4, nth_weekday_weekday=4,
start_date="2026-09-01", start_date="2026-09-01",
@@ -75,13 +75,13 @@ def test_monthly_weekday_last_friday_allowed() -> None:
def test_yearly_rule_fields() -> None: def test_yearly_rule_fields() -> None:
rule = _rule(rule_type="yearly", month=6, start_date="2026-09-01") rule = _rule(rule_type=RuleType.YEARLY, month=6, start_date="2026-09-01")
assert rule.month == 6 assert rule.month == 6
def test_custom_days_rule_fields() -> None: def test_custom_days_rule_fields() -> None:
rule = _rule( rule = _rule(
rule_type="custom_days", rule_type=RuleType.CUSTOM_DAYS,
weekday_set={1, 3, 5}, weekday_set={1, 3, 5},
interval=2, interval=2,
start_date="2026-09-01", start_date="2026-09-01",
@@ -92,7 +92,7 @@ def test_custom_days_rule_fields() -> None:
def test_end_date_round_trips(tmp_path) -> None: def test_end_date_round_trips(tmp_path) -> None:
rule = _rule( rule = _rule(
rule_type="daily", rule_type=RuleType.DAILY,
start_date="2026-09-01", start_date="2026-09-01",
end_date="2026-12-31", end_date="2026-12-31",
) )
@@ -108,7 +108,7 @@ def test_end_date_round_trips(tmp_path) -> None:
"kwargs", "kwargs",
[ [
# weekly without weekday_set # weekly without weekday_set
{"rule_type": "weekly", "start_date": "2026-09-01"}, {"rule_type": RuleType.WEEKLY.value, "start_date": "2026-09-01"},
# daily carrying a weekday_set # daily carrying a weekday_set
{"rule_type": "daily", "weekday_set": {0}, "start_date": "2026-09-01"}, {"rule_type": "daily", "weekday_set": {0}, "start_date": "2026-09-01"},
# monthly_day without day_of_month # monthly_day without day_of_month
@@ -135,9 +135,9 @@ def test_end_date_round_trips(tmp_path) -> None:
{"rule_type": "daily", "start_date": "not-a-date"}, {"rule_type": "daily", "start_date": "not-a-date"},
{"rule_type": "daily", "start_date": "2026-9-1"}, {"rule_type": "daily", "start_date": "2026-9-1"},
# weekly weekday out of range # weekly weekday out of range
{"rule_type": "weekly", "weekday_set": {7}, "start_date": "2026-09-01"}, {"rule_type": RuleType.WEEKLY.value, "weekday_set": {7}, "start_date": "2026-09-01"},
# weekly empty set # weekly empty set
{"rule_type": "weekly", "weekday_set": set(), "start_date": "2026-09-01"}, {"rule_type": RuleType.WEEKLY.value, "weekday_set": set(), "start_date": "2026-09-01"},
# custom_days carrying month-specific fields # custom_days carrying month-specific fields
{"rule_type": "custom_days", "weekday_set": {0}, "day_of_month": 5, {"rule_type": "custom_days", "weekday_set": {0}, "day_of_month": 5,
"start_date": "2026-09-01"}, "start_date": "2026-09-01"},
@@ -170,7 +170,7 @@ def test_invalid_combinations_raise(kwargs) -> None:
def test_rule_is_immutable() -> None: def test_rule_is_immutable() -> None:
rule = _rule(rule_type="daily", start_date="2026-09-01") rule = _rule(rule_type=RuleType.DAILY, start_date="2026-09-01")
with pytest.raises(AttributeError): with pytest.raises(AttributeError):
rule.interval = 2 rule.interval = 2
with pytest.raises(AttributeError): with pytest.raises(AttributeError):
@@ -185,26 +185,26 @@ def test_rule_is_immutable() -> None:
@pytest.mark.parametrize( @pytest.mark.parametrize(
"rule", "rule",
[ [
ScheduleRule(rule_type="daily", start_date="2026-09-01"), ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01"),
ScheduleRule(rule_type="daily", interval=3, start_date="2026-09-01", ScheduleRule(rule_type=RuleType.DAILY, interval=3, start_date="2026-09-01",
end_date="2026-12-31"), end_date="2026-12-31"),
ScheduleRule(rule_type="weekly", weekday_set={0, 2, 4}, ScheduleRule(rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
start_date="2026-09-01"), start_date="2026-09-01"),
ScheduleRule(rule_type="weekly", weekday_set={6}, interval=2, ScheduleRule(rule_type=RuleType.WEEKLY, weekday_set={6}, interval=2,
start_date="2026-09-01"), start_date="2026-09-01"),
ScheduleRule(rule_type="monthly_day", day_of_month=15, ScheduleRule(rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
start_date="2026-09-01"), start_date="2026-09-01"),
ScheduleRule(rule_type="monthly_day", day_of_month=31, interval=3, ScheduleRule(rule_type=RuleType.MONTHLY_DAY, day_of_month=31, interval=3,
start_date="2026-09-01"), start_date="2026-09-01"),
ScheduleRule(rule_type="monthly_weekday", nth_weekday=2, ScheduleRule(rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=2,
nth_weekday_weekday=1, start_date="2026-09-01"), nth_weekday_weekday=1, start_date="2026-09-01"),
ScheduleRule(rule_type="monthly_weekday", nth_weekday=-1, ScheduleRule(rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=-1,
nth_weekday_weekday=4, interval=2, nth_weekday_weekday=4, interval=2,
start_date="2026-09-01"), start_date="2026-09-01"),
ScheduleRule(rule_type="yearly", month=6, start_date="2026-09-01"), ScheduleRule(rule_type=RuleType.YEARLY, month=6, start_date="2026-09-01"),
ScheduleRule(rule_type="yearly", month=2, day_of_month=None, ScheduleRule(rule_type=RuleType.YEARLY, month=2, day_of_month=None,
interval=2, start_date="2026-09-01"), interval=2, start_date="2026-09-01"),
ScheduleRule(rule_type="custom_days", weekday_set={1, 3, 5}, ScheduleRule(rule_type=RuleType.CUSTOM_DAYS, weekday_set={1, 3, 5},
interval=2, start_date="2026-09-01"), interval=2, start_date="2026-09-01"),
], ],
) )
@@ -237,7 +237,7 @@ def test_from_dict_defaults_start_date_when_absent() -> None:
def test_from_dict_accepts_list_weekday_set() -> None: def test_from_dict_accepts_list_weekday_set() -> None:
rule = ScheduleRule.from_dict( rule = ScheduleRule.from_dict(
{"rule_type": "weekly", "weekday_set": [0, 2, 4]} {"rule_type": RuleType.WEEKLY.value, "weekday_set": [0, 2, 4]}
) )
assert rule.weekday_set == frozenset({0, 2, 4}) assert rule.weekday_set == frozenset({0, 2, 4})
@@ -248,20 +248,20 @@ def test_from_dict_accepts_list_weekday_set() -> None:
def test_equal_rules_compare_equal_and_hash_equal() -> None: def test_equal_rules_compare_equal_and_hash_equal() -> None:
a = _rule(rule_type="weekly", weekday_set={0, 2}, start_date="2026-09-01") a = _rule(rule_type=RuleType.WEEKLY, weekday_set={0, 2}, start_date="2026-09-01")
b = _rule(rule_type="weekly", weekday_set={2, 0}, start_date="2026-09-01") b = _rule(rule_type=RuleType.WEEKLY, weekday_set={2, 0}, start_date="2026-09-01")
assert a == b assert a == b
assert hash(a) == hash(b) assert hash(a) == hash(b)
def test_different_rules_compare_unequal() -> None: def test_different_rules_compare_unequal() -> None:
a = _rule(rule_type="daily", start_date="2026-09-01") a = _rule(rule_type=RuleType.DAILY, start_date="2026-09-01")
b = _rule(rule_type="daily", interval=2, start_date="2026-09-01") b = _rule(rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01")
assert a != b assert a != b
def test_rule_not_equal_to_other_types() -> None: def test_rule_not_equal_to_other_types() -> None:
rule = _rule(rule_type="daily", start_date="2026-09-01") rule = _rule(rule_type=RuleType.DAILY, start_date="2026-09-01")
assert rule != "daily" assert rule != "daily"
assert (rule == 42) is False assert (rule == 42) is False
@@ -272,29 +272,140 @@ def test_rule_not_equal_to_other_types() -> None:
def test_recurrence_module_imports_only_stdlib() -> None: def test_recurrence_module_imports_only_stdlib() -> None:
"""The model must carry no Home Assistant and no DB imports.""" """The model must carry no Home Assistant and no DB imports — checked
against the RESOLVED module path so relative forms (from .db import
x) are caught too."""
import ast import ast
import importlib.util
from pathlib import Path from pathlib import Path
source = ( module_file = Path(
Path(ScheduleRule.__module__.replace(".", "/")).parent importlib.util.find_spec(
/ "recurrence.py" "custom_components.nestquest.recurrence"
) if False else ( ).origin
Path(__import__("custom_components.nestquest.recurrence",
fromlist=["__file__"]).__file__)
) )
tree = ast.parse(source.read_text()) tree = ast.parse(module_file.read_text())
forbidden = ("homeassistant", "sqlite3", ".db", "dao_")
for node in ast.walk(tree): for node in ast.walk(tree):
if isinstance(node, ast.Import): if isinstance(node, ast.Import):
for alias in node.names: for alias in node.names:
for fragment in forbidden: assert "homeassistant" not in alias.name.lower(), (
assert not alias.name.startswith(fragment), ( f"recurrence.py must not import {alias.name}"
f"recurrence.py must not import {alias.name}"
)
elif isinstance(node, ast.ImportFrom):
module = node.module or ""
for fragment in forbidden:
assert not module.startswith(fragment), (
f"recurrence.py must not import from {module}"
) )
assert "sqlite" not in alias.name.lower(), (
f"recurrence.py must not import {alias.name}"
)
elif isinstance(node, ast.ImportFrom):
module = (node.module or "").lower()
level = node.level # 1+ means relative import
# Relative imports into the integration package (db/dao
# clients) are forbidden; schema.py is also off-limits.
if level:
assert "db" not in module.split(".") and module not in {
"db",
"schema",
"dao_children",
"dao_rules",
"dao_presence",
"dao_instances",
"migrations",
"store",
}, f"recurrence.py must not import from .{module}"
assert "homeassistant" not in module, (
f"recurrence.py must not import from {module}"
)
assert "sqlite" not in module, (
f"recurrence.py must not import from {module}"
)
def test_non_string_rule_type_rejected() -> None:
"""Only RuleType members and their string values construct."""
for bad in (123, True, None, 3.5, b"daily"):
with pytest.raises(RuleValidationError):
ScheduleRule(rule_type=bad, start_date="2026-09-01")
def test_weekly_rejects_month_and_nth_fields() -> None:
with pytest.raises(RuleValidationError, match="must not set"):
ScheduleRule(
rule_type="weekly",
weekday_set={0},
day_of_month=5,
start_date="2026-09-01",
)
with pytest.raises(RuleValidationError, match="must not set"):
ScheduleRule(
rule_type=RuleType.WEEKLY,
weekday_set={0},
month=3,
start_date="2026-09-01",
)
def test_daily_rejects_dangling_nth_weekday_weekday() -> None:
with pytest.raises(RuleValidationError, match="must not set"):
ScheduleRule(
rule_type=RuleType.DAILY,
nth_weekday_weekday=0,
start_date="2026-09-01",
)
def test_monthly_day_rejects_month() -> None:
with pytest.raises(RuleValidationError, match="must not set"):
ScheduleRule(
rule_type=RuleType.MONTHLY_DAY,
day_of_month=5,
month=3,
start_date="2026-09-01",
)
def test_yearly_rejects_day_and_nth_fields() -> None:
with pytest.raises(RuleValidationError, match="must not set"):
ScheduleRule(
rule_type=RuleType.YEARLY,
month=6,
day_of_month=15,
start_date="2026-09-01",
)
with pytest.raises(RuleValidationError, match="must not set"):
ScheduleRule(
rule_type=RuleType.YEARLY,
month=6,
nth_weekday=1,
nth_weekday_weekday=0,
start_date="2026-09-01",
)
def test_storage_value_mapping() -> None:
"""The enum exposes the schema's CHECK values via storage_value."""
assert RuleType.DAILY.storage_value == "daily"
assert RuleType.WEEKLY.storage_value == "weekly"
assert RuleType.MONTHLY_DAY.storage_value == "monthly"
assert RuleType.MONTHLY_WEEKDAY.storage_value == "monthly"
assert RuleType.YEARLY.storage_value == "yearly"
assert RuleType.CUSTOM_DAYS.storage_value == "custom"
def test_from_dict_bool_weekday_entry_rejected_before_set_coercion() -> None:
"""[0, False] must NOT collapse False into 0 via set() coercion."""
with pytest.raises(RuleValidationError, match="plain integers"):
ScheduleRule.from_dict(
{"rule_type": "weekly", "weekday_set": [0, False]}
)
def test_from_dict_unknown_key_types_never_leak_typeerror() -> None:
"""Mixed unhashable unknown keys must surface as
RuleValidationError, not a raw TypeError from sorting."""
data = {"rule_type": "daily", 42: "value"}
with pytest.raises(RuleValidationError):
ScheduleRule.from_dict(data)
def test_rule_type_is_a_dataclass() -> None:
import dataclasses
assert dataclasses.is_dataclass(ScheduleRule)