Compare commits

...
3 changed files with 356 additions and 10 deletions
+52 -7
View File
@@ -39,11 +39,16 @@ reach storage):
- ``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.
Month-end policy (implemented in :func:`_occurs_monthly_day` and
documented for Feature 04): in each interval-eligible month, a
MONTHLY_DAY rule for a day that does not exist in that month — the
31st in April, the 30th or 29th in February, February 29th in a
non-leap year — fires on the LAST day of the month instead of being
skipped. A monthly chore still happens in every eligible month,
pinned to the month's end when its nominal date is absent; exactly one
day per eligible month matches the rule (nominal day when it exists,
month end when it does not). The interval anchors on nominal months
elapsed from start_date's month, so clamping never shifts the cadence.
"""
from __future__ import annotations
@@ -418,6 +423,40 @@ def _within_window(rule: ScheduleRule, target: datetime.date) -> bool:
return True
def _month_end(year: int, month: int) -> int:
"""The last calendar day of the given month (leap-aware)."""
if month == 12:
return 31
return (datetime.date(year, month + 1, 1)
- datetime.timedelta(days=1)).day
def _occurs_monthly_day(rule: ScheduleRule, target: datetime.date) -> bool:
"""MONTHLY_DAY evaluation with the documented month-end policy.
Policy: in each interval-eligible month, a rule for a day-of-month
that does not exist in that month (the 31st in April, the 30th in
February, the 29th of February in a non-leap year) fires on the
LAST day of that month instead of being skipped — the chore still
happens that month, pinned to the month's end when its nominal date
is absent.
The interval anchors on months ELAPSED from start_date's month: the
rule fires when the months between the anchor month and the target's
month form a multiple of the interval, keeping "every 2 months on
the 15th" aligned to the anchor month even when the firing lands on
a clamped month-end day.
"""
anchor = _parse_date(rule.start_date)
months_elapsed = (
(target.year - anchor.year) * 12 + (target.month - anchor.month)
)
if months_elapsed % rule.interval != 0:
return False
month_end = _month_end(target.year, target.month)
return target.day == min(rule.day_of_month, month_end)
def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
"""Return True when ``rule`` fires on ``target``.
@@ -428,7 +467,10 @@ def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
# 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 not in (RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS):
if shape not in (
RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS,
RuleType.MONTHLY_DAY,
):
raise NotImplementedError(
f"occurs_on for {shape} lands with its own engine task "
"(Features 04 task sequence)"
@@ -438,6 +480,8 @@ def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
if shape is RuleType.DAILY:
offset = (target - _parse_date(rule.start_date)).days
return offset % rule.interval == 0
if shape is RuleType.MONTHLY_DAY:
return _occurs_monthly_day(rule, target)
# WEEKLY / CUSTOM_DAYS: whole-week anchoring from start_date, never
# ISO week numbers (Feature 04/05 guardrail: 53-week years flip ISO
# parity on January 1st; anchor arithmetic does not).
@@ -462,7 +506,8 @@ def occurrences_between(
rule.end_date] yields an empty list.
"""
if rule.rule_type not in (
RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS
RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS,
RuleType.MONTHLY_DAY,
):
raise NotImplementedError(
f"occurrences_between for {rule.rule_type} lands with its own "
+3 -3
View File
@@ -217,7 +217,7 @@ def test_occurs_on_rejects_unimplemented_shapes_loudly() -> None:
"""Monthly/yearly land with their own engine tasks; a loud
NotImplementedError beats a silent wrong answer."""
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
rule_type=RuleType.YEARLY, month=6,
start_date="2026-09-01",
)
with pytest.raises(NotImplementedError):
@@ -273,7 +273,7 @@ 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.MONTHLY_DAY, day_of_month=15,
rule_type=RuleType.YEARLY, month=6,
start_date="2026-09-01", end_date="2026-09-10",
)
# Before start, inside window, after end: all must raise.
@@ -286,7 +286,7 @@ def test_occurrences_between_unimplemented_shape_raises_even_disjoint(
tmp_path,
) -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
rule_type=RuleType.YEARLY, month=6,
start_date="2026-09-01"
)
# A range entirely before the rule's start still raises: the shape
+301
View File
@@ -0,0 +1,301 @@
"""Tests for MONTHLY_DAY recurrence with the month-end clamp policy."""
from __future__ import annotations
import datetime
import pytest
from custom_components.nestquest.recurrence import (
RuleType,
RuleValidationError,
ScheduleRule,
_month_end,
occurs_on,
occurrences_between,
)
def _d(date: str) -> datetime.date:
return datetime.date.fromisoformat(date)
# ---------------------------------------------------------------------------
# Nominal-day firings
# ---------------------------------------------------------------------------
def test_monthly_day_15_fires_on_the_15th() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
start_date="2026-01-15",
)
for month in range(1, 13):
assert occurs_on(rule, datetime.date(2026, month, 15)) is True
assert occurs_on(rule, datetime.date(2026, 1, 14)) is False
assert occurs_on(rule, datetime.date(2026, 1, 16)) is False
def test_monthly_day_never_fires_on_other_days() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=10,
start_date="2026-01-10",
)
# Every non-10th day of a short month must be False.
for day in range(1, 30):
if day != 10:
assert occurs_on(
rule, datetime.date(2026, 1, day)
) is False, f"Jan {day}"
def test_monthly_day_never_fires_before_start() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
start_date="2026-03-15",
)
# Feb 15 precedes the start: no firing even though the day matches.
assert occurs_on(rule, datetime.date(2026, 2, 15)) is False
assert occurs_on(rule, datetime.date(2026, 3, 15)) is True
def test_monthly_day_respects_end_date() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
start_date="2026-01-15", end_date="2026-06-15",
)
assert occurs_on(rule, datetime.date(2026, 6, 15)) is True
assert occurs_on(rule, datetime.date(2026, 7, 15)) is False
# ---------------------------------------------------------------------------
# Month-end clamp policy
# ---------------------------------------------------------------------------
def test_month_end_helper_is_leap_aware() -> None:
assert _month_end(2026, 1) == 31
assert _month_end(2026, 4) == 30
assert _month_end(2026, 2) == 28 # non-leap
assert _month_end(2028, 2) == 29 # leap
assert _month_end(2000, 2) == 29 # divisible-by-400 leap
assert _month_end(1900, 2) == 28 # divisible-by-100 non-leap
assert _month_end(2026, 12) == 31
def test_day_31_clamps_to_month_end_in_shorter_months() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-01-31",
)
# Months with a 31st: fire on the 31st.
for month, day in ((1, 31), (3, 31), (5, 31), (7, 31), (8, 31),
(10, 31), (12, 31)):
assert occurs_on(rule, datetime.date(2026, month, day)) is True, (
f"2026-{month:02d}-{day:02d}"
)
# Months WITHOUT a 31st: fire on the month's last day.
assert occurs_on(rule, datetime.date(2026, 4, 30)) is True # Apr 30
assert occurs_on(rule, datetime.date(2026, 6, 30)) is True # Jun 30
assert occurs_on(rule, datetime.date(2026, 9, 30)) is True # Sep 30
assert occurs_on(rule, datetime.date(2026, 11, 30)) is True # Nov 30
# The day BEFORE the clamp is not a firing.
assert occurs_on(rule, datetime.date(2026, 4, 29)) is False
assert occurs_on(rule, datetime.date(2026, 4, 28)) is False
def test_day_30_clamps_to_feb_28_non_leap() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=30,
start_date="2026-01-30",
)
assert occurs_on(rule, datetime.date(2026, 2, 28)) is True # clamp
# No Feb 29 in 2026 (non-leap): nothing else in February fires.
assert occurs_on(rule, datetime.date(2026, 2, 27)) is False
# Months with a 30th: nominal day.
assert occurs_on(rule, datetime.date(2026, 4, 30)) is True
assert occurs_on(rule, datetime.date(2026, 3, 30)) is True
def test_day_29_feb_clamps_to_28_non_leap_and_29_leap() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=29,
start_date="2026-01-29",
)
# Non-leap 2027: Feb 28 is the clamp.
assert occurs_on(rule, datetime.date(2027, 2, 28)) is True
assert occurs_on(rule, datetime.date(2027, 2, 27)) is False
# Leap 2028: Feb 29 exists — nominal day.
assert occurs_on(rule, datetime.date(2028, 2, 29)) is True
assert occurs_on(rule, datetime.date(2028, 2, 28)) is False
def test_day_30_feb_leap_year_clamps_to_29() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=30,
start_date="2027-01-30",
)
# Leap 2028: Feb has 29 days; the 30th clamps to Feb 29.
assert occurs_on(rule, datetime.date(2028, 2, 29)) is True
# Non-leap 2026 (rule starts 2027 so Feb 2027 is non-leap): clamps 28.
assert occurs_on(rule, datetime.date(2027, 2, 28)) is True
# ---------------------------------------------------------------------------
# Full-year sweeps: the 29th, 30th, 31st across leap and non-leap years
# ---------------------------------------------------------------------------
def _sweep_monthly(rule: ScheduleRule, year: int) -> list[tuple[int, int]]:
"""Every (month, day) in ``rule``'s year the engine says fires."""
hits = []
for month in range(1, 13):
for day in range(1, _month_end(year, month) + 1):
if occurs_on(rule, datetime.date(year, month, day)):
hits.append((month, day))
return hits
def test_full_year_day_31_sweep_2026() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-01-31",
)
# Exactly one firing per month, twelve total.
hits = _sweep_monthly(rule, 2026)
assert len(hits) == 12
by_month = dict(hits)
assert (by_month[1], by_month[2], by_month[4]) == (31, 28, 30)
assert by_month[2] == 28 # 2026 Feb is non-leap
assert by_month[4] == 30
# Each month has exactly one firing.
months = [month for month, _ in hits]
assert months == list(range(1, 13))
def test_full_year_day_31_sweep_2028_leap() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2028-01-31",
)
hits = _sweep_monthly(rule, 2028)
by_month = dict(hits)
assert by_month[2] == 29 # leap February clamps to the 29th
assert by_month[4] == 30
months = [month for month, _ in hits]
assert months == list(range(1, 13))
def test_full_year_day_30_sweep_2028() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=30,
start_date="2028-01-30",
)
hits = _sweep_monthly(rule, 2028)
by_month = dict(hits)
assert by_month[2] == 29 # leap Feb clamps to 29
months = [month for month, _ in hits]
assert months == list(range(1, 13))
def test_full_year_day_29_sweep_2027_non_leap() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=29,
start_date="2027-01-29",
)
hits = _sweep_monthly(rule, 2027)
by_month = dict(hits)
assert by_month[2] == 28 # non-leap Feb clamps to the 28th
months = [month for month, _ in hits]
assert months == list(range(1, 13))
def test_full_year_day_29_sweep_2028_leap() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=29,
start_date="2028-01-29",
)
hits = _sweep_monthly(rule, 2028)
by_month = dict(hits)
assert by_month[2] == 29 # leap Feb has a real 29th: nominal day
months = [month for month, _ in hits]
assert months == list(range(1, 13))
# ---------------------------------------------------------------------------
# Interval arithmetic
# ---------------------------------------------------------------------------
def test_monthly_interval_2_every_second_month() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15, interval=2,
start_date="2026-01-15",
)
assert occurs_on(rule, datetime.date(2026, 1, 15)) is True
assert occurs_on(rule, datetime.date(2026, 2, 15)) is False
assert occurs_on(rule, datetime.date(2026, 3, 15)) is True
assert occurs_on(rule, datetime.date(2026, 4, 15)) is False
assert occurs_on(rule, datetime.date(2026, 5, 15)) is True
def test_monthly_interval_2_clamped_month_end_alignment() -> None:
"""Interval counts nominal months from the anchor: Jan 31 anchor,
every 2 months -> Jan (m0), Mar (m2), May (m4), Jul (m6). Clamped
firings (Feb 28, Apr 30, Jun 30) occur only in interval months."""
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31, interval=2,
start_date="2026-01-31",
)
assert occurs_on(rule, datetime.date(2026, 1, 31)) is True # m0
assert occurs_on(rule, datetime.date(2026, 2, 28)) is False # m1
assert occurs_on(rule, datetime.date(2026, 3, 31)) is True # m2
assert occurs_on(rule, datetime.date(2026, 4, 30)) is False # m3
assert occurs_on(rule, datetime.date(2026, 5, 31)) is True # m4
assert occurs_on(rule, datetime.date(2026, 6, 30)) is False # m5
assert occurs_on(rule, datetime.date(2026, 7, 31)) is True # m6
def test_monthly_interval_arithmetic_across_year_boundary() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15, interval=2,
start_date="2026-11-15",
)
# Nov (m0) fires, Jan 2027 (m2) fires, Mar (m4) fires.
assert occurs_on(rule, datetime.date(2026, 11, 15)) is True
assert occurs_on(rule, datetime.date(2026, 12, 15)) is False
assert occurs_on(rule, datetime.date(2027, 1, 15)) is True
assert occurs_on(rule, datetime.date(2027, 2, 15)) is False
assert occurs_on(rule, datetime.date(2027, 3, 15)) is True
def test_monthly_day_occurrences_between_span() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-01-31",
)
result = occurrences_between(rule, "2026-01-01", "2026-06-30")
assert result == [
"2026-01-31", "2026-02-28", "2026-03-31", "2026-04-30",
"2026-05-31", "2026-06-30",
]
def test_monthly_day_occurrences_between_rejects_inverted() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
start_date="2026-01-15",
)
with pytest.raises(RuleValidationError, match="on or after"):
occurrences_between(rule, "2026-06-15", "2026-01-15")
def test_monthly_day_clamp_is_one_firing_per_month(tmp_path) -> None:
"""The clamp must not double-fire: only one day per month matches."""
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-01-31",
)
hits = _sweep_monthly(rule, 2026)
months = [month for month, _ in hits]
assert months == sorted(months)
assert len(months) == len(set(months)) == 12