297 lines
12 KiB
Python
297 lines
12 KiB
Python
"""Тесты recurrence-ядра (ADR-001, п.3): `RecurrenceRule` и `expand_occurrences`.
|
||
|
||
Пишутся ДО реализации `services/recurrence.py` (TDD). Покрывают все 4 типа
|
||
повторения, клэмп 31-го числа, переход года, пустой диапазон, включительность
|
||
границ диапазона и корректность конвертации локального времени в UTC.
|
||
"""
|
||
|
||
from datetime import UTC, datetime, timedelta
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from services.recurrence import RecurrenceRule, expand_occurrences
|
||
|
||
|
||
def _rule(**overrides: object) -> RecurrenceRule:
|
||
"""Собрать валидное правило с разумными умолчаниями, переопределяя нужные поля."""
|
||
defaults: dict[str, object] = {
|
||
"type": "weekly",
|
||
"weekdays": [0],
|
||
"anchor_date": datetime(2026, 1, 5).date(), # понедельник
|
||
"time_local": "10:00",
|
||
"timezone": "UTC",
|
||
"duration_minutes": 60,
|
||
}
|
||
defaults.update(overrides)
|
||
return RecurrenceRule.model_validate(defaults)
|
||
|
||
|
||
class TestRecurrenceRuleValidation:
|
||
"""Валидация Pydantic-модели правила."""
|
||
|
||
def test_weekly_requires_weekdays(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(type="weekly", weekdays=[])
|
||
|
||
def test_monthly_requires_day_of_month(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(type="monthly", day_of_month=None)
|
||
|
||
def test_every_n_days_requires_interval_days(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(type="every_n_days", interval_days=None)
|
||
|
||
def test_weekday_out_of_range_rejected(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(weekdays=[7])
|
||
|
||
def test_day_of_month_out_of_range_rejected(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(type="monthly", day_of_month=32)
|
||
|
||
def test_interval_days_must_be_positive(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(type="every_n_days", interval_days=0)
|
||
|
||
def test_invalid_time_format_rejected(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(time_local="25:99")
|
||
|
||
def test_invalid_timezone_rejected(self) -> None:
|
||
with pytest.raises(ValidationError):
|
||
_rule(timezone="Not/AZone")
|
||
|
||
|
||
class TestExpandOccurrencesWeekly:
|
||
"""type='weekly': вхождение на каждой указанной неделе для всех weekdays."""
|
||
|
||
def test_multiple_weekdays_in_one_week(self) -> None:
|
||
# Пн(0)/Ср(2)/Пт(4), 10:00 UTC, неделя с 2026-01-05 (пн) по 2026-01-11 (вс)
|
||
rule = _rule(type="weekly", weekdays=[0, 2, 4])
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 5, tzinfo=UTC),
|
||
datetime(2026, 1, 11, 23, 59, tzinfo=UTC),
|
||
)
|
||
assert occurrences == [
|
||
datetime(2026, 1, 5, 10, 0, tzinfo=UTC),
|
||
datetime(2026, 1, 7, 10, 0, tzinfo=UTC),
|
||
datetime(2026, 1, 9, 10, 0, tzinfo=UTC),
|
||
]
|
||
|
||
def test_repeats_every_week(self) -> None:
|
||
rule = _rule(type="weekly", weekdays=[0])
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 1, 31, tzinfo=UTC),
|
||
)
|
||
# Понедельники января 2026: 5, 12, 19, 26
|
||
assert [d.day for d in occurrences] == [5, 12, 19, 26]
|
||
assert all(d.hour == 10 for d in occurrences)
|
||
|
||
|
||
class TestExpandOccurrencesBiweekly:
|
||
"""type='biweekly': чётность недель считается от `anchor_date`."""
|
||
|
||
def test_every_other_week_from_anchor(self) -> None:
|
||
# anchor 2026-01-05 (пн) — неделя-0 включена, неделя-1 (12-е) исключена,
|
||
# неделя-2 (19-е) включена и т.д.
|
||
rule = _rule(type="biweekly", weekdays=[0], anchor_date=datetime(2026, 1, 5).date())
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 2, 28, tzinfo=UTC),
|
||
)
|
||
assert [d.day for d in occurrences[:3]] == [5, 19, 2] # 2 февраля
|
||
|
||
def test_anchor_not_matching_weekday_still_defines_parity(self) -> None:
|
||
# anchor в среду той же недели, что и понедельник 2026-01-05 —
|
||
# чётность определяется по началу недели anchor, а не по его weekday.
|
||
rule = _rule(type="biweekly", weekdays=[0], anchor_date=datetime(2026, 1, 7).date())
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 1, 31, tzinfo=UTC),
|
||
)
|
||
assert [d.day for d in occurrences] == [5, 19]
|
||
|
||
|
||
class TestExpandOccurrencesMonthly:
|
||
"""type='monthly': day_of_month=31 клэмпится к последнему дню короткого месяца."""
|
||
|
||
def test_31_clamped_in_short_months(self) -> None:
|
||
rule = _rule(type="monthly", weekdays=[], day_of_month=31)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 4, 30, 23, 59, tzinfo=UTC),
|
||
)
|
||
# 2026 — невисокосный год: январь(31), февраль(28), март(31), апрель(30)
|
||
assert [(d.month, d.day) for d in occurrences] == [
|
||
(1, 31),
|
||
(2, 28),
|
||
(3, 31),
|
||
(4, 30),
|
||
]
|
||
|
||
def test_31_clamped_to_29_in_leap_february(self) -> None:
|
||
rule = _rule(type="monthly", weekdays=[], day_of_month=31)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2024, 2, 1, tzinfo=UTC),
|
||
datetime(2024, 2, 29, 23, 59, tzinfo=UTC),
|
||
)
|
||
assert [(d.month, d.day) for d in occurrences] == [(2, 29)]
|
||
|
||
def test_regular_day_of_month(self) -> None:
|
||
rule = _rule(type="monthly", weekdays=[], day_of_month=15)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 3, 31, tzinfo=UTC),
|
||
)
|
||
assert [d.day for d in occurrences] == [15, 15, 15]
|
||
|
||
|
||
class TestExpandOccurrencesEveryNDays:
|
||
"""type='every_n_days': шаг фиксированного числа дней от `anchor_date`."""
|
||
|
||
def test_step_of_five_days(self) -> None:
|
||
rule = _rule(
|
||
type="every_n_days",
|
||
weekdays=[],
|
||
interval_days=5,
|
||
anchor_date=datetime(2026, 1, 1).date(),
|
||
)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 1, 20, tzinfo=UTC),
|
||
)
|
||
assert [d.day for d in occurrences] == [1, 6, 11, 16]
|
||
|
||
def test_before_anchor_no_occurrences(self) -> None:
|
||
rule = _rule(
|
||
type="every_n_days",
|
||
weekdays=[],
|
||
interval_days=3,
|
||
anchor_date=datetime(2026, 1, 10).date(),
|
||
)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 1, 9, 23, 59, tzinfo=UTC),
|
||
)
|
||
assert occurrences == []
|
||
|
||
|
||
class TestYearTransition:
|
||
"""Развёртка не должна ломаться на переходе декабрь -> январь."""
|
||
|
||
def test_weekly_across_year_boundary(self) -> None:
|
||
rule = _rule(type="weekly", weekdays=[3]) # четверг
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2025, 12, 25, tzinfo=UTC),
|
||
datetime(2026, 1, 8, 23, 59, tzinfo=UTC),
|
||
)
|
||
assert [(d.year, d.month, d.day) for d in occurrences] == [
|
||
(2025, 12, 25),
|
||
(2026, 1, 1),
|
||
(2026, 1, 8),
|
||
]
|
||
|
||
def test_every_n_days_across_year_boundary(self) -> None:
|
||
rule = _rule(
|
||
type="every_n_days",
|
||
weekdays=[],
|
||
interval_days=10,
|
||
anchor_date=datetime(2025, 12, 20).date(),
|
||
)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2025, 12, 20, tzinfo=UTC),
|
||
datetime(2026, 1, 10, tzinfo=UTC),
|
||
)
|
||
assert [(d.year, d.month, d.day) for d in occurrences] == [
|
||
(2025, 12, 20),
|
||
(2025, 12, 30),
|
||
(2026, 1, 9),
|
||
]
|
||
|
||
|
||
class TestRangeEdgeCases:
|
||
"""Пустой диапазон и включительность границ."""
|
||
|
||
def test_empty_range_when_from_after_to(self) -> None:
|
||
rule = _rule(type="weekly", weekdays=[0, 1, 2, 3, 4, 5, 6])
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 10, tzinfo=UTC),
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
)
|
||
assert occurrences == []
|
||
|
||
def test_boundaries_are_inclusive(self) -> None:
|
||
# Вхождение ровно в момент t_from и ровно в момент t_to должно попасть
|
||
# в результат (обе границы включительны).
|
||
rule = _rule(type="weekly", weekdays=[0], time_local="10:00", timezone="UTC")
|
||
t_from = datetime(2026, 1, 5, 10, 0, tzinfo=UTC)
|
||
t_to = datetime(2026, 1, 12, 10, 0, tzinfo=UTC)
|
||
occurrences = expand_occurrences(rule, t_from, t_to)
|
||
assert occurrences[0] == t_from
|
||
assert occurrences[-1] == t_to
|
||
|
||
def test_one_minute_outside_boundaries_excluded(self) -> None:
|
||
rule = _rule(type="weekly", weekdays=[0], time_local="10:00", timezone="UTC")
|
||
t_from = datetime(2026, 1, 5, 10, 1, tzinfo=UTC)
|
||
t_to = datetime(2026, 1, 12, 9, 59, tzinfo=UTC)
|
||
occurrences = expand_occurrences(rule, t_from, t_to)
|
||
assert occurrences == []
|
||
|
||
|
||
class TestTimezoneOffset:
|
||
"""Правило хранит локальное время + IANA tz; развёртка возвращает aware UTC."""
|
||
|
||
def test_local_time_converted_to_utc(self) -> None:
|
||
# Europe/Moscow — постоянное смещение UTC+3 без перехода на летнее время.
|
||
rule = _rule(
|
||
type="weekly",
|
||
weekdays=[0],
|
||
time_local="10:00",
|
||
timezone="Europe/Moscow",
|
||
)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 5, tzinfo=UTC),
|
||
datetime(2026, 1, 5, 23, 59, tzinfo=UTC),
|
||
)
|
||
assert occurrences == [datetime(2026, 1, 5, 7, 0, tzinfo=UTC)]
|
||
|
||
def test_negative_offset_timezone(self) -> None:
|
||
# America/New_York в начале января — зимнее время, UTC-5.
|
||
rule = _rule(
|
||
type="weekly",
|
||
weekdays=[0],
|
||
time_local="09:00",
|
||
timezone="America/New_York",
|
||
)
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 5, tzinfo=UTC),
|
||
datetime(2026, 1, 5, 23, 59, tzinfo=UTC),
|
||
)
|
||
assert occurrences == [datetime(2026, 1, 5, 14, 0, tzinfo=UTC)]
|
||
|
||
def test_result_timestamps_are_utc_aware(self) -> None:
|
||
rule = _rule(type="weekly", weekdays=[0], timezone="Europe/Moscow")
|
||
occurrences = expand_occurrences(
|
||
rule,
|
||
datetime(2026, 1, 1, tzinfo=UTC),
|
||
datetime(2026, 1, 31, tzinfo=UTC),
|
||
)
|
||
assert occurrences
|
||
assert all(d.tzinfo is not None and d.utcoffset() == timedelta(0) for d in occurrences)
|