Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b7911f66f6 | ||
|
|
87251eeec5 | ||
|
|
ceb32074e7 |
@@ -0,0 +1,478 @@
|
||||
"""The ScheduleRule model: rule types, validation, and serialization.
|
||||
|
||||
A :class:`ScheduleRule` describes WHEN a recurring chore fires — a pure
|
||||
frozen dataclass with no Home Assistant and no database imports. The
|
||||
evaluation engine (Feature 04) consumes it; the DAO layer (Feature 02)
|
||||
serializes its dict form into the ``schedule_rules`` table.
|
||||
|
||||
Type model: one enum member per recurrence shape. ``MONTHLY_DAY``
|
||||
("the 15th of every month") and ``MONTHLY_WEEKDAY`` ("the second
|
||||
Tuesday of every month") are distinct types rather than one monthly
|
||||
type with optional fields, so validation is total: a rule either
|
||||
constructs, or raises :class:`RuleValidationError` naming the offending
|
||||
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
|
||||
constraints are a second line of defense, but invalid rules must never
|
||||
reach storage):
|
||||
|
||||
- ``interval`` >= 1 for every rule.
|
||||
- ``weekday_set`` is required for WEEKLY and CUSTOM_DAYS, forbidden
|
||||
elsewhere; entries are 0 (Monday) .. 6 (Sunday), no booleans.
|
||||
- ``day_of_month`` 1..31 is required for MONTHLY_DAY.
|
||||
- ``nth_weekday`` 1..5 or -1 (0 is meaningless as a position and is
|
||||
rejected) with a weekday 0..6 for MONTHLY_WEEKDAY.
|
||||
- ``month`` 1..12 is required for YEARLY.
|
||||
- ``start_date`` is a strict ISO date; ``end_date`` is optional and,
|
||||
when present, on or after ``start_date``.
|
||||
|
||||
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
|
||||
February 29th does NOT fire in a non-leap year — the engine skips
|
||||
instead of clamping to month end, because clamping silently moves
|
||||
"the 31st" to "the 30th" and double-fires tasks.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import enum
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
#: Date-shape policy: strict YYYY-MM-DD, matching the DAO layers.
|
||||
_DATE_FORMAT = "%Y-%m-%d"
|
||||
|
||||
#: Weekday-name policy: 0 = Monday .. 6 = Sunday (ISO order).
|
||||
|
||||
|
||||
class RuleType(enum.Enum):
|
||||
"""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"
|
||||
WEEKLY = "weekly"
|
||||
MONTHLY_DAY = "monthly_day"
|
||||
MONTHLY_WEEKDAY = "monthly_weekday"
|
||||
YEARLY = "yearly"
|
||||
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):
|
||||
"""Raised when a ScheduleRule field combination is invalid."""
|
||||
|
||||
|
||||
def _validate_date(value: str, field_name: str) -> None:
|
||||
"""Raise unless ``value`` is a strict YYYY-MM-DD calendar date."""
|
||||
try:
|
||||
parsed = datetime.datetime.strptime(value, _DATE_FORMAT).date()
|
||||
except (TypeError, ValueError):
|
||||
raise RuleValidationError(
|
||||
f"{field_name} must be an ISO date (YYYY-MM-DD), got {value!r}"
|
||||
) from None
|
||||
if parsed.isoformat() != value:
|
||||
raise RuleValidationError(
|
||||
f"{field_name} must be a strict YYYY-MM-DD date, got {value!r}"
|
||||
)
|
||||
|
||||
|
||||
def _validate_weekday_set(weekday_set) -> frozenset[int]:
|
||||
"""Return a normalized frozenset or raise with a clear message.
|
||||
|
||||
Sequence entries are validated BEFORE set() coercion: a list like
|
||||
[0, False] must not collapse the boolean into 0, and unhashable
|
||||
entries must raise RuleValidationError, never a raw TypeError.
|
||||
"""
|
||||
if isinstance(weekday_set, (list, tuple)):
|
||||
for entry in weekday_set:
|
||||
# _is_plain_int: bool and int subclasses (potentially
|
||||
# unhashable) are rejected before set() coercion.
|
||||
if not _is_plain_int(entry):
|
||||
raise RuleValidationError(
|
||||
f"weekday_set entries must be plain integers, got "
|
||||
f"{entry!r}"
|
||||
)
|
||||
weekday_set = set(weekday_set)
|
||||
if not isinstance(weekday_set, (set, frozenset)):
|
||||
raise RuleValidationError(
|
||||
f"weekday_set must be a set of weekday integers, got "
|
||||
f"{type(weekday_set).__name__}"
|
||||
)
|
||||
if not weekday_set:
|
||||
raise RuleValidationError(
|
||||
"weekday_set must not be empty for weekly/custom rules"
|
||||
)
|
||||
for entry in weekday_set:
|
||||
if not _is_plain_int(entry):
|
||||
raise RuleValidationError(
|
||||
f"weekday_set entries must be plain integers, got "
|
||||
f"{entry!r}"
|
||||
)
|
||||
if not 0 <= entry <= 6:
|
||||
raise RuleValidationError(
|
||||
f"weekday_set entries must be 0 (Monday)..6 (Sunday), "
|
||||
f"got {entry!r}"
|
||||
)
|
||||
return frozenset(weekday_set)
|
||||
|
||||
|
||||
def _is_plain_int(value) -> bool:
|
||||
"""True only for real ints: bool excluded AND int subclasses too
|
||||
(an unhashable int subclass would break the frozen-hashable model)."""
|
||||
return type(value) is int
|
||||
|
||||
|
||||
def _validate_monthly_weekday(nth_weekday, weekday) -> None:
|
||||
"""Validate the MONTHLY_WEEKDAY position fields."""
|
||||
if nth_weekday is None or not _is_plain_int(nth_weekday):
|
||||
raise RuleValidationError(
|
||||
"nth_weekday must be an integer for monthly_weekday rules, "
|
||||
f"got {nth_weekday!r}"
|
||||
)
|
||||
if not (-1 <= nth_weekday <= 5) or nth_weekday == 0:
|
||||
raise RuleValidationError(
|
||||
f"nth_weekday must be -1 (last) or 1..5, got {nth_weekday!r}"
|
||||
)
|
||||
if weekday is None or not _is_plain_int(weekday) or not 0 <= weekday <= 6:
|
||||
raise RuleValidationError(
|
||||
f"monthly_weekday rules require a weekday 0 (Monday)..6 "
|
||||
f"(Sunday), got {weekday!r}"
|
||||
)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ScheduleRule:
|
||||
"""An immutable, validated recurrence rule (see module docstring).
|
||||
|
||||
Construct with keyword arguments; invalid combinations raise
|
||||
:class:`RuleValidationError` at construction time. Fields map
|
||||
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).
|
||||
"""
|
||||
|
||||
rule_type: RuleType
|
||||
interval: int = 1
|
||||
weekday_set: frozenset[int] | None = None
|
||||
day_of_month: int | None = None
|
||||
nth_weekday: int | None = None
|
||||
nth_weekday_weekday: int | None = None
|
||||
month: int | None = None
|
||||
start_date: str = "1970-01-01"
|
||||
end_date: str | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
# Normalize a string rule_type into the enum; anything else
|
||||
# that is not already a RuleType is rejected.
|
||||
if isinstance(self.rule_type, str):
|
||||
# EXACT model names only: no whitespace stripping, no storage
|
||||
# aliases. The DAO maps storage strings to model fields when
|
||||
# reading rows; the model boundary stays unambiguous.
|
||||
candidates = {rt.name.lower(): rt for rt in RuleType}
|
||||
if self.rule_type in candidates:
|
||||
object.__setattr__(
|
||||
self, "rule_type", candidates[self.rule_type]
|
||||
)
|
||||
else:
|
||||
raise RuleValidationError(
|
||||
f"rule_type must be a RuleType or one of "
|
||||
f"{sorted(candidates)}, got {self.rule_type!r}"
|
||||
)
|
||||
if not isinstance(self.rule_type, RuleType):
|
||||
raise RuleValidationError(
|
||||
f"rule_type must be a RuleType or one of its string "
|
||||
f"values, got {self.rule_type!r}"
|
||||
)
|
||||
if not _is_plain_int(self.interval):
|
||||
raise RuleValidationError(
|
||||
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(self.start_date, "start_date")
|
||||
if self.end_date is not None:
|
||||
_validate_date(self.end_date, "end_date")
|
||||
if self.end_date < self.start_date:
|
||||
raise RuleValidationError(
|
||||
f"end_date {self.end_date!r} must be on or after "
|
||||
f"start_date {self.start_date!r}"
|
||||
)
|
||||
|
||||
# Type-exclusive field policy, enforced for every type: build
|
||||
# the dict of present fields, then reject every field the shape
|
||||
# must not carry (weekday_set present is detectable via None vs
|
||||
# set even after normalization below).
|
||||
shape = self.rule_type
|
||||
if shape is RuleType.DAILY:
|
||||
forbidden_fields = (
|
||||
"weekday_set",
|
||||
"day_of_month",
|
||||
"nth_weekday",
|
||||
"nth_weekday_weekday",
|
||||
"month",
|
||||
)
|
||||
elif shape is RuleType.WEEKLY:
|
||||
forbidden_fields = (
|
||||
"day_of_month",
|
||||
"nth_weekday",
|
||||
"nth_weekday_weekday",
|
||||
"month",
|
||||
)
|
||||
elif shape is RuleType.CUSTOM_DAYS:
|
||||
forbidden_fields = (
|
||||
"day_of_month",
|
||||
"nth_weekday",
|
||||
"nth_weekday_weekday",
|
||||
"month",
|
||||
)
|
||||
elif shape is RuleType.MONTHLY_DAY:
|
||||
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(
|
||||
"monthly_day rules require day_of_month (1..31)"
|
||||
)
|
||||
if not 1 <= self.day_of_month <= 31:
|
||||
raise RuleValidationError(
|
||||
f"day_of_month must be 1..31, got {self.day_of_month!r}"
|
||||
)
|
||||
elif shape is RuleType.MONTHLY_WEEKDAY:
|
||||
_validate_monthly_weekday(
|
||||
self.nth_weekday, self.nth_weekday_weekday
|
||||
)
|
||||
elif shape is RuleType.YEARLY:
|
||||
if self.month is None or not _is_plain_int(self.month):
|
||||
raise RuleValidationError(
|
||||
"yearly rules require month (1..12)"
|
||||
)
|
||||
if not 1 <= self.month <= 12:
|
||||
raise RuleValidationError(
|
||||
f"month must be 1..12, got {self.month!r}"
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""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 {
|
||||
"rule_type": self.rule_type.name.lower(), # model name
|
||||
"interval": self.interval,
|
||||
"weekday_set": (
|
||||
sorted(self.weekday_set) if self.weekday_set is not None
|
||||
else None
|
||||
),
|
||||
"day_of_month": self.day_of_month,
|
||||
"nth_weekday": self.nth_weekday,
|
||||
"nth_weekday_weekday": self.nth_weekday_weekday,
|
||||
"month": self.month,
|
||||
"start_date": self.start_date,
|
||||
"end_date": self.end_date,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ScheduleRule":
|
||||
"""Rebuild a rule from :meth:`to_dict` output.
|
||||
|
||||
Raises :class:`RuleValidationError` on unknown keys, missing
|
||||
keys, or invalid combinations — never a raw TypeError.
|
||||
"""
|
||||
if not isinstance(data, dict):
|
||||
raise RuleValidationError(
|
||||
f"rule data must be a dict, got {type(data).__name__}"
|
||||
)
|
||||
allowed = {
|
||||
"rule_type",
|
||||
"interval",
|
||||
"weekday_set",
|
||||
"day_of_month",
|
||||
"nth_weekday",
|
||||
"nth_weekday_weekday",
|
||||
"month",
|
||||
"start_date",
|
||||
"end_date",
|
||||
}
|
||||
unknown = [key for key in data if key not in allowed]
|
||||
if unknown:
|
||||
raise RuleValidationError(
|
||||
f"unknown rule fields: {sorted(unknown, key=repr)}"
|
||||
)
|
||||
if "rule_type" not in data:
|
||||
raise RuleValidationError("missing required rule field: rule_type")
|
||||
weekday_set = data.get("weekday_set")
|
||||
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 and an
|
||||
# unhashable/int-subclass entry cannot leak a raw TypeError.
|
||||
for entry in weekday_set:
|
||||
if not _is_plain_int(entry):
|
||||
raise RuleValidationError(
|
||||
f"weekday_set entries must be plain integers, got "
|
||||
f"{entry!r}"
|
||||
)
|
||||
weekday_set = set(weekday_set)
|
||||
return cls(
|
||||
rule_type=data["rule_type"],
|
||||
interval=data.get("interval", 1),
|
||||
weekday_set=weekday_set,
|
||||
day_of_month=data.get("day_of_month"),
|
||||
nth_weekday=data.get("nth_weekday"),
|
||||
nth_weekday_weekday=data.get("nth_weekday_weekday"),
|
||||
month=data.get("month"),
|
||||
start_date=data.get("start_date", "1970-01-01"),
|
||||
end_date=data.get("end_date"),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Evaluation engine (Feature 04): pure date arithmetic on ScheduleRule
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _parse_date(value: str) -> datetime.date:
|
||||
"""Parse a strict ISO date already validated by the model."""
|
||||
return datetime.datetime.strptime(value, _DATE_FORMAT).date()
|
||||
|
||||
|
||||
def _anchor_offset(rule: ScheduleRule, target: datetime.date) -> int:
|
||||
"""Day offset from the rule's anchor date to ``target``.
|
||||
|
||||
The anchor for DAILY/WEEKLY-interval arithmetic is start_date: the
|
||||
first day the rule CAN fire. A negative offset means the target
|
||||
precedes the rule's window entirely.
|
||||
"""
|
||||
return (target - _parse_date(rule.start_date)).days
|
||||
|
||||
|
||||
def _within_window(rule: ScheduleRule, target: datetime.date) -> bool:
|
||||
"""True when ``target`` is inside the rule's [start, end] window."""
|
||||
if target < _parse_date(rule.start_date):
|
||||
return False
|
||||
if rule.end_date is not None and target > _parse_date(rule.end_date):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
|
||||
"""Return True when ``rule`` fires on ``target``.
|
||||
|
||||
Pure function: no I/O, no timezone awareness, no mutation. Dates
|
||||
are plain calendar dates (Feature 04 guardrail); the caller decides
|
||||
the timezone that produced them.
|
||||
"""
|
||||
# Dispatch FIRST: an unimplemented shape must raise loudly even for
|
||||
# dates outside the window, never silently evaluate to False.
|
||||
shape = rule.rule_type
|
||||
if shape is not RuleType.DAILY:
|
||||
raise NotImplementedError(
|
||||
f"occurs_on for {shape} lands with its own engine task "
|
||||
"(Features 04 task sequence)"
|
||||
)
|
||||
if not _within_window(rule, target):
|
||||
return False
|
||||
offset = (target - _parse_date(rule.start_date)).days
|
||||
return offset % rule.interval == 0
|
||||
|
||||
|
||||
def occurrences_between(
|
||||
rule: ScheduleRule, start: str, end: str
|
||||
) -> list[str]:
|
||||
"""Return every date string in [start, end] the rule fires on.
|
||||
|
||||
Inclusive on both ends. Raises RuleValidationError on malformed or
|
||||
inverted bounds; a window entirely outside [rule.start_date,
|
||||
rule.end_date] yields an empty list.
|
||||
"""
|
||||
if rule.rule_type is not RuleType.DAILY:
|
||||
raise NotImplementedError(
|
||||
f"occurrences_between for {rule.rule_type} lands with its own "
|
||||
"engine task (Features 04 task sequence)"
|
||||
)
|
||||
_validate_date(start, "start")
|
||||
_validate_date(end, "end")
|
||||
start_date = _parse_date(start)
|
||||
end_date = _parse_date(end)
|
||||
if end_date < start_date:
|
||||
raise RuleValidationError(
|
||||
f"end {end!r} must be on or after start {start!r}"
|
||||
)
|
||||
rule_start = _parse_date(rule.start_date)
|
||||
if end_date < rule_start:
|
||||
return []
|
||||
# Clamp the walk to the rule's own window and the queried range.
|
||||
effective_start = max(start_date, rule_start)
|
||||
effective_end = end_date
|
||||
if rule.end_date is not None:
|
||||
effective_end = min(effective_end, _parse_date(rule.end_date))
|
||||
results: list[str] = []
|
||||
cursor = effective_start
|
||||
while cursor <= effective_end:
|
||||
if occurs_on(rule, cursor):
|
||||
results.append(cursor.isoformat())
|
||||
cursor += datetime.timedelta(days=1)
|
||||
return results
|
||||
@@ -0,0 +1,293 @@
|
||||
"""Tests for the daily recurrence evaluation (occurs_on/occurrences_between)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from custom_components.nestquest.recurrence import (
|
||||
RuleType,
|
||||
RuleValidationError,
|
||||
ScheduleRule,
|
||||
occurs_on,
|
||||
occurrences_between,
|
||||
)
|
||||
|
||||
|
||||
def _d(date: str) -> datetime.date:
|
||||
return datetime.date.fromisoformat(date)
|
||||
|
||||
|
||||
def test_daily_fires_every_day_from_start() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
assert occurs_on(rule, _d("2026-09-01")) is True
|
||||
assert occurs_on(rule, _d("2026-09-02")) is True
|
||||
assert occurs_on(rule, _d("2026-09-30")) is True
|
||||
|
||||
|
||||
def test_daily_never_fires_before_start() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
assert occurs_on(rule, _d("2026-08-31")) is False
|
||||
|
||||
|
||||
def test_daily_respects_end_date() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, start_date="2026-09-01",
|
||||
end_date="2026-09-10",
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-10")) is True
|
||||
assert occurs_on(rule, _d("2026-09-11")) is False
|
||||
|
||||
|
||||
def test_daily_interval_2_fires_every_second_day() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
# Odd offsets fire (0, 2, 4...); odd offsets don't.
|
||||
assert occurs_on(rule, _d("2026-09-01")) is True # offset 0
|
||||
assert occurs_on(rule, _d("2026-09-02")) is False # offset 1
|
||||
assert occurs_on(rule, _d("2026-09-03")) is True # offset 2
|
||||
assert occurs_on(rule, _d("2026-09-04")) is False # offset 3
|
||||
assert occurs_on(rule, _d("2026-09-05")) is True # offset 4
|
||||
|
||||
|
||||
def test_daily_interval_3_fires_every_third_day() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=3, start_date="2026-09-01"
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-01")) is True # 0
|
||||
assert occurs_on(rule, _d("2026-09-02")) is False # 1
|
||||
assert occurs_on(rule, _d("2026-09-03")) is False # 2
|
||||
assert occurs_on(rule, _d("2026-09-04")) is True # 3
|
||||
assert occurs_on(rule, _d("2026-09-07")) is True # 6
|
||||
|
||||
|
||||
def test_hand_checked_30_day_sequence_interval_1() -> None:
|
||||
"""The done-condition's hand-checked sequence: 30 days, all fire."""
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
expected = [
|
||||
(2026, 9, day) for day in range(1, 31)
|
||||
]
|
||||
for (year, month, day) in expected:
|
||||
assert occurs_on(rule, datetime.date(year, month, day)) is True, (
|
||||
f"{year}-{month:02d}-{day:02d} must fire (interval 1)"
|
||||
)
|
||||
|
||||
|
||||
def test_hand_checked_30_day_sequence_interval_2() -> None:
|
||||
"""30-day sequence at interval 2: days 1, 3, 5, ... (odd days)."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
for day in range(1, 31):
|
||||
fired = occurs_on(rule, datetime.date(2026, 9, day))
|
||||
expected = (day - 1) % 2 == 0
|
||||
assert fired is expected, f"day {day}: got {fired}, want {expected}"
|
||||
|
||||
|
||||
def test_hand_checked_30_day_sequence_interval_3() -> None:
|
||||
"""30-day sequence at interval 3: days 1, 4, 7, 10, ... 28."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=3, start_date="2026-09-01"
|
||||
)
|
||||
fired_days = [
|
||||
day
|
||||
for day in range(1, 31)
|
||||
if occurs_on(rule, datetime.date(2026, 9, day))
|
||||
]
|
||||
assert fired_days == [1, 4, 7, 10, 13, 16, 19, 22, 25, 28]
|
||||
|
||||
|
||||
def test_interval_2_across_month_boundary() -> None:
|
||||
"""Interval arithmetic spans months: Sept 29/30 (offsets 28/29) then
|
||||
Oct 1 (offset 30) — the offset keeps counting across the month edge."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-29")) is True # offset 28
|
||||
assert occurs_on(rule, _d("2026-09-30")) is False # offset 29
|
||||
assert occurs_on(rule, _d("2026-10-01")) is True # offset 30
|
||||
assert occurs_on(rule, _d("2026-10-02")) is False # offset 31
|
||||
|
||||
|
||||
def test_interval_2_across_year_boundary() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-12-30"
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-12-30")) is True # 0
|
||||
assert occurs_on(rule, _d("2026-12-31")) is False # 1
|
||||
assert occurs_on(rule, _d("2027-01-01")) is True # 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# occurrences_between
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_occurrences_between_daily_interval_1() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
result = occurrences_between(rule, "2026-09-05", "2026-09-08")
|
||||
assert result == [
|
||||
"2026-09-05", "2026-09-06", "2026-09-07", "2026-09-08",
|
||||
]
|
||||
|
||||
|
||||
def test_occurrences_between_daily_interval_2() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-10")
|
||||
assert result == [
|
||||
"2026-09-01", "2026-09-03", "2026-09-05", "2026-09-07", "2026-09-09",
|
||||
]
|
||||
|
||||
|
||||
def test_occurrences_between_clamps_to_rule_start() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-05")
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-08")
|
||||
assert result == ["2026-09-05", "2026-09-06", "2026-09-07", "2026-09-08"]
|
||||
|
||||
|
||||
def test_occurrences_between_clamps_to_rule_end() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, start_date="2026-09-01",
|
||||
end_date="2026-09-06",
|
||||
)
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-20")
|
||||
assert result == [
|
||||
f"2026-09-{day:02d}" for day in range(1, 7)
|
||||
]
|
||||
|
||||
|
||||
def test_occurrences_between_disjoint_range_is_empty() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, start_date="2026-09-01",
|
||||
end_date="2026-09-15",
|
||||
)
|
||||
# The rule ends Sept 15: October is entirely outside its window.
|
||||
assert occurrences_between(rule, "2026-10-01", "2026-10-05") == []
|
||||
assert occurrences_between(rule, "2026-08-01", "2026-08-05") == []
|
||||
|
||||
|
||||
def test_occurrences_between_inclusive_bounds() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
# Both boundary dates fire at interval 2 (offsets 0 and 6).
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-07")
|
||||
assert result[0] == "2026-09-01"
|
||||
assert result[-1] == "2026-09-07"
|
||||
|
||||
|
||||
def test_occurrences_between_rejects_inverted_range() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
with pytest.raises(RuleValidationError, match="on or after"):
|
||||
occurrences_between(rule, "2026-09-10", "2026-09-05")
|
||||
|
||||
|
||||
def test_occurrences_between_rejects_malformed_dates() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
with pytest.raises(RuleValidationError, match="YYYY-MM-DD"):
|
||||
occurrences_between(rule, "oops", "2026-09-05")
|
||||
with pytest.raises(RuleValidationError, match="YYYY-MM-DD"):
|
||||
occurrences_between(rule, "2026-09-05", "2026-9-5")
|
||||
|
||||
|
||||
def test_occurrences_between_leap_day_fires() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-02-27")
|
||||
# 2028 is a leap year: Feb 29 exists and is 2 days after Feb 27.
|
||||
result = occurrences_between(rule, "2028-02-28", "2028-03-01")
|
||||
assert result == ["2028-02-28", "2028-02-29", "2028-03-01"]
|
||||
|
||||
|
||||
def test_occurrences_between_year_boundary() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-12-30")
|
||||
result = occurrences_between(rule, "2026-12-30", "2027-01-02")
|
||||
assert result == [
|
||||
"2026-12-30", "2026-12-31", "2027-01-01", "2027-01-02",
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Purity: the engine raises on unknown shapes rather than guessing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_occurs_on_rejects_unimplemented_shapes_loudly() -> None:
|
||||
"""Weekly and beyond land with their own engine tasks; a loud
|
||||
NotImplementedError beats a silent wrong answer."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0}, start_date="2026-09-01"
|
||||
)
|
||||
with pytest.raises(NotImplementedError):
|
||||
occurs_on(rule, _d("2026-09-07"))
|
||||
|
||||
def test_interval_7_preserves_weekday_alignment_across_months() -> None:
|
||||
"""Weekly-by-interval semantics: every firing lands on the SAME
|
||||
weekday as start_date, no matter how many month edges intervene."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=7, start_date="2026-09-01"
|
||||
)
|
||||
# 2026-09-01 is a Tuesday; every firing must be a Tuesday too.
|
||||
assert _d("2026-09-01").weekday() == 1
|
||||
for day in ("2026-09-08", "2026-09-29", "2026-10-06", "2026-11-03"):
|
||||
assert occurs_on(rule, _d(day)) is True, day
|
||||
assert _d(day).weekday() == 1, f"{day} must be a Tuesday"
|
||||
# The day after a firing is never a firing.
|
||||
assert occurs_on(rule, _d("2026-10-07")) is False
|
||||
|
||||
|
||||
def test_interval_2_across_leap_day() -> None:
|
||||
"""Interval > 1 arithmetic across a leap day: the extra day shifts
|
||||
subsequent offsets by one, and the mod arithmetic reflects it."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2028-02-27"
|
||||
)
|
||||
assert occurs_on(rule, _d("2028-02-27")) is True # 0
|
||||
assert occurs_on(rule, _d("2028-02-28")) is False # 1
|
||||
assert occurs_on(rule, _d("2028-02-29")) is True # 2 (leap day)
|
||||
assert occurs_on(rule, _d("2028-03-01")) is False # 3
|
||||
assert occurs_on(rule, _d("2028-03-02")) is True # 4
|
||||
|
||||
|
||||
def test_occurrences_between_leap_day_interval_2() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2028-02-27"
|
||||
)
|
||||
result = occurrences_between(rule, "2028-02-27", "2028-03-03")
|
||||
assert result == ["2028-02-27", "2028-02-29", "2028-03-02"]
|
||||
|
||||
|
||||
def test_occurrences_between_year_boundary_interval_2() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-12-30"
|
||||
)
|
||||
result = occurrences_between(rule, "2026-12-30", "2027-01-04")
|
||||
# Offsets 0, 2, 4, 6 = Dec 30, Jan 1, Jan 3 ... wait: offsets are
|
||||
# Dec30=0, Dec31=1, Jan1=2, Jan2=3, Jan3=4, Jan4=5. Even offsets:
|
||||
assert result == ["2026-12-30", "2027-01-01", "2027-01-03"]
|
||||
|
||||
|
||||
def test_occurs_on_unimplemented_shape_raises_even_outside_window() -> None:
|
||||
"""An unimplemented shape must raise whether the date is inside or
|
||||
outside the window — silent False would hide a missing engine."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0},
|
||||
start_date="2026-09-01", end_date="2026-09-10",
|
||||
)
|
||||
# Before start, inside window, after end: all must raise.
|
||||
for day in ("2026-08-31", "2026-09-07", "2026-09-20"):
|
||||
with pytest.raises(NotImplementedError):
|
||||
occurs_on(rule, _d(day))
|
||||
|
||||
|
||||
def test_occurrences_between_unimplemented_shape_raises_even_disjoint(
|
||||
tmp_path,
|
||||
) -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0}, start_date="2026-09-01"
|
||||
)
|
||||
# A range entirely before the rule's start still raises: the shape
|
||||
# is unimplemented, and silence would hide it.
|
||||
with pytest.raises(NotImplementedError):
|
||||
occurrences_between(rule, "2026-08-01", "2026-08-10")
|
||||
@@ -0,0 +1,480 @@
|
||||
"""Tests for recurrence.py: the ScheduleRule model and serialization."""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from custom_components.nestquest.recurrence import (
|
||||
RuleType,
|
||||
RuleValidationError,
|
||||
ScheduleRule,
|
||||
)
|
||||
|
||||
|
||||
def _rule(**kwargs) -> ScheduleRule:
|
||||
return ScheduleRule(**kwargs)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Valid construction for every rule type
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_daily_rule_constructs_with_defaults() -> None:
|
||||
rule = _rule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
assert rule.rule_type is RuleType.DAILY
|
||||
assert rule.interval == 1
|
||||
assert rule.weekday_set is None
|
||||
assert rule.day_of_month is None
|
||||
assert rule.nth_weekday is None
|
||||
assert rule.month is None
|
||||
assert rule.end_date is None
|
||||
|
||||
|
||||
def test_string_rule_type_coerces() -> None:
|
||||
rule = _rule(rule_type=RuleType.WEEKLY, weekday_set={0, 2}, start_date="2026-09-01")
|
||||
assert rule.rule_type is RuleType.WEEKLY
|
||||
|
||||
|
||||
def test_weekly_rule_requires_and_freezes_weekday_set() -> None:
|
||||
rule = _rule(rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4}, start_date="2026-09-01")
|
||||
assert rule.weekday_set == frozenset({0, 2, 4})
|
||||
# A mutable set passed in is frozen on construction.
|
||||
source = {1, 3}
|
||||
rule2 = _rule(rule_type=RuleType.WEEKLY, weekday_set=source, start_date="2026-09-01")
|
||||
source.add(5)
|
||||
assert rule2.weekday_set == frozenset({1, 3})
|
||||
|
||||
|
||||
def test_monthly_day_rule_fields() -> None:
|
||||
rule = _rule(rule_type=RuleType.MONTHLY_DAY, day_of_month=15, start_date="2026-09-01")
|
||||
assert rule.day_of_month == 15
|
||||
assert rule.nth_weekday is None
|
||||
|
||||
|
||||
def test_monthly_weekday_rule_fields() -> None:
|
||||
rule = _rule(
|
||||
rule_type=RuleType.MONTHLY_WEEKDAY,
|
||||
nth_weekday=2,
|
||||
nth_weekday_weekday=1,
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
assert rule.nth_weekday == 2
|
||||
assert rule.nth_weekday_weekday == 1
|
||||
|
||||
|
||||
def test_monthly_weekday_last_friday_allowed() -> None:
|
||||
rule = _rule(
|
||||
rule_type=RuleType.MONTHLY_WEEKDAY,
|
||||
nth_weekday=-1,
|
||||
nth_weekday_weekday=4,
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
assert rule.nth_weekday == -1
|
||||
|
||||
|
||||
def test_yearly_rule_fields() -> None:
|
||||
rule = _rule(rule_type=RuleType.YEARLY, month=6, start_date="2026-09-01")
|
||||
assert rule.month == 6
|
||||
|
||||
|
||||
def test_custom_days_rule_fields() -> None:
|
||||
rule = _rule(
|
||||
rule_type=RuleType.CUSTOM_DAYS,
|
||||
weekday_set={1, 3, 5},
|
||||
interval=2,
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
assert rule.weekday_set == frozenset({1, 3, 5})
|
||||
assert rule.interval == 2
|
||||
|
||||
|
||||
def test_end_date_round_trips(tmp_path) -> None:
|
||||
rule = _rule(
|
||||
rule_type=RuleType.DAILY,
|
||||
start_date="2026-09-01",
|
||||
end_date="2026-12-31",
|
||||
)
|
||||
assert rule.end_date == "2026-12-31"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Invalid combinations are rejected at construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"kwargs",
|
||||
[
|
||||
# weekly without weekday_set
|
||||
{"rule_type": RuleType.WEEKLY.value, "start_date": "2026-09-01"},
|
||||
# daily carrying a weekday_set
|
||||
{"rule_type": "daily", "weekday_set": {0}, "start_date": "2026-09-01"},
|
||||
# monthly_day without day_of_month
|
||||
{"rule_type": "monthly_day", "start_date": "2026-09-01"},
|
||||
# monthly_day out of range
|
||||
{"rule_type": "monthly_day", "day_of_month": 32, "start_date": "2026-09-01"},
|
||||
# monthly_day carrying weekday_set
|
||||
{"rule_type": "monthly_day", "day_of_month": 5, "weekday_set": {0},
|
||||
"start_date": "2026-09-01"},
|
||||
# yearly without month
|
||||
{"rule_type": "yearly", "start_date": "2026-09-01"},
|
||||
# yearly out of range
|
||||
{"rule_type": "yearly", "month": 13, "start_date": "2026-09-01"},
|
||||
# unknown type
|
||||
{"rule_type": "bogus", "start_date": "2026-09-01"},
|
||||
# interval zero
|
||||
{"rule_type": "daily", "interval": 0, "start_date": "2026-09-01"},
|
||||
# interval non-integer
|
||||
{"rule_type": "daily", "interval": "2", "start_date": "2026-09-01"},
|
||||
# end before start
|
||||
{"rule_type": "daily", "start_date": "2026-09-01",
|
||||
"end_date": "2026-08-31"},
|
||||
# malformed dates
|
||||
{"rule_type": "daily", "start_date": "not-a-date"},
|
||||
{"rule_type": "daily", "start_date": "2026-9-1"},
|
||||
# weekly weekday out of range
|
||||
{"rule_type": RuleType.WEEKLY.value, "weekday_set": {7}, "start_date": "2026-09-01"},
|
||||
# weekly empty set
|
||||
{"rule_type": RuleType.WEEKLY.value, "weekday_set": set(), "start_date": "2026-09-01"},
|
||||
# custom_days carrying month-specific fields
|
||||
{"rule_type": "custom_days", "weekday_set": {0}, "day_of_month": 5,
|
||||
"start_date": "2026-09-01"},
|
||||
{"rule_type": "custom_days", "weekday_set": {0}, "nth_weekday": 1,
|
||||
"nth_weekday_weekday": 0, "start_date": "2026-09-01"},
|
||||
{"rule_type": "custom_days", "weekday_set": {0}, "month": 3,
|
||||
"start_date": "2026-09-01"},
|
||||
# monthly_weekday position invalid
|
||||
{"rule_type": "monthly_weekday", "nth_weekday": 0,
|
||||
"nth_weekday_weekday": 1, "start_date": "2026-09-01"},
|
||||
{"rule_type": "monthly_weekday", "nth_weekday": 6,
|
||||
"nth_weekday_weekday": 1, "start_date": "2026-09-01"},
|
||||
{"rule_type": "monthly_weekday", "nth_weekday": 2,
|
||||
"nth_weekday_weekday": None, "start_date": "2026-09-01"},
|
||||
{"rule_type": "monthly_weekday", "nth_weekday": 2,
|
||||
"nth_weekday_weekday": 7, "start_date": "2026-09-01"},
|
||||
# monthly_day carrying nth fields
|
||||
{"rule_type": "monthly_day", "day_of_month": 5, "nth_weekday": 1,
|
||||
"nth_weekday_weekday": 0, "start_date": "2026-09-01"},
|
||||
],
|
||||
)
|
||||
def test_invalid_combinations_raise(kwargs) -> None:
|
||||
with pytest.raises(RuleValidationError):
|
||||
_rule(**kwargs)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Immutability
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_rule_is_immutable() -> None:
|
||||
rule = _rule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
with pytest.raises(AttributeError):
|
||||
rule.interval = 2
|
||||
with pytest.raises(AttributeError):
|
||||
del rule.interval
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Serialization round-trip for every rule type
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"rule",
|
||||
[
|
||||
ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.DAILY, interval=3, start_date="2026-09-01",
|
||||
end_date="2026-12-31"),
|
||||
ScheduleRule(rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
|
||||
start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.WEEKLY, weekday_set={6}, interval=2,
|
||||
start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
|
||||
start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.MONTHLY_DAY, day_of_month=31, interval=3,
|
||||
start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=2,
|
||||
nth_weekday_weekday=1, start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=-1,
|
||||
nth_weekday_weekday=4, interval=2,
|
||||
start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.YEARLY, month=6, start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.YEARLY, month=2, day_of_month=None,
|
||||
interval=2, start_date="2026-09-01"),
|
||||
ScheduleRule(rule_type=RuleType.CUSTOM_DAYS, weekday_set={1, 3, 5},
|
||||
interval=2, start_date="2026-09-01"),
|
||||
],
|
||||
)
|
||||
def test_to_dict_from_dict_round_trips_losslessly(rule) -> None:
|
||||
rebuilt = ScheduleRule.from_dict(rule.to_dict())
|
||||
assert rebuilt == rule
|
||||
# Round-tripping twice is stable.
|
||||
assert ScheduleRule.from_dict(rebuilt.to_dict()) == rule
|
||||
|
||||
|
||||
def test_from_dict_rejects_unknown_fields() -> None:
|
||||
with pytest.raises(RuleValidationError, match="unknown rule fields"):
|
||||
ScheduleRule.from_dict({"rule_type": "daily", "bogus": 1})
|
||||
|
||||
|
||||
def test_from_dict_rejects_missing_rule_type() -> None:
|
||||
with pytest.raises(RuleValidationError, match="missing required"):
|
||||
ScheduleRule.from_dict({"interval": 1})
|
||||
|
||||
|
||||
def test_from_dict_rejects_non_dict() -> None:
|
||||
with pytest.raises(RuleValidationError):
|
||||
ScheduleRule.from_dict("daily")
|
||||
|
||||
|
||||
def test_from_dict_defaults_start_date_when_absent() -> None:
|
||||
rule = ScheduleRule.from_dict({"rule_type": "daily"})
|
||||
assert rule.start_date == "1970-01-01"
|
||||
|
||||
|
||||
def test_from_dict_accepts_list_weekday_set() -> None:
|
||||
rule = ScheduleRule.from_dict(
|
||||
{"rule_type": RuleType.WEEKLY.value, "weekday_set": [0, 2, 4]}
|
||||
)
|
||||
assert rule.weekday_set == frozenset({0, 2, 4})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Equality and hashing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_equal_rules_compare_equal_and_hash_equal() -> None:
|
||||
a = _rule(rule_type=RuleType.WEEKLY, weekday_set={0, 2}, start_date="2026-09-01")
|
||||
b = _rule(rule_type=RuleType.WEEKLY, weekday_set={2, 0}, start_date="2026-09-01")
|
||||
assert a == b
|
||||
assert hash(a) == hash(b)
|
||||
|
||||
|
||||
def test_different_rules_compare_unequal() -> None:
|
||||
a = _rule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
b = _rule(rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01")
|
||||
assert a != b
|
||||
|
||||
|
||||
def test_rule_not_equal_to_other_types() -> None:
|
||||
rule = _rule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
assert rule != "daily"
|
||||
assert (rule == 42) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Purity guards: no HA imports, no database access
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_recurrence_module_imports_only_stdlib() -> None:
|
||||
"""The model must carry no Home Assistant and no DB imports."""
|
||||
from pathlib import Path
|
||||
|
||||
source = Path(
|
||||
__import__("custom_components.nestquest.recurrence",
|
||||
fromlist=["__file__"]).__file__
|
||||
).read_text()
|
||||
offenders = _scan_forbidden_imports(source)
|
||||
assert offenders == [], (
|
||||
f"recurrence.py must not hold forbidden imports: {offenders}"
|
||||
)
|
||||
|
||||
|
||||
def _scan_forbidden_imports(source: str) -> list[str]:
|
||||
"""The ACTUAL purity-guard logic, shared by the model scan and the
|
||||
probe tests below: returns human-readable offenders (empty = clean).
|
||||
|
||||
Forbidden, at any relative level, for both Import and ImportFrom:
|
||||
- modules containing 'homeassistant' or 'sqlite'
|
||||
- modules whose LEAF name is a DB/store/schema/DAO module
|
||||
- aliases importing a DB/store/schema/DAO module by leaf name
|
||||
('from . import db', 'from custom_components.nestquest import db')
|
||||
"""
|
||||
import ast
|
||||
|
||||
forbidden_targets = {
|
||||
"db", "schema", "dao_children", "dao_rules", "dao_presence",
|
||||
"dao_instances", "migrations", "store",
|
||||
}
|
||||
offenders: list[str] = []
|
||||
tree = ast.parse(source)
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
lowered = alias.name.lower()
|
||||
leaf = lowered.split(".")[-1]
|
||||
if (
|
||||
"homeassistant" in lowered
|
||||
or "sqlite" in lowered
|
||||
or leaf in forbidden_targets
|
||||
):
|
||||
offenders.append(f"import {alias.name}")
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
module = (node.module or "").lower()
|
||||
module_leaf = module.split(".")[-1] if module else ""
|
||||
if (
|
||||
"homeassistant" in module
|
||||
or "sqlite" in module
|
||||
or (module and module_leaf in forbidden_targets)
|
||||
or (node.level and module_leaf in forbidden_targets)
|
||||
):
|
||||
offenders.append(
|
||||
f"from {node.module!r} import ... (level={node.level})"
|
||||
)
|
||||
for alias in node.names:
|
||||
alias_leaf = alias.name.split(".")[-1].lower()
|
||||
if alias_leaf in forbidden_targets:
|
||||
offenders.append(
|
||||
f"from {node.module!r} import {alias.name} "
|
||||
f"(level={node.level})"
|
||||
)
|
||||
return offenders
|
||||
|
||||
|
||||
def test_purity_guard_logic_catches_probe_statements(tmp_path) -> None:
|
||||
"""The shared scan function catches every forbidden import shape,
|
||||
including level-0 absolute package DB imports and 'from . import db'.
|
||||
"""
|
||||
probes = [
|
||||
"from homeassistant.core import HomeAssistant",
|
||||
"import sqlite3",
|
||||
"from .db import NestQuestDatabase",
|
||||
"from custom_components.nestquest.db import NestQuestDatabase",
|
||||
"from custom_components.nestquest import db",
|
||||
"import custom_components.nestquest.dao_rules",
|
||||
"from . import db",
|
||||
"from . import schema",
|
||||
"from custom_components.nestquest.migrations import apply_migrations",
|
||||
]
|
||||
for probe in probes:
|
||||
offenders = _scan_forbidden_imports(probe)
|
||||
assert offenders, f"probe must be caught: {probe!r}"
|
||||
|
||||
|
||||
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=RuleType.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_direct_set_with_int_subclass_entry_rejected() -> None:
|
||||
"""A hashable-but-non-plain int subclass is still not a plain int."""
|
||||
|
||||
class IntChild(int):
|
||||
pass
|
||||
|
||||
with pytest.raises(RuleValidationError, match="plain integers"):
|
||||
ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY,
|
||||
weekday_set={IntChild(1)},
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
|
||||
|
||||
def test_from_dict_with_int_subclass_entry_rejected() -> None:
|
||||
class IntChild(int):
|
||||
pass
|
||||
|
||||
with pytest.raises(RuleValidationError, match="plain integers"):
|
||||
ScheduleRule.from_dict(
|
||||
{"rule_type": "weekly", "weekday_set": [IntChild(1)]}
|
||||
)
|
||||
|
||||
|
||||
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."""
|
||||
with pytest.raises(RuleValidationError, match="unknown rule fields"):
|
||||
ScheduleRule.from_dict({"rule_type": "daily", 42: "value"})
|
||||
|
||||
|
||||
def test_storage_aliases_and_whitespace_are_not_model_names() -> None:
|
||||
"""The model boundary is exact names only: storage strings, padded
|
||||
and case variants are all rejected."""
|
||||
for bad in ("monthly", "custom", " DAILY ", "daily ", "DAILY"):
|
||||
with pytest.raises(RuleValidationError):
|
||||
ScheduleRule(rule_type=bad, start_date="2026-09-01")
|
||||
|
||||
|
||||
def test_unhashable_int_subclass_interval_rejected() -> None:
|
||||
class UnhashableInt(int):
|
||||
__hash__ = None
|
||||
|
||||
with pytest.raises(RuleValidationError):
|
||||
ScheduleRule(
|
||||
rule_type=RuleType.DAILY,
|
||||
interval=UnhashableInt(1),
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
|
||||
|
||||
def test_rule_type_is_a_dataclass() -> None:
|
||||
import dataclasses
|
||||
|
||||
assert dataclasses.is_dataclass(ScheduleRule)
|
||||
Reference in New Issue
Block a user