125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
"""Тесты расписаний бота: совпадение времени, таймзоны и сид V16."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
|
||
from bot.services import notifications
|
||
from bot.services.notifications import DEFAULT_SCHEDULES, ScheduledMessage, _from_row
|
||
|
||
SEED_FILE = Path(__file__).resolve().parents[1] / "sql" / "V16__seed_schedules.sql"
|
||
_ROW_RE = re.compile(r"\('([^']+)',\s*(\d+),\s*(\d+),\s*'\{([^}]*)\}',\s*'([^']+)'")
|
||
|
||
|
||
def test_matches_daily_in_schedule_timezone() -> None:
|
||
schedule = ScheduledMessage(16, 30, "notify.blago_30")
|
||
|
||
# 13:30 UTC = 16:30 МСК
|
||
assert schedule.matches(datetime(2026, 10, 19, 13, 30, tzinfo=UTC))
|
||
assert not schedule.matches(datetime(2026, 10, 19, 13, 31, tzinfo=UTC))
|
||
|
||
|
||
def test_matches_respects_weekdays_iso() -> None:
|
||
friday = ScheduledMessage(15, 0, "notify.vklad", weekdays=(5,))
|
||
|
||
# 2026-10-16 — пятница, 12:00 UTC = 15:00 МСК
|
||
assert friday.matches(datetime(2026, 10, 16, 12, 0, tzinfo=UTC))
|
||
# 2026-10-15 — четверг
|
||
assert not friday.matches(datetime(2026, 10, 15, 12, 0, tzinfo=UTC))
|
||
|
||
|
||
def test_matches_uses_schedule_timezone() -> None:
|
||
berlin = ScheduledMessage(16, 30, "notify.blago_30", timezone="Europe/Berlin")
|
||
|
||
# Зимой Берлин = UTC+1: 15:30 UTC = 16:30
|
||
assert berlin.matches(datetime(2026, 1, 15, 15, 30, tzinfo=UTC))
|
||
|
||
|
||
def test_from_row_converts_fields() -> None:
|
||
schedule = _from_row(
|
||
{
|
||
"hour": 14,
|
||
"minute": 5,
|
||
"template_key": "notify.vklad",
|
||
"weekdays": [7],
|
||
"timezone": "Europe/Moscow",
|
||
}
|
||
)
|
||
|
||
assert schedule == ScheduledMessage(14, 5, "notify.vklad", weekdays=(7,), timezone="Europe/Moscow")
|
||
|
||
|
||
def test_default_schedules_match_seed() -> None:
|
||
"""Расписание-фолбэк в коде не должно расходиться с сидом V16."""
|
||
seeded = []
|
||
for key, hour, minute, days, timezone in _ROW_RE.findall(SEED_FILE.read_text(encoding="utf-8")):
|
||
weekdays = tuple(int(day) for day in days.split(",") if day.strip())
|
||
seeded.append(
|
||
ScheduledMessage(
|
||
hour=int(hour),
|
||
minute=int(minute),
|
||
template_key=key,
|
||
weekdays=weekdays,
|
||
timezone=timezone,
|
||
)
|
||
)
|
||
|
||
assert seeded == list(DEFAULT_SCHEDULES)
|
||
|
||
|
||
async def test_tick_skips_already_sent_minute(monkeypatch) -> None:
|
||
schedule = ScheduledMessage(12, 0, "notify.blago", schedule_id=1)
|
||
sent: list[str] = []
|
||
|
||
async def load() -> tuple[ScheduledMessage, ...]:
|
||
return (schedule,)
|
||
|
||
async def claim(schedule_id: int, run_at: datetime) -> bool:
|
||
return False
|
||
|
||
async def subscribers() -> list[dict]:
|
||
return []
|
||
|
||
async def broadcast(bot, key: str, rows: list[dict]) -> None:
|
||
sent.append(key)
|
||
|
||
monkeypatch.setattr(notifications, "_load_schedules", load)
|
||
monkeypatch.setattr(notifications.schedules_repo, "claim_run", claim)
|
||
monkeypatch.setattr(notifications.users_service, "list_subscribers", subscribers)
|
||
monkeypatch.setattr(notifications, "_send_broadcast", broadcast)
|
||
|
||
await notifications._tick(None, datetime(2026, 1, 1, 9, 0, tzinfo=UTC))
|
||
|
||
assert sent == []
|
||
|
||
|
||
async def test_tick_sends_once_when_claimed(monkeypatch) -> None:
|
||
schedule = ScheduledMessage(12, 0, "notify.blago", schedule_id=1)
|
||
sent: list[str] = []
|
||
claims: list[tuple[int, datetime]] = []
|
||
|
||
async def load() -> tuple[ScheduledMessage, ...]:
|
||
return (schedule,)
|
||
|
||
async def claim(schedule_id: int, run_at: datetime) -> bool:
|
||
claims.append((schedule_id, run_at))
|
||
return True
|
||
|
||
async def subscribers() -> list[dict]:
|
||
return []
|
||
|
||
async def broadcast(bot, key: str, rows: list[dict]) -> None:
|
||
sent.append(key)
|
||
|
||
monkeypatch.setattr(notifications, "_load_schedules", load)
|
||
monkeypatch.setattr(notifications.schedules_repo, "claim_run", claim)
|
||
monkeypatch.setattr(notifications.users_service, "list_subscribers", subscribers)
|
||
monkeypatch.setattr(notifications, "_send_broadcast", broadcast)
|
||
|
||
await notifications._tick(None, datetime(2026, 1, 1, 9, 0, tzinfo=UTC))
|
||
|
||
assert sent == ["notify.blago"]
|
||
assert claims == [(1, datetime(2026, 1, 1, 9, 0, tzinfo=UTC))]
|