Compare commits

...
2 changed files with 685 additions and 0 deletions
+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)