"""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