Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6246c73c90 | ||
|
|
d54156198e | ||
|
|
72d75b5b88 | ||
|
|
3b638cae6f |
@@ -387,3 +387,107 @@ class ScheduleRule:
|
||||
start_date=data.get("start_date", "1970-01-01"),
|
||||
end_date=data.get("end_date"),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Evaluation engine (Feature 04): pure date arithmetic on ScheduleRule
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _parse_date(value: str) -> datetime.date:
|
||||
"""Parse a strict ISO date already validated by the model."""
|
||||
return datetime.datetime.strptime(value, _DATE_FORMAT).date()
|
||||
|
||||
|
||||
def _anchor_offset(rule: ScheduleRule, target: datetime.date) -> int:
|
||||
"""Day offset from the rule's anchor date to ``target``.
|
||||
|
||||
The anchor for DAILY/WEEKLY-interval arithmetic is start_date: the
|
||||
first day the rule CAN fire. A negative offset means the target
|
||||
precedes the rule's window entirely.
|
||||
"""
|
||||
return (target - _parse_date(rule.start_date)).days
|
||||
|
||||
|
||||
def _within_window(rule: ScheduleRule, target: datetime.date) -> bool:
|
||||
"""True when ``target`` is inside the rule's [start, end] window."""
|
||||
if target < _parse_date(rule.start_date):
|
||||
return False
|
||||
if rule.end_date is not None and target > _parse_date(rule.end_date):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
|
||||
"""Return True when ``rule`` fires on ``target``.
|
||||
|
||||
Pure function: no I/O, no timezone awareness, no mutation. Dates
|
||||
are plain calendar dates (Feature 04 guardrail); the caller decides
|
||||
the timezone that produced them.
|
||||
"""
|
||||
# Dispatch FIRST: an unimplemented shape must raise loudly even for
|
||||
# dates outside the window, never silently evaluate to False.
|
||||
shape = rule.rule_type
|
||||
if shape not in (RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS):
|
||||
raise NotImplementedError(
|
||||
f"occurs_on for {shape} lands with its own engine task "
|
||||
"(Features 04 task sequence)"
|
||||
)
|
||||
if not _within_window(rule, target):
|
||||
return False
|
||||
if shape is RuleType.DAILY:
|
||||
offset = (target - _parse_date(rule.start_date)).days
|
||||
return offset % rule.interval == 0
|
||||
# WEEKLY / CUSTOM_DAYS: whole-week anchoring from start_date, never
|
||||
# ISO week numbers (Feature 04/05 guardrail: 53-week years flip ISO
|
||||
# parity on January 1st; anchor arithmetic does not).
|
||||
anchor = _parse_date(rule.start_date)
|
||||
if shape is RuleType.CUSTOM_DAYS:
|
||||
interval = 1
|
||||
else:
|
||||
interval = rule.interval
|
||||
week_offset = (target - anchor).days // 7
|
||||
if week_offset % interval != 0:
|
||||
return False
|
||||
return target.weekday() in rule.weekday_set
|
||||
|
||||
|
||||
def occurrences_between(
|
||||
rule: ScheduleRule, start: str, end: str
|
||||
) -> list[str]:
|
||||
"""Return every date string in [start, end] the rule fires on.
|
||||
|
||||
Inclusive on both ends. Raises RuleValidationError on malformed or
|
||||
inverted bounds; a window entirely outside [rule.start_date,
|
||||
rule.end_date] yields an empty list.
|
||||
"""
|
||||
if rule.rule_type not in (
|
||||
RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS
|
||||
):
|
||||
raise NotImplementedError(
|
||||
f"occurrences_between for {rule.rule_type} lands with its own "
|
||||
"engine task (Features 04 task sequence)"
|
||||
)
|
||||
_validate_date(start, "start")
|
||||
_validate_date(end, "end")
|
||||
start_date = _parse_date(start)
|
||||
end_date = _parse_date(end)
|
||||
if end_date < start_date:
|
||||
raise RuleValidationError(
|
||||
f"end {end!r} must be on or after start {start!r}"
|
||||
)
|
||||
rule_start = _parse_date(rule.start_date)
|
||||
if end_date < rule_start:
|
||||
return []
|
||||
# Clamp the walk to the rule's own window and the queried range.
|
||||
effective_start = max(start_date, rule_start)
|
||||
effective_end = end_date
|
||||
if rule.end_date is not None:
|
||||
effective_end = min(effective_end, _parse_date(rule.end_date))
|
||||
results: list[str] = []
|
||||
cursor = effective_start
|
||||
while cursor <= effective_end:
|
||||
if occurs_on(rule, cursor):
|
||||
results.append(cursor.isoformat())
|
||||
cursor += datetime.timedelta(days=1)
|
||||
return results
|
||||
|
||||
@@ -0,0 +1,295 @@
|
||||
"""Tests for the daily recurrence evaluation (occurs_on/occurrences_between)."""
|
||||
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 test_daily_fires_every_day_from_start() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
assert occurs_on(rule, _d("2026-09-01")) is True
|
||||
assert occurs_on(rule, _d("2026-09-02")) is True
|
||||
assert occurs_on(rule, _d("2026-09-30")) is True
|
||||
|
||||
|
||||
def test_daily_never_fires_before_start() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
assert occurs_on(rule, _d("2026-08-31")) is False
|
||||
|
||||
|
||||
def test_daily_respects_end_date() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, start_date="2026-09-01",
|
||||
end_date="2026-09-10",
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-10")) is True
|
||||
assert occurs_on(rule, _d("2026-09-11")) is False
|
||||
|
||||
|
||||
def test_daily_interval_2_fires_every_second_day() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
# Odd offsets fire (0, 2, 4...); odd offsets don't.
|
||||
assert occurs_on(rule, _d("2026-09-01")) is True # offset 0
|
||||
assert occurs_on(rule, _d("2026-09-02")) is False # offset 1
|
||||
assert occurs_on(rule, _d("2026-09-03")) is True # offset 2
|
||||
assert occurs_on(rule, _d("2026-09-04")) is False # offset 3
|
||||
assert occurs_on(rule, _d("2026-09-05")) is True # offset 4
|
||||
|
||||
|
||||
def test_daily_interval_3_fires_every_third_day() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=3, start_date="2026-09-01"
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-01")) is True # 0
|
||||
assert occurs_on(rule, _d("2026-09-02")) is False # 1
|
||||
assert occurs_on(rule, _d("2026-09-03")) is False # 2
|
||||
assert occurs_on(rule, _d("2026-09-04")) is True # 3
|
||||
assert occurs_on(rule, _d("2026-09-07")) is True # 6
|
||||
|
||||
|
||||
def test_hand_checked_30_day_sequence_interval_1() -> None:
|
||||
"""The done-condition's hand-checked sequence: 30 days, all fire."""
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
expected = [
|
||||
(2026, 9, day) for day in range(1, 31)
|
||||
]
|
||||
for (year, month, day) in expected:
|
||||
assert occurs_on(rule, datetime.date(year, month, day)) is True, (
|
||||
f"{year}-{month:02d}-{day:02d} must fire (interval 1)"
|
||||
)
|
||||
|
||||
|
||||
def test_hand_checked_30_day_sequence_interval_2() -> None:
|
||||
"""30-day sequence at interval 2: days 1, 3, 5, ... (odd days)."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
for day in range(1, 31):
|
||||
fired = occurs_on(rule, datetime.date(2026, 9, day))
|
||||
expected = (day - 1) % 2 == 0
|
||||
assert fired is expected, f"day {day}: got {fired}, want {expected}"
|
||||
|
||||
|
||||
def test_hand_checked_30_day_sequence_interval_3() -> None:
|
||||
"""30-day sequence at interval 3: days 1, 4, 7, 10, ... 28."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=3, start_date="2026-09-01"
|
||||
)
|
||||
fired_days = [
|
||||
day
|
||||
for day in range(1, 31)
|
||||
if occurs_on(rule, datetime.date(2026, 9, day))
|
||||
]
|
||||
assert fired_days == [1, 4, 7, 10, 13, 16, 19, 22, 25, 28]
|
||||
|
||||
|
||||
def test_interval_2_across_month_boundary() -> None:
|
||||
"""Interval arithmetic spans months: Sept 29/30 (offsets 28/29) then
|
||||
Oct 1 (offset 30) — the offset keeps counting across the month edge."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-29")) is True # offset 28
|
||||
assert occurs_on(rule, _d("2026-09-30")) is False # offset 29
|
||||
assert occurs_on(rule, _d("2026-10-01")) is True # offset 30
|
||||
assert occurs_on(rule, _d("2026-10-02")) is False # offset 31
|
||||
|
||||
|
||||
def test_interval_2_across_year_boundary() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-12-30"
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-12-30")) is True # 0
|
||||
assert occurs_on(rule, _d("2026-12-31")) is False # 1
|
||||
assert occurs_on(rule, _d("2027-01-01")) is True # 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# occurrences_between
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_occurrences_between_daily_interval_1() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
result = occurrences_between(rule, "2026-09-05", "2026-09-08")
|
||||
assert result == [
|
||||
"2026-09-05", "2026-09-06", "2026-09-07", "2026-09-08",
|
||||
]
|
||||
|
||||
|
||||
def test_occurrences_between_daily_interval_2() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-10")
|
||||
assert result == [
|
||||
"2026-09-01", "2026-09-03", "2026-09-05", "2026-09-07", "2026-09-09",
|
||||
]
|
||||
|
||||
|
||||
def test_occurrences_between_clamps_to_rule_start() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-05")
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-08")
|
||||
assert result == ["2026-09-05", "2026-09-06", "2026-09-07", "2026-09-08"]
|
||||
|
||||
|
||||
def test_occurrences_between_clamps_to_rule_end() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, start_date="2026-09-01",
|
||||
end_date="2026-09-06",
|
||||
)
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-20")
|
||||
assert result == [
|
||||
f"2026-09-{day:02d}" for day in range(1, 7)
|
||||
]
|
||||
|
||||
|
||||
def test_occurrences_between_disjoint_range_is_empty() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, start_date="2026-09-01",
|
||||
end_date="2026-09-15",
|
||||
)
|
||||
# The rule ends Sept 15: October is entirely outside its window.
|
||||
assert occurrences_between(rule, "2026-10-01", "2026-10-05") == []
|
||||
assert occurrences_between(rule, "2026-08-01", "2026-08-05") == []
|
||||
|
||||
|
||||
def test_occurrences_between_inclusive_bounds() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-09-01"
|
||||
)
|
||||
# Both boundary dates fire at interval 2 (offsets 0 and 6).
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-09-07")
|
||||
assert result[0] == "2026-09-01"
|
||||
assert result[-1] == "2026-09-07"
|
||||
|
||||
|
||||
def test_occurrences_between_rejects_inverted_range() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
with pytest.raises(RuleValidationError, match="on or after"):
|
||||
occurrences_between(rule, "2026-09-10", "2026-09-05")
|
||||
|
||||
|
||||
def test_occurrences_between_rejects_malformed_dates() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01")
|
||||
with pytest.raises(RuleValidationError, match="YYYY-MM-DD"):
|
||||
occurrences_between(rule, "oops", "2026-09-05")
|
||||
with pytest.raises(RuleValidationError, match="YYYY-MM-DD"):
|
||||
occurrences_between(rule, "2026-09-05", "2026-9-5")
|
||||
|
||||
|
||||
def test_occurrences_between_leap_day_fires() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-02-27")
|
||||
# 2028 is a leap year: Feb 29 exists and is 2 days after Feb 27.
|
||||
result = occurrences_between(rule, "2028-02-28", "2028-03-01")
|
||||
assert result == ["2028-02-28", "2028-02-29", "2028-03-01"]
|
||||
|
||||
|
||||
def test_occurrences_between_year_boundary() -> None:
|
||||
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-12-30")
|
||||
result = occurrences_between(rule, "2026-12-30", "2027-01-02")
|
||||
assert result == [
|
||||
"2026-12-30", "2026-12-31", "2027-01-01", "2027-01-02",
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Purity: the engine raises on unknown shapes rather than guessing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_occurs_on_rejects_unimplemented_shapes_loudly() -> None:
|
||||
"""Monthly/yearly land with their own engine tasks; a loud
|
||||
NotImplementedError beats a silent wrong answer."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
with pytest.raises(NotImplementedError):
|
||||
occurs_on(rule, _d("2026-09-15"))
|
||||
|
||||
def test_interval_7_preserves_weekday_alignment_across_months() -> None:
|
||||
"""Weekly-by-interval semantics: every firing lands on the SAME
|
||||
weekday as start_date, no matter how many month edges intervene."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=7, start_date="2026-09-01"
|
||||
)
|
||||
# 2026-09-01 is a Tuesday; every firing must be a Tuesday too.
|
||||
assert _d("2026-09-01").weekday() == 1
|
||||
for day in ("2026-09-08", "2026-09-29", "2026-10-06", "2026-11-03"):
|
||||
assert occurs_on(rule, _d(day)) is True, day
|
||||
assert _d(day).weekday() == 1, f"{day} must be a Tuesday"
|
||||
# The day after a firing is never a firing.
|
||||
assert occurs_on(rule, _d("2026-10-07")) is False
|
||||
|
||||
|
||||
def test_interval_2_across_leap_day() -> None:
|
||||
"""Interval > 1 arithmetic across a leap day: the extra day shifts
|
||||
subsequent offsets by one, and the mod arithmetic reflects it."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2028-02-27"
|
||||
)
|
||||
assert occurs_on(rule, _d("2028-02-27")) is True # 0
|
||||
assert occurs_on(rule, _d("2028-02-28")) is False # 1
|
||||
assert occurs_on(rule, _d("2028-02-29")) is True # 2 (leap day)
|
||||
assert occurs_on(rule, _d("2028-03-01")) is False # 3
|
||||
assert occurs_on(rule, _d("2028-03-02")) is True # 4
|
||||
|
||||
|
||||
def test_occurrences_between_leap_day_interval_2() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2028-02-27"
|
||||
)
|
||||
result = occurrences_between(rule, "2028-02-27", "2028-03-03")
|
||||
assert result == ["2028-02-27", "2028-02-29", "2028-03-02"]
|
||||
|
||||
|
||||
def test_occurrences_between_year_boundary_interval_2() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.DAILY, interval=2, start_date="2026-12-30"
|
||||
)
|
||||
result = occurrences_between(rule, "2026-12-30", "2027-01-04")
|
||||
# Offsets 0, 2, 4, 6 = Dec 30, Jan 1, Jan 3 ... wait: offsets are
|
||||
# Dec30=0, Dec31=1, Jan1=2, Jan2=3, Jan3=4, Jan4=5. Even offsets:
|
||||
assert result == ["2026-12-30", "2027-01-01", "2027-01-03"]
|
||||
|
||||
|
||||
def test_occurs_on_unimplemented_shape_raises_even_outside_window() -> None:
|
||||
"""An unimplemented shape must raise whether the date is inside or
|
||||
outside the window — silent False would hide a missing engine."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
|
||||
start_date="2026-09-01", end_date="2026-09-10",
|
||||
)
|
||||
# Before start, inside window, after end: all must raise.
|
||||
for day in ("2026-08-31", "2026-09-07", "2026-09-20"):
|
||||
with pytest.raises(NotImplementedError):
|
||||
occurs_on(rule, _d(day))
|
||||
|
||||
|
||||
def test_occurrences_between_unimplemented_shape_raises_even_disjoint(
|
||||
tmp_path,
|
||||
) -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.MONTHLY_DAY, day_of_month=15,
|
||||
start_date="2026-09-01"
|
||||
)
|
||||
# A range entirely before the rule's start still raises: the shape
|
||||
# is unimplemented, and silence would hide it.
|
||||
with pytest.raises(NotImplementedError):
|
||||
occurrences_between(rule, "2026-08-01", "2026-08-10")
|
||||
@@ -0,0 +1,299 @@
|
||||
"""Tests for weekly and custom day-set recurrence evaluation."""
|
||||
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)
|
||||
|
||||
|
||||
# 2026-09-01 is a TUESDAY (weekday 1).
|
||||
# Anchor-week arithmetic (start_date = Tue Sep 1): week 0 spans
|
||||
# Sep 1-7, week 1 spans Sep 8-14, week 2 spans Sep 15-21, and so on —
|
||||
# whole weeks elapsed from the anchor, NOT calendar-Sunday weeks.
|
||||
# A weekly Mon/Wed/Fri (0,2,4) rule at interval 1 anchored here fires
|
||||
# on Wed Sep 2, Fri Sep 4 (both week 0), then Mon Sep 7, Wed Sep 9,
|
||||
# Fri Sep 11 (week 1), Mon Sep 14... (week 2), and so on.
|
||||
|
||||
|
||||
def test_weekly_fires_exactly_on_set_weekdays() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
# Sep 2026: 1=Tue(1), 2=Wed(2), 4=Fri(4), 7=Mon(0), 8=Tue(1), 9=Wed,
|
||||
# 10=Thu(3), 11=Fri(4)
|
||||
assert occurs_on(rule, _d("2026-09-01")) is False # Tue NOT in set
|
||||
assert occurs_on(rule, _d("2026-09-02")) is True # Wed in set
|
||||
assert occurs_on(rule, _d("2026-09-03")) is False # Thu not in set
|
||||
assert occurs_on(rule, _d("2026-09-04")) is True # Fri in set
|
||||
assert occurs_on(rule, _d("2026-09-05")) is False # Sat
|
||||
assert occurs_on(rule, _d("2026-09-06")) is False # Sun
|
||||
assert occurs_on(rule, _d("2026-09-07")) is True # Mon
|
||||
assert occurs_on(rule, _d("2026-09-08")) is False # Tue not in set
|
||||
|
||||
|
||||
def test_weekly_start_date_itself_may_not_be_in_set() -> None:
|
||||
"""start_date anchors the WEEK count, not a firing day: a Tuesday
|
||||
start for a Mon/Wed/Fri rule fires on Wed Sep 2, not Sep 1."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-01")) is False # Tue=1, not in set
|
||||
assert occurs_on(rule, _d("2026-09-02")) is True # Wed=2
|
||||
assert occurs_on(rule, _d("2026-09-04")) is True # Fri=4
|
||||
assert occurs_on(rule, _d("2026-09-07")) is True # Mon, next week
|
||||
|
||||
|
||||
def test_weekly_never_fires_before_start() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={2}, # Wednesdays
|
||||
start_date="2026-09-02", # a Wednesday
|
||||
)
|
||||
# Wed Sep 2 is the anchor day and in-set: fires. Wed Aug 26 (one
|
||||
# week BEFORE the anchor, in-set) must not fire: pre-window.
|
||||
assert occurs_on(rule, _d("2026-09-02")) is True
|
||||
assert occurs_on(rule, _d("2026-08-26")) is False # Wed, before start
|
||||
assert occurs_on(rule, _d("2026-09-09")) is True # Wed, next week
|
||||
|
||||
|
||||
def test_weekly_respects_end_date() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
|
||||
start_date="2026-09-01", end_date="2026-09-10",
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-09")) is True # Wed within end
|
||||
assert occurs_on(rule, _d("2026-09-11")) is False # Fri after end
|
||||
|
||||
|
||||
def test_weekly_interval_2_fires_on_alternate_weeks() -> None:
|
||||
"""Interval 2: fires on set weekdays only in even week-offsets."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0}, interval=2,
|
||||
start_date="2026-09-07", # a Monday
|
||||
)
|
||||
# Week 0 (Sep 7-13): fires Mon Sep 7.
|
||||
assert occurs_on(rule, _d("2026-09-07")) is True
|
||||
# Week 1 (Sep 14-20): skipped.
|
||||
assert occurs_on(rule, _d("2026-09-14")) is False
|
||||
# Week 2 (Sep 21-27): fires Mon Sep 21.
|
||||
assert occurs_on(rule, _d("2026-09-21")) is True
|
||||
# Week 3 (Sep 28-Oct 4): skipped; Mon Sep 28 does not fire.
|
||||
assert occurs_on(rule, _d("2026-09-28")) is False
|
||||
# Week 4 (Oct 5-11): fires Mon Oct 5.
|
||||
assert occurs_on(rule, _d("2026-10-05")) is True
|
||||
|
||||
|
||||
def test_weekly_interval_3_every_third_week() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={3}, interval=3,
|
||||
start_date="2026-09-03", # a Thursday
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-03")) is True # week 0
|
||||
assert occurs_on(rule, _d("2026-09-10")) is False # week 1
|
||||
assert occurs_on(rule, _d("2026-09-17")) is False # week 2
|
||||
assert occurs_on(rule, _d("2026-09-24")) is True # week 3
|
||||
assert occurs_on(rule, _d("2026-10-15")) is True # week 6
|
||||
|
||||
|
||||
def test_weekly_interval_2_never_fires_in_odd_weeks_across_months() -> None:
|
||||
"""The week offset is anchored to start_date across month edges."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={2}, interval=2,
|
||||
start_date="2026-09-02", # a Wednesday
|
||||
)
|
||||
# Week 0: Wed Sep 2 fires. Week 1: Wed Sep 9 skipped.
|
||||
# Week 2: Wed Sep 16 fires. Week 3: Wed Sep 23 skipped.
|
||||
# Week 4 (Oct): Wed Sep 30 fires (still week 4!). Week 5: Wed Oct 7 skipped.
|
||||
assert occurs_on(rule, _d("2026-09-02")) is True
|
||||
assert occurs_on(rule, _d("2026-09-09")) is False
|
||||
assert occurs_on(rule, _d("2026-09-16")) is True
|
||||
assert occurs_on(rule, _d("2026-09-23")) is False
|
||||
assert occurs_on(rule, _d("2026-09-30")) is True
|
||||
assert occurs_on(rule, _d("2026-10-07")) is False
|
||||
assert occurs_on(rule, _d("2026-10-14")) is True
|
||||
|
||||
|
||||
def test_custom_days_interval_1_behaves_like_weekly() -> None:
|
||||
"""CUSTOM_DAYS with interval 1 behaves identically to WEEKLY."""
|
||||
weekly = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={1, 3, 5},
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
custom = ScheduleRule(
|
||||
rule_type=RuleType.CUSTOM_DAYS, weekday_set={1, 3, 5},
|
||||
start_date="2026-09-01",
|
||||
)
|
||||
# Whole month of September must match exactly.
|
||||
for day in range(1, 31):
|
||||
date = datetime.date(2026, 9, day)
|
||||
assert occurs_on(weekly, date) == occurs_on(custom, date), date
|
||||
|
||||
|
||||
def test_custom_days_never_fires_before_start(tmp_path) -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.CUSTOM_DAYS, weekday_set={0},
|
||||
start_date="2026-09-07", # a Monday
|
||||
)
|
||||
assert occurs_on(rule, _d("2026-09-06")) is False # Sun, before
|
||||
assert occurs_on(rule, _d("2026-09-07")) is True # Mon, on
|
||||
|
||||
|
||||
def test_hand_checked_8_week_sequence_weekly_mon_wed_fri() -> None:
|
||||
"""The done-condition's hand-checked sequence: 8 weeks of a
|
||||
Mon/Wed/Fri rule anchored at Tue 2026-09-01, interval 1.
|
||||
|
||||
Anchor weeks from Tue Sep 1 (week 0 = Sep 1-7, week 1 = Sep 8-14,
|
||||
... week 7 = Oct 20-26). Mon/Wed/Fri firings (interval 1, so every
|
||||
in-set weekday fires regardless of week):
|
||||
Week 0 (Sep 1-7): Wed 2, Fri 4 (Mon Sep 7 is ALSO week 0:
|
||||
6 elapsed days // 7 = 0 — the anchor week
|
||||
boundary sits on Sep 8, not on the Monday)
|
||||
Week 1 (Sep 8-14): Wed 9, Fri 11, Mon 14 (13 elapsed days // 7 = 1)
|
||||
Week 2 (Sep 15-21): Wed 16, Fri 18, Mon 21
|
||||
Week 3 (Sep 22-28): Wed 23, Fri 25, Mon 28
|
||||
Week 4 (Sep 29-Oct 5): Wed 30, Fri 2 (Oct), Mon 5 (Oct)
|
||||
Week 5 (Oct 6-12): Wed 7, Fri 9, Mon 12
|
||||
Week 6 (Oct 13-19): Wed 14, Fri 16, Mon 19
|
||||
Week 7 (Oct 20-26): Wed 21, Fri 23, Mon 26
|
||||
"""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0, 2, 4},
|
||||
start_date="2026-09-01", end_date="2026-10-26",
|
||||
)
|
||||
expected = [
|
||||
"2026-09-02", "2026-09-04",
|
||||
"2026-09-07", "2026-09-09", "2026-09-11",
|
||||
"2026-09-14", "2026-09-16", "2026-09-18",
|
||||
"2026-09-21", "2026-09-23", "2026-09-25",
|
||||
"2026-09-28", "2026-09-30", "2026-10-02",
|
||||
"2026-10-05", "2026-10-07", "2026-10-09",
|
||||
"2026-10-12", "2026-10-14", "2026-10-16",
|
||||
"2026-10-19", "2026-10-21", "2026-10-23", "2026-10-26",
|
||||
]
|
||||
actual = occurrences_between(rule, "2026-09-01", "2026-10-26")
|
||||
assert actual == expected, (
|
||||
f"diff: {set(expected) ^ set(actual)}"
|
||||
)
|
||||
|
||||
|
||||
def test_hand_checked_8_week_sequence_weekly_interval_2_mondays() -> None:
|
||||
"""8 weeks of a Monday rule at interval 2, anchored Mon 2026-09-07
|
||||
(week 0 = Sep 7-13): fires in weeks 0, 2, 4, 6 -> Mon Sep 7, Mon Sep 21,
|
||||
Mon Oct 5, Mon Oct 19."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0}, interval=2,
|
||||
start_date="2026-09-07", end_date="2026-10-31",
|
||||
)
|
||||
actual = occurrences_between(rule, "2026-09-07", "2026-10-31")
|
||||
assert actual == ["2026-09-07", "2026-09-21", "2026-10-05", "2026-10-19"]
|
||||
|
||||
|
||||
def test_custom_days_hand_checked_8_week_sequence() -> None:
|
||||
"""8 weeks of a Tue/Thu/Sun rule (1,3,6) anchored Mon 2026-09-07:
|
||||
fires Tue 8, Thu 10, Sun 13, then weekly."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.CUSTOM_DAYS, weekday_set={1, 3, 6},
|
||||
start_date="2026-09-07",
|
||||
)
|
||||
# Anchor weeks from Mon Sep 7 (w0 = Sep 7-13 ... w7 = Oct 26-Nov 1):
|
||||
# Tue/Thu/Sun firings per week, 8 weeks fully covered.
|
||||
actual = occurrences_between(rule, "2026-09-07", "2026-11-01")
|
||||
assert actual == [
|
||||
"2026-09-08", "2026-09-10", "2026-09-13",
|
||||
"2026-09-15", "2026-09-17", "2026-09-20",
|
||||
"2026-09-22", "2026-09-24", "2026-09-27",
|
||||
"2026-09-29", "2026-10-01", "2026-10-04",
|
||||
"2026-10-06", "2026-10-08", "2026-10-11",
|
||||
"2026-10-13", "2026-10-15", "2026-10-18",
|
||||
"2026-10-20", "2026-10-22", "2026-10-25",
|
||||
"2026-10-27", "2026-10-29", "2026-11-01",
|
||||
]
|
||||
|
||||
|
||||
def test_weekly_survives_53_week_year_boundary() -> None:
|
||||
"""Anchor arithmetic never flips at Jan 1 (the ISO parity trap):
|
||||
a weekly rule spanning New Year keeps its weekday pattern."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={3}, # Thursdays
|
||||
start_date="2026-12-24", # a Thursday
|
||||
)
|
||||
# Thursdays around New Year 2026->2027: Dec 24, 31, Jan 7, 14.
|
||||
assert occurs_on(rule, _d("2026-12-24")) is True
|
||||
assert occurs_on(rule, _d("2026-12-31")) is True
|
||||
assert occurs_on(rule, _d("2027-01-07")) is True
|
||||
assert occurs_on(rule, _d("2027-01-14")) is True
|
||||
# Interleaved days are not Thursdays.
|
||||
assert occurs_on(rule, _d("2027-01-04")) is False # Monday
|
||||
assert occurs_on(rule, _d("2027-01-05")) is False # Tuesday
|
||||
|
||||
|
||||
def test_weekly_interval_2_across_year_boundary() -> None:
|
||||
"""Interval weeks continue across Jan 1 with no parity flip."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={4}, interval=2,
|
||||
start_date="2026-12-25", # a Friday
|
||||
)
|
||||
# Weeks: Dec 25 (w0) fires, Jan 1 (w1) skipped, Jan 8 (w2) fires,
|
||||
# Jan 15 (w3) skipped, Jan 22 (w4) fires.
|
||||
assert occurs_on(rule, _d("2026-12-25")) is True
|
||||
assert occurs_on(rule, _d("2027-01-01")) is False
|
||||
assert occurs_on(rule, _d("2027-01-08")) is True
|
||||
assert occurs_on(rule, _d("2027-01-15")) is False
|
||||
assert occurs_on(rule, _d("2027-01-22")) is True
|
||||
|
||||
|
||||
def test_occurrences_between_weekly_disjoint(tmp_path) -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={1},
|
||||
start_date="2026-09-01", end_date="2026-09-30",
|
||||
)
|
||||
assert occurrences_between(rule, "2026-10-01", "2026-10-31") == []
|
||||
|
||||
|
||||
def test_occurrences_between_weekly_interval_2_span() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0}, interval=2,
|
||||
start_date="2026-09-07", end_date="2026-10-31",
|
||||
)
|
||||
result = occurrences_between(rule, "2026-09-01", "2026-10-31")
|
||||
assert result == ["2026-09-07", "2026-09-21", "2026-10-05", "2026-10-19"]
|
||||
|
||||
|
||||
def test_occurrences_between_rejects_inverted_bounds() -> None:
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0}, start_date="2026-09-01"
|
||||
)
|
||||
with pytest.raises(RuleValidationError, match="on or after"):
|
||||
occurrences_between(rule, "2026-10-10", "2026-09-01")
|
||||
|
||||
def test_weekly_interval_2_non_monday_anchor_boundary() -> None:
|
||||
"""The anchor week is anchored at start_date, NOT a calendar week:
|
||||
a Tuesday-anchored Monday rule fires on Mon Sep 7 (week 0 extends
|
||||
through Sep 7), skips Mon Sep 14 (week 1), fires Mon Sep 21 (week 2)."""
|
||||
rule = ScheduleRule(
|
||||
rule_type=RuleType.WEEKLY, weekday_set={0}, interval=2,
|
||||
start_date="2026-09-01", # a TUESDAY anchor
|
||||
)
|
||||
# Sep 7 - Sep 1 = 6 days -> week 0 (even): fires.
|
||||
assert occurs_on(rule, _d("2026-09-07")) is True
|
||||
# Sep 14: 13 days -> week 1 (odd): skipped.
|
||||
assert occurs_on(rule, _d("2026-09-14")) is False
|
||||
# Sep 21: 20 days -> week 2 (even): fires.
|
||||
assert occurs_on(rule, _d("2026-09-21")) is True
|
||||
# Sep 28: 27 days -> week 3 (odd): skipped.
|
||||
assert occurs_on(rule, _d("2026-09-28")) is False
|
||||
# Oct 5: 34 days -> week 4 (even): fires.
|
||||
assert occurs_on(rule, _d("2026-10-05")) is True
|
||||
Reference in New Issue
Block a user