Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
725c3158df | ||
|
|
058f414405 | ||
|
|
b4c9ad35c3 | ||
|
|
4230c1d0d4 |
@@ -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
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user