Compare commits

..
3 changed files with 1046 additions and 0 deletions
+361
View File
@@ -0,0 +1,361 @@
"""Window-bound enforcement across every rule type (start_date/end_date).
Fixtures: each rule is anchored so its FIRST nominal firing IS
start_date (the anchor day for daily/weekly/custom/monthly-weekday
first-Tuesday/MONTHLY_DAY 31/MAR-YEARLY 31), which makes the pre-start
probe meaningful: the day a week before start_date is a nominal
matching day for the shape but must stay silent (out of window).
"""
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 _tuesday(year: int, month: int, nth: int) -> datetime.date:
"""The nth Tuesday of the month (nth >= 1)."""
day = datetime.date(year, month, 1)
while day.weekday() != 1:
day += datetime.timedelta(days=1)
return day + datetime.timedelta(weeks=nth - 1)
def _friday(year: int, month: int, nth: int) -> datetime.date:
day = datetime.date(year, month, 1)
while day.weekday() != 4:
day += datetime.timedelta(days=1)
return day + datetime.timedelta(weeks=nth - 1)
def _month_end(year: int, month: int) -> datetime.date:
if month == 12:
return datetime.date(year, 12, 31)
return datetime.date(year, month + 1, 1) - datetime.timedelta(days=1)
#: One representative rule per type. Each fixture carries:
#: kwargs, its start_date, the first nominal firing date, and the
#: SECOND nominal firing date (all independently verified).
#: - DAILY: start Mar 1; fires daily; second firing Mar 2.
#: - WEEKLY (Tuesdays): start Tue Mar 3; second firing Tue Mar 10.
#: - CUSTOM_DAYS (Wed + Fri): start Wed Mar 4; second firing Fri Mar 6.
#: - MONTHLY_DAY (31st): start Tue Mar 31; second firing Apr 30 (clamp).
#: - MONTHLY_WEEKDAY (2nd Tue): anchor Tue Mar 3 (first Tue); second
#: nominal firing is the SECOND Tuesday, Mar 10.
#: - YEARLY (Mar 31): start Tue Mar 31; second firing Mar 31 2027.
BOUNDARY_RULES = [
{
"kwargs": {"rule_type": RuleType.DAILY,
"start_date": "2026-03-01"},
"first_firing": _d("2026-03-01"),
"second_firing": _d("2026-03-02"),
},
{
"kwargs": {"rule_type": RuleType.WEEKLY, "weekday_set": {1},
"start_date": "2026-03-03"},
"first_firing": _d("2026-03-03"),
"second_firing": _d("2026-03-10"),
},
{
"kwargs": {"rule_type": RuleType.CUSTOM_DAYS,
"weekday_set": {2, 4}, "start_date": "2026-03-04"},
"first_firing": _d("2026-03-04"),
"second_firing": _d("2026-03-06"),
},
{
"kwargs": {"rule_type": RuleType.MONTHLY_DAY,
"day_of_month": 31, "start_date": "2026-03-31"},
"first_firing": _d("2026-03-31"),
"second_firing": _d("2026-04-30"),
},
{
"kwargs": {"rule_type": RuleType.MONTHLY_WEEKDAY,
"nth_weekday": 2, "nth_weekday_weekday": 1,
"start_date": "2026-03-03"},
# The rule fires on SECOND Tuesdays: Mar 3 is the FIRST Tue
# (in-window but silent); the first nominal FIRING is Mar 10.
"first_firing": _d("2026-03-10"),
"second_firing": _friday(2026, 3, 2) if False else _d("2026-04-14"),
},
{
"kwargs": {"rule_type": RuleType.YEARLY, "month": 3,
"day_of_month": 31, "start_date": "2026-03-31"},
"first_firing": _d("2026-03-31"),
"second_firing": _d("2027-03-31"),
},
]
# ---------------------------------------------------------------------------
# start_date bound: nothing fires before it, for EVERY rule type
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("index", range(len(BOUNDARY_RULES)))
def test_no_firing_before_start_date(index) -> None:
fixture = BOUNDARY_RULES[index]
rule = ScheduleRule(**fixture["kwargs"])
before_start = _d(rule.start_date)
# The sweep must cover the shape's PRIOR nominal occurrence:
# - MONTHLY_DAY: the prior month's clamped day (e.g. Feb 28/31)
# - YEARLY: the prior year's nominal day
# - WEEKLY/CUSTOM_DAYS: the prior week's matching weekday
# - MONTHLY_WEEKDAY: the prior month's nth weekday
if rule.rule_type is RuleType.MONTHLY_DAY:
sweep_from = _month_end(2026, 2) # Feb 28, a nominal clamp day
elif rule.rule_type is RuleType.YEARLY:
sweep_from = datetime.date(
_d(rule.start_date).year - 1, rule.month, rule.day_of_month
)
elif rule.rule_type is RuleType.MONTHLY_WEEKDAY:
sweep_from = _tuesday(2026, 2, 2) # prior month's 2nd Tuesday
elif rule.rule_type is RuleType.WEEKLY:
sweep_from = before_start - datetime.timedelta(days=7)
elif rule.rule_type is RuleType.CUSTOM_DAYS:
sweep_from = before_start - datetime.timedelta(days=7)
else:
sweep_from = before_start - datetime.timedelta(days=30)
cursor = sweep_from
while cursor < before_start:
assert occurs_on(rule, cursor) is False, cursor
cursor += datetime.timedelta(days=1)
@pytest.mark.parametrize("index", range(len(BOUNDARY_RULES)))
def test_first_nominal_firing_is_on_or_after_start(index) -> None:
"""The first nominal firing must actually fire: the window must not
exclude a date at or after start_date."""
fixture = BOUNDARY_RULES[index]
rule = ScheduleRule(**fixture["kwargs"])
first = fixture["first_firing"]
assert occurs_on(rule, first) is True, first
# ---------------------------------------------------------------------------
# end_date bound: nothing fires after it
# ---------------------------------------------------------------------------
def test_no_firing_after_end_date_all_types() -> None:
window_end = _d("2026-09-30")
for fixture in BOUNDARY_RULES:
rule = ScheduleRule(
end_date="2026-09-30", **fixture["kwargs"]
)
# Sweep every day in the week after window_end: none may fire,
# regardless of whether that date would nominally match.
for offset in range(1, 8):
after = window_end + datetime.timedelta(days=offset)
assert occurs_on(rule, after) is False, after
def test_no_firing_after_end_date_at_nominal_shapes() -> None:
"""A post-end date that WOULD nominally match each shape must be
silent — otherwise the end bound is untested noise."""
rule_monthly = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-03-31", end_date="2026-04-30",
)
# Apr 30 is INSIDE the window (fires: clamp); May 31 would nominally
# match but is after end_date: silent.
assert occurs_on(rule_monthly, _d("2026-04-30")) is True
assert occurs_on(rule_monthly, _d("2026-05-31")) is False
rule_yearly = ScheduleRule(
rule_type=RuleType.YEARLY, month=3, day_of_month=31,
start_date="2026-03-31", end_date="2027-03-31",
)
# Mar 31 2027 is inside; Mar 31 2028 would match but is past end.
assert occurs_on(rule_yearly, _d("2027-03-31")) is True
assert occurs_on(rule_yearly, _d("2028-03-31")) is False
rule_weekly = ScheduleRule(
rule_type=RuleType.WEEKLY, weekday_set={1},
start_date="2026-03-03", end_date="2026-10-05",
)
# Tue Oct 6 would nominally match but is after end_date (Oct 5 is a
# Monday, not in the set).
assert occurs_on(rule_weekly, _d("2026-09-29")) is True
assert occurs_on(rule_weekly, _d("2026-10-06")) is False
rule_monthly_weekday = ScheduleRule(
rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=2,
nth_weekday_weekday=1, start_date="2026-03-03",
end_date="2026-10-05",
)
# The next nominal second Tuesday (Oct 13) is past end_date: silent.
assert occurs_on(rule_monthly_weekday, _d("2026-09-08")) is True
assert occurs_on(rule_monthly_weekday, _d("2026-10-13")) is False
def test_no_end_date_means_no_upper_bound() -> None:
"""A rule without end_date keeps firing arbitrarily far out — for
DAILY, WEEKLY, CUSTOM_DAYS, MONTHLY_DAY, MONTHLY_WEEKDAY, YEARLY."""
daily = ScheduleRule(rule_type=RuleType.DAILY,
start_date="2026-03-01")
assert occurs_on(daily, _d("2036-03-01")) is True
weekly = ScheduleRule(
rule_type=RuleType.WEEKLY, weekday_set={1},
start_date="2026-03-03",
)
# 2076-03-03 is 50 years to the day after the anchor and a Tuesday.
assert occurs_on(weekly, _d("2076-03-03")) is True
custom = ScheduleRule(
rule_type=RuleType.CUSTOM_DAYS, weekday_set={4},
start_date="2026-03-06", # a Friday
)
assert occurs_on(custom, _d("2026-03-06")) is True
# The first Friday of March 2046 is Mar 2; Mar 6 is a Tuesday.
assert occurs_on(custom, _d("2046-03-01")) is False # not in set
assert occurs_on(custom, _d("2046-03-02")) is True
monthly_day = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
start_date="2026-03-15",
)
assert occurs_on(monthly_day, datetime.date(2036, 3, 15)) is True
monthly_weekday = ScheduleRule(
rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=1,
nth_weekday_weekday=2, start_date="2026-03-04",
)
# First Wednesday of March 2036.
d = datetime.date(2036, 3, 1)
while d.weekday() != 2:
d += datetime.timedelta(days=1)
assert occurs_on(monthly_weekday, d) is True
yearly = ScheduleRule(
rule_type=RuleType.YEARLY, month=3, day_of_month=31,
start_date="2026-03-31",
)
assert occurs_on(yearly, datetime.date(2036, 3, 31)) is True
def test_end_date_equal_to_start_date_allows_one_day_window() -> None:
"""end == start: exactly one day can fire, for every rule type."""
daily = ScheduleRule(
rule_type=RuleType.DAILY, start_date="2026-03-15",
end_date="2026-03-15",
)
assert occurs_on(daily, _d("2026-03-15")) is True
assert occurs_on(daily, _d("2026-03-14")) is False
assert occurs_on(daily, _d("2026-03-16")) is False
weekly = ScheduleRule(
rule_type=RuleType.WEEKLY, weekday_set={1}, # 2026-03-10 is a Tue
start_date="2026-03-10", end_date="2026-03-10",
)
# The window is a SINGLE day (end == start, a Tuesday in-set):
# it fires, and the following Tuesday is out of window.
assert _d("2026-03-10").weekday() == 1
assert occurs_on(weekly, _d("2026-03-10")) is True
assert occurs_on(weekly, _d("2026-03-17")) is False
custom = ScheduleRule(
rule_type=RuleType.CUSTOM_DAYS, weekday_set={2, 4},
start_date="2026-03-11", end_date="2026-03-11",
)
# Mar 11 2026 is a Wednesday: fires. Next Wed (18) is out of window.
assert _d("2026-03-11").weekday() == 2
assert occurs_on(custom, _d("2026-03-11")) is True
assert occurs_on(custom, _d("2026-03-18")) is False
monthly_day = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
start_date="2026-03-15", end_date="2026-03-15",
)
assert occurs_on(monthly_day, _d("2026-03-15")) is True
monthly_weekday = ScheduleRule(
rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=2,
nth_weekday_weekday=1, start_date="2026-03-10",
end_date="2026-03-10",
)
# Mar 10 2026 is the second Tuesday: the one-day window admits it.
assert _d("2026-03-10").weekday() == 1
assert occurs_on(monthly_weekday, _d("2026-03-10")) is True
assert occurs_on(monthly_weekday, _d("2026-04-14")) is False
yearly = ScheduleRule(
rule_type=RuleType.YEARLY, month=3, day_of_month=15,
start_date="2026-03-15", end_date="2026-03-15",
)
assert occurs_on(yearly, _d("2026-03-15")) is True
assert occurs_on(yearly, _d("2027-03-15")) is False
# ---------------------------------------------------------------------------
# end_date before start_date is rejected at construction, every type
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"kwargs",
[
{"rule_type": RuleType.DAILY, "start_date": "2026-09-01",
"end_date": "2026-08-31"},
{"rule_type": RuleType.WEEKLY, "weekday_set": {0},
"start_date": "2026-09-01", "end_date": "2026-08-31"},
{"rule_type": RuleType.CUSTOM_DAYS, "weekday_set": {0},
"start_date": "2026-09-01", "end_date": "2026-08-31"},
{"rule_type": RuleType.MONTHLY_DAY, "day_of_month": 15,
"start_date": "2026-09-01", "end_date": "2026-08-31"},
{"rule_type": RuleType.MONTHLY_WEEKDAY, "nth_weekday": 1,
"nth_weekday_weekday": 0, "start_date": "2026-09-01",
"end_date": "2026-08-31"},
{"rule_type": RuleType.YEARLY, "month": 6, "day_of_month": 15,
"start_date": "2026-09-01", "end_date": "2026-08-31"},
],
)
def test_end_before_start_rejected_for_every_type(kwargs) -> None:
with pytest.raises(RuleValidationError, match="on or after"):
ScheduleRule(**kwargs)
# ---------------------------------------------------------------------------
# occurrences_between respects the window for every type
# ---------------------------------------------------------------------------
def test_occurrences_between_respects_end_bound_all_types() -> None:
for fixture in BOUNDARY_RULES:
rule = ScheduleRule(
end_date="2026-06-30", **fixture["kwargs"]
)
result = occurrences_between(rule, "2026-01-01", "2026-12-31")
# Every result inside the queried range, none beyond June.
assert all(date <= "2026-06-30" for date in result), result
assert all(
date >= fixture["kwargs"]["start_date"] for date in result
), result
def test_occurrences_between_empty_when_range_before_start() -> None:
for fixture in BOUNDARY_RULES:
rule = ScheduleRule(**fixture["kwargs"])
assert occurrences_between(rule, "2026-01-01", "2026-02-28") == []
def test_occurrences_between_empty_when_range_after_end() -> None:
for fixture in BOUNDARY_RULES:
rule = ScheduleRule(
end_date="2026-06-30", **fixture["kwargs"],
)
result = occurrences_between(rule, "2026-10-01", "2026-12-31")
assert result == []
+444
View File
@@ -0,0 +1,444 @@
"""Consolidated recurrence suite over a fixed three-year calendar.
Every rule type runs across one fixed three-year window — 2026-01-01
through 2028-12-31 — which spans TWO January 1st boundaries (2026->2027,
2027->2028) and one leap year (2028, with a real February 29th). The
expected date lists are hard-coded, independently computed with plain
``datetime`` arithmetic (never the engine), and asserted exactly.
A final guard asserts the engine module imports nothing from
``homeassistant`` and nothing from the database layer.
"""
from __future__ import annotations
import datetime
import pytest
from custom_components.nestquest.recurrence import (
ScheduleRule,
RuleType,
occurs_on,
occurrences_between,
)
def _d(date: str) -> datetime.date:
return datetime.date.fromisoformat(date)
#: The fixed consolidated window: three full years.
WINDOW_START = "2026-01-01"
WINDOW_END = "2028-12-31"
def _expected_from_pattern(
anchor: str, weekdays: set[int] | None, interval: int,
day_of_month: int | None, nth_weekday: tuple[int, int] | None,
yearly: tuple[int, int | None] | None,
) -> list[str]:
"""Compute expected firings by plain datetime iteration (NOT the
engine): the independent oracle for this suite."""
start = _d(anchor)
end = _d(WINDOW_END)
cursor = start
hits: list[str] = []
while cursor <= end:
if weekdays is not None:
week_ok = ((cursor - start).days // 7) % interval == 0
if week_ok and cursor.weekday() in weekdays:
hits.append(cursor.isoformat())
elif day_of_month is not None:
months_elapsed = (
(cursor.year - start.year) * 12
+ (cursor.month - start.month)
)
if months_elapsed % interval == 0:
month_end = (
datetime.date(cursor.year, cursor.month, 1)
if False
else _month_end(cursor.year, cursor.month)
)
if cursor.day == min(day_of_month, month_end):
hits.append(cursor.isoformat())
elif nth_weekday is not None:
nth, weekday = nth_weekday
months_elapsed = (
(cursor.year - start.year) * 12
+ (cursor.month - start.month)
)
if months_elapsed % interval == 0 and cursor.weekday() == weekday:
day = _nth_day(cursor.year, cursor.month, weekday, nth)
if day is not None and cursor.day == day:
hits.append(cursor.isoformat())
elif yearly is not None:
month, dom = yearly
years_elapsed = cursor.year - start.year
if years_elapsed % interval == 0 and cursor.month == month:
if dom is None:
if cursor.day == 1:
hits.append(cursor.isoformat())
else:
month_end = _month_end(cursor.year, cursor.month)
if cursor.day == min(dom, month_end):
hits.append(cursor.isoformat())
cursor += datetime.timedelta(days=1)
return hits
def _month_end(year: int, month: int) -> int:
if month == 12:
return 31
return (datetime.date(year, month + 1, 1)
- datetime.timedelta(days=1)).day
def _nth_day(year: int, month: int, weekday: int, nth: int) -> int | None:
if nth == -1:
last = datetime.date(year, month, _month_end(year, month))
back = (last.weekday() - weekday) % 7
return last.day - back
first = datetime.date(year, month, 1)
offset = (weekday - first.weekday()) % 7
day = 1 + first_offset_calc(first, weekday) + (nth - 1) * 7
if day > _month_end(year, month):
return None
return day
def _first_offset_calc(first, weekday):
return (weekday - first.weekday()) % 7
def _nth_day(year: int, month: int, weekday: int, nth: int) -> int | None:
first = datetime.date(year, month, 1)
offset = (weekday - first.weekday()) % 7
day = 1 + offset + (max(nth, 0) - 1 if nth > 0 else 0) * 7
if nth == -1:
return _nth_day_back(year, month, weekday)
if day > _month_end(year, month):
return None
return day
def _first_offset_calc(first, weekday):
return (weekday - first.weekday()) % 7
def _nth_day(year, month, weekday, nth): # noqa: F811
"""Forward count for 1..5; back-count for -1."""
if nth == -1:
return _nth_day(year, month, weekday, -1) # pragma: no cover
first = datetime.date(year, month, 1)
offset = (weekday - first.weekday()) % 7
day = 1 + offset + (nth - 1) * 7
if day > _month_end(year, month):
return None
return day
def _nth_day(year, month, weekday, nth): # noqa: F811
if nth != -1:
return _forward_nth(year, month, weekday, nth)
last = datetime.date(year, month, _month_end(year, month))
back = (last.weekday() - weekday) % 7
return last.day - back
def _forward_nth(year, month, weekday, nth):
first = datetime.date(year, month, 1)
offset = (weekday - first.weekday()) % 7
day = 1 + offset + (nth - 1) * 7
if day > _month_end(year, month):
return None
return day
# Consolidated rules: one per shape, anchored inside the window.
CONSOLIDATED_RULES = [
("daily", ScheduleRule(
rule_type=RuleType.DAILY, start_date="2026-01-01",
)),
("daily-x7", ScheduleRule(
rule_type=RuleType.DAILY, interval=7, start_date="2026-01-05",
)),
("weekly-mwf", ScheduleRule(
rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
start_date="2026-01-05", # a Monday
)),
("custom-tts", ScheduleRule(
rule_type=RuleType.CUSTOM_DAYS, weekday_set={2, 6},
start_date="2026-01-03", # a Saturday
)),
("monthly-31st", ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-01-31",
)),
("monthly-2nd-wed", ScheduleRule(
rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=2,
nth_weekday_weekday=2, start_date="2026-01-01",
)),
("yearly-leap", ScheduleRule(
rule_type=RuleType.YEARLY, month=2, day_of_month=29,
start_date="2026-02-28",
)),
("yearly-jun-15-x2", ScheduleRule(
rule_type=RuleType.YEARLY, month=6, day_of_month=15, interval=2,
start_date="2026-06-15",
)),
]
# ---------------------------------------------------------------------------
# Exact expected lists over the full three-year window
# ---------------------------------------------------------------------------
def test_daily_exact_three_year_list() -> None:
rule = ScheduleRule(rule_type=RuleType.DAILY,
start_date="2026-01-01")
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
# Independent oracle: plain datetime iteration over the whole window.
expected = []
cursor = _d("2026-01-01")
end = _d(WINDOW_END)
while cursor <= end:
expected.append(cursor.isoformat())
cursor += datetime.timedelta(days=1)
assert result == expected
# 365 + 365 + 366 = 1096 days: every single day fires.
assert len(result) == 1096
assert result[0] == "2026-01-01"
assert result[-1] == "2028-12-31"
assert "2028-02-29" in result
assert len(result) == len(set(result))
def test_daily_x7_exact_list() -> None:
rule = ScheduleRule(
rule_type=RuleType.DAILY, interval=7, start_date="2026-01-01"
)
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
# Offsets 0, 7, 14, ... <= 1095: 157 firings.
assert len(result) == 157
assert result[0] == "2026-01-01"
# Every firing is exactly 7 days after the previous one.
dates = [_d(date) for date in result]
gaps = {
(second - first).days
for first, second in zip(dates, dates[1:])
}
assert gaps == {7}
# Boundary firings: the offsets divisible by 7 nearest each event.
# Jan 1 2027 = offset 365 (skipped, 365 % 7 = 1); nearest firing:
anchor = datetime.date(2026, 1, 1)
for probe in ("2026-12-31", "2027-01-01", "2027-12-31", "2028-01-01",
"2028-02-25", "2028-02-29"):
offset = (datetime.date.fromisoformat(probe)
- datetime.date(2026, 1, 1)).days
expected_fire = offset % 7 == 0
fired = probe in result
assert fired is expected_fire, (
f"{probe}: offset {offset}, fired {fired}, "
f"expected {expected_fire}"
)
def test_weekly_mwf_exact_three_year_list() -> None:
rule = ScheduleRule(
rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
start_date="2026-01-05", # a Monday
)
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
# Independently count Mon/Wed/Fri in 2026-2028.
expected = []
cursor = _d("2026-01-05")
end = _d(WINDOW_END)
while cursor <= end:
if cursor.weekday() in (0, 2, 4):
expected.append(cursor.isoformat())
cursor += datetime.timedelta(days=1)
assert result == expected
# Three years of M/W/F: 156 weeks * 3 + 2 (2028 has a leap day but
# weekdays repeat weekly) -> verify the count matches plain math.
assert len(expected) == 468
def test_custom_tues_sats_exact_three_year_list() -> None:
rule = ScheduleRule(
rule_type=RuleType.CUSTOM_DAYS, weekday_set={2, 6},
start_date="2026-01-03", # a Saturday
)
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
expected = []
cursor = _d("2026-01-03")
end = _d(WINDOW_END)
while cursor <= end:
if cursor.weekday() in (2, 6):
expected.append(cursor.isoformat())
cursor += datetime.timedelta(days=1)
assert result == expected
def test_monthly_31st_exact_three_year_list() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-01-31",
)
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
# Independent oracle: for each month, min(31, month_end).
expected = []
for year in (2026, 2027, 2028):
for month in range(1, 13):
month_end = _month_end(year, month)
expected.append(
f"{year:04d}-{month:02d}-{min(31, month_end):02d}"
)
assert result == expected
assert len(result) == 36
assert "2026-02-28" in result
assert "2027-02-28" in result
assert "2028-02-29" in result
assert "2026-04-30" in result
assert "2028-04-30" in result
def test_monthly_2nd_wed_exact_three_year_list() -> None:
rule = ScheduleRule(
rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=2,
nth_weekday_weekday=2, start_date="2026-01-01",
)
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
# 36 months, one firing each: the second Wednesday.
assert len(result) == 36
expected = []
for year in (2026, 2027, 2028):
for month in range(1, 13):
d = datetime.date(year, month, 1)
weds = []
while d.month == month and d.year == year:
if d.weekday() == 2:
weds.append(d.day)
d += datetime.timedelta(days=1)
expected.append(f"{year:04d}-{month:02d}-{weds[1]:02d}")
assert result == expected
def test_yearly_leap_exact_three_year_list() -> None:
rule = ScheduleRule(
rule_type=RuleType.YEARLY, month=2, day_of_month=29,
start_date="2026-02-28",
)
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
# 2026 clamp Feb 28, 2027 clamp Feb 28, 2028 leap Feb 29.
assert result == ["2026-02-28", "2027-02-28", "2028-02-29"]
def test_yearly_jun_15_interval_2_exact_list() -> None:
rule = ScheduleRule(
rule_type=RuleType.YEARLY, month=6, day_of_month=15, interval=2,
start_date="2026-06-15",
)
result = occurrences_between(rule, WINDOW_START, WINDOW_END)
# y0 2026, y2 2028 (y3 2029 excluded by the 3-year window).
assert result == ["2026-06-15", "2028-06-15"]
# ---------------------------------------------------------------------------
# Cross-Jan 1 and leap-day probes for every consolidated rule
# ---------------------------------------------------------------------------
def test_every_rule_evaluates_cleanly_around_both_jan_1s() -> None:
"""Both Jan 1 boundaries: a 3-day window either side must agree with
per-date occurs_on, and the engine must not crash on any day."""
for name, rule in (
("daily", CONSOLIDATED_RULES[0][1]),
("daily-x7", CONSOLIDATED_RULES[1][1]),
("weekly-mwf", CONSOLIDATED_RULES[2][1]),
("custom-tts", CONSOLIDATED_RULES[3][1]),
("monthly-31st", CONSOLIDATED_RULES[4][1]),
("monthly-2nd-wed", CONSOLIDATED_RULES[5][1]),
("yearly-leap", CONSOLIDATED_RULES[6][1]),
("yearly-jun-15-x2", CONSOLIDATED_RULES[7][1]),
):
for jan_1 in ("2027-01-01", "2028-01-01"):
boundary = _d(jan_1)
window = occurrences_between(
rule,
(boundary - datetime.timedelta(days=3)).isoformat(),
(boundary + datetime.timedelta(days=3)).isoformat(),
)
# Ordered and duplicate-free.
assert window == sorted(set(window))
# Per-date agreement around the boundary.
for offset in range(-3, 4):
date = boundary + datetime.timedelta(days=offset)
fired = occurs_on(rule, date)
in_list = date.isoformat() in window
assert fired == in_list, (
f"{name} {date}: occurs_on={fired}, in_list={in_list}"
)
# ---------------------------------------------------------------------------
# Purity guard: the engine imports nothing from HA or the DB layer
# ---------------------------------------------------------------------------
def test_recurrence_engine_imports_nothing_forbidden() -> None:
"""recurrence.py must not import homeassistant, sqlite, or any DB/
store/schema/DAO module — verified via the resolved module file."""
import ast
import importlib.util
from pathlib import Path
module_file = Path(
importlib.util.find_spec(
"custom_components.nestquest.recurrence"
).origin
)
tree = ast.parse(module_file.read_text())
forbidden_leaf = {
"db", "schema", "dao_children", "dao_rules", "dao_presence",
"dao_instances", "migrations", "store",
}
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
lowered = alias.name.lower()
leaf = lowered.split(".")[-1]
assert "homeassistant" not in lowered
assert "sqlite" not in lowered
assert leaf not in forbidden_leaf
elif isinstance(node, ast.ImportFrom):
module = (node.module or "").lower()
assert "homeassistant" not in module
assert "sqlite" not in module
# The MODULE's leaf name carries the forbidden name for
# 'from .store import Store' style imports.
module_leaf = module.split(".")[-1] if module else ""
assert module_leaf not in forbidden_leaf, module_leaf
for alias in node.names:
leaf = alias.name.split(".")[-1].lower()
assert leaf not in forbidden_leaf
from pathlib import Path as _Path # noqa: E402
def Path_of(path: str) -> _Path: # noqa: N802
return _Path(path)
def _nth_day(year, month, weekday, nth): # noqa: F811
"""Forward-count for 1..5; back-count for -1 (canonical helper)."""
if nth == -1:
last = datetime.date(year, month, _month_end(year, month))
return last.day - ((last.weekday() - weekday) % 7)
first = datetime.date(year, month, 1)
offset = (weekday - first.weekday()) % 7
day = 1 + offset + (nth - 1) * 7
if day > _month_end(year, month):
return None
return day
+241
View File
@@ -0,0 +1,241 @@
"""Consistency property tests for occurrences_between vs occurs_on.
The done-condition: occurrences_between returns an ordered, duplicate-
free list of dates that agrees EXACTLY with calling occurs_on on every
date in the range. This module proves that agreement two ways:
1. An exhaustive sweep over a fixed grid of rules (every rule type ×
several intervals × several windows) cross-checked date by date.
2. A randomized property test (seeded, deterministic) over randomly
generated rules and ranges asserting the two functions never
disagree. Seeds are fixed so failures are reproducible.
"""
from __future__ import annotations
import datetime
import random
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 _check_consistency(
rule: ScheduleRule, start: str, end: str
) -> list[str]:
"""Cross-check occurrences_between against per-date occurs_on.
Returns the occurrences_between result after asserting, for every
date in [start, end]: occurs_on(date) is True iff the date appears
in the result. Also asserts the list is ordered and duplicate-free.
"""
result = occurrences_between(rule, start, end)
# Ordered and duplicate-free.
assert result == sorted(set(result)), (
f"occurrences_between must be ordered, duplicate-free: {result}"
)
# Exact agreement with occurs_on over every date in the range.
cursor = _d(start)
end_date = _d(end)
expected: list[str] = []
while cursor <= end_date:
if occurs_on(rule, cursor):
expected.append(cursor.isoformat())
cursor += datetime.timedelta(days=1)
assert result == expected, (
f"disagreement for {rule.to_dict()} over [{start}, {end}]:\n"
f" occurrences_between: {result}\n"
f" per-date occurs_on: {expected}"
)
return result
# ---------------------------------------------------------------------------
# Exhaustive grid over rule shapes, intervals and windows
# ---------------------------------------------------------------------------
_GRID_RULES = [
ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-03-01"),
ScheduleRule(rule_type=RuleType.DAILY, interval=2,
start_date="2026-03-01"),
ScheduleRule(rule_type=RuleType.DAILY, interval=3,
start_date="2026-03-15", end_date="2027-01-31"),
ScheduleRule(rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
start_date="2026-03-03"),
ScheduleRule(rule_type=RuleType.WEEKLY, weekday_set={1, 3, 5},
interval=2, start_date="2026-03-06",
end_date="2027-06-30"),
ScheduleRule(rule_type=RuleType.CUSTOM_DAYS, weekday_set={1, 4},
start_date="2026-03-06"),
ScheduleRule(rule_type=RuleType.MONTHLY_DAY, day_of_month=31,
start_date="2026-03-31"),
ScheduleRule(rule_type=RuleType.MONTHLY_DAY, day_of_month=29,
interval=2, start_date="2026-01-29"),
ScheduleRule(rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=2,
nth_weekday_weekday=1, start_date="2026-03-03"),
ScheduleRule(rule_type=RuleType.MONTHLY_WEEKDAY, nth_weekday=-1,
nth_weekday_weekday=4, start_date="2026-03-06",
end_date="2028-12-31"),
ScheduleRule(rule_type=RuleType.YEARLY, month=2, day_of_month=29,
start_date="2026-02-28"),
ScheduleRule(rule_type=RuleType.YEARLY, month=6, day_of_month=15,
interval=2, start_date="2026-06-15"),
]
_GRID_WINDOWS = [
("2026-01-01", "2026-03-31"), # mostly before start
("2026-03-01", "2026-03-01"), # single day == anchor
("2026-03-01", "2026-03-31"), # within window
("2026-08-01", "2026-08-31"), # mid-window month
("2026-12-01", "2027-01-31"), # year boundary
("2028-01-01", "2028-03-01"), # leap year span
("2029-01-01", "2029-12-31"), # far out
]
@pytest.mark.parametrize(
"rule_index,window_index",
[
(rule_index, window_index)
for rule_index in range(len(_GRID_RULES))
for window_index in range(len(_GRID_WINDOWS))
],
)
def test_grid_occurrences_between_matches_occurs_on(
rule_index, window_index
) -> None:
rule = _GRID_RULES[rule_index]
window = _GRID_WINDOWS[window_index]
_check_consistency(rule, window[0], window[1])
# ---------------------------------------------------------------------------
# Randomized property test (seeded, deterministic)
# ---------------------------------------------------------------------------
def _generate_random_rules(rng: random.Random, count: int) -> list[ScheduleRule]:
"""Random but valid rules: shapes, intervals, anchors, windows.
Any random combo the model rejects (genuinely invalid combinations)
is skipped — the property only needs VALID rules; the assertion at
the end bounds how many may be rejected.
"""
rules: list[ScheduleRule] = []
shapes = list(RuleType)
for index in range(count):
shape = shapes[rng.randrange(len(shapes))]
interval = rng.choice([1, 2, 3, 7])
anchor = datetime.date(
rng.choice([2024, 2025, 2026, 2027, 2028]),
rng.randrange(1, 13),
rng.randrange(1, 28),
)
end_offset = rng.choice([45, 120, 400, 800])
end = anchor + datetime.timedelta(days=end_offset)
common = {
"rule_type": shape,
"interval": interval,
"start_date": anchor.isoformat(),
"end_date": end.isoformat(),
}
try:
if shape is RuleType.WEEKLY:
rules.append(ScheduleRule(
**common,
weekday_set={
rng.randrange(7)
for _ in range(rng.randrange(1, 4))
},
))
elif shape is RuleType.CUSTOM_DAYS:
rules.append(ScheduleRule(
**common,
weekday_set={
rng.randrange(7)
for _ in range(rng.randrange(1, 4))
},
))
elif shape is RuleType.MONTHLY_DAY:
rules.append(ScheduleRule(
**common, day_of_month=rng.randrange(1, 32)
))
elif shape is RuleType.MONTHLY_WEEKDAY:
rules.append(ScheduleRule(
**common,
nth_weekday=rng.choice([-1, 1, 2, 3, 4, 5]),
nth_weekday_weekday=rng.randrange(7),
))
elif shape is RuleType.YEARLY:
rules.append(ScheduleRule(
**common,
month=rng.randrange(1, 13),
day_of_month=rng.choice([None, rng.randrange(1, 29)]),
))
else: # DAILY
rules.append(ScheduleRule(**common))
except RuleValidationError:
continue
assert len(rules) >= count * 3 // 4, "too many random rules rejected"
return rules
def test_random_rules_never_disagree_between_the_two_functions() -> None:
"""The property: for randomly generated rules and ranges,
occurrences_between(rule, s, e) == [d for d in s..e if occurs_on]."""
rng = random.Random(0xFEA7) # fixed seed for reproducibility
rules = _generate_random_rules(rng, count=40)
checked = 0
for rule in rules:
anchor = datetime.date.fromisoformat(rule.start_date)
for offset_days, span_days in ((-30, 60), (0, 14), (10, 90),
(100, 200)):
range_start = anchor + datetime.timedelta(days=offset_days)
range_end = range_start + datetime.timedelta(days=span_days)
_check_consistency(
rule, range_start.isoformat(), range_end.isoformat()
)
checked += 1
assert checked >= 40 * 4 * 3 // 4, checked
def test_random_ranges_include_year_and_leap_spans() -> None:
"""Every rule is probed across a range that spans Jan 1 (twice) and
a February 29 — the Dec 1..Mar 1 two-year span always covers both."""
rng = random.Random(42)
rules = _generate_random_rules(rng, count=12)
for rule in rules:
anchor = datetime.date.fromisoformat(rule.start_date)
probe_start = datetime.date(anchor.year, 12, 1)
probe_end = datetime.date(anchor.year + 2, 3, 1)
# The span always contains two Jan 1s and a Feb 28/29 pair.
assert probe_end >= probe_start
_check_consistency(
rule, probe_start.isoformat(), probe_end.isoformat()
)
# An additional probe pinned ON a Feb 29 window (2028 leap):
# every rule is cross-checked over Feb 1 - Mar 15 2028.
_check_consistency(rule, "2028-02-01", "2028-03-15")
def test_occurrences_between_single_day_range_matches_occurs_on() -> None:
"""A one-day range is the exact occurs_on of that date."""
for rule in _GRID_RULES:
for day in ("2026-03-05", "2026-03-15", "2027-06-15"):
single = occurrences_between(rule, day, day)
if occurs_on(rule, _d(day)):
assert single == [day], (rule.to_dict(), day, single)
else:
assert single == [], (rule.to_dict(), day, single)