242 lines
8.8 KiB
Python
242 lines
8.8 KiB
Python
import sys
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
SCRIPTS = Path(__file__).parent.parent / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
from db import get_db, init_db
|
|
import remind_send
|
|
|
|
|
|
def _run_send(db_path, now=None):
|
|
"""Run remind_send.main with a temporary DB, a stubbed config, and an optional fixed now."""
|
|
original_db_path = remind_send.DB_PATH
|
|
original_now = remind_send._now_prague
|
|
original_config = remind_send._telegram_config
|
|
try:
|
|
remind_send.DB_PATH = db_path
|
|
remind_send._telegram_config = lambda: ("token", "chat")
|
|
if now is not None:
|
|
remind_send._now_prague = lambda: now
|
|
remind_send.main()
|
|
finally:
|
|
remind_send.DB_PATH = original_db_path
|
|
remind_send._now_prague = original_now
|
|
remind_send._telegram_config = original_config
|
|
|
|
|
|
def _add_reminder(conn, text, enabled=1, deleted_at=None):
|
|
conn.execute(
|
|
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at, deleted_at) "
|
|
"VALUES (?, ?, 'Europe/Prague', 'now', 'now', ?)",
|
|
(text, enabled, deleted_at),
|
|
)
|
|
return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
|
|
|
|
def test_due_at_delivers_once(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "at reminder")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T10:00:00"))
|
|
finally:
|
|
conn.close()
|
|
|
|
now = datetime(2026, 6, 10, 10, 0, 0)
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_called_once_with("⏰ Reminder: at reminder", "token", "chat")
|
|
|
|
# second run — dedup
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_not_called()
|
|
|
|
|
|
def test_due_cron_delivers_once(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "cron reminder")
|
|
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid, "0 10 * * *"))
|
|
finally:
|
|
conn.close()
|
|
|
|
now = datetime(2026, 6, 10, 10, 0, 0)
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_called_once_with("⏰ Reminder: cron reminder", "token", "chat")
|
|
|
|
# second run — dedup
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_not_called()
|
|
|
|
|
|
def test_due_random_delivers_once(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "random reminder")
|
|
conn.execute(
|
|
"INSERT INTO schedule_random (reminder_id, times_per_day, window_start, window_end) VALUES (?, ?, ?, ?)",
|
|
(rid, 2, 540, 1260),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
from random_times import compute_fire_times
|
|
fires = compute_fire_times(date(2026, 6, 10), "random reminder", {"times_per_day": 2, "window": "09:00-21:00"})
|
|
assert len(fires) == 2
|
|
|
|
for ft in fires:
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, ft)
|
|
mock_send.assert_called_once_with("⏰ Reminder: random reminder", "token", "chat")
|
|
|
|
# all deduped now
|
|
for ft in fires:
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, ft)
|
|
mock_send.assert_not_called()
|
|
|
|
|
|
def test_due_random_with_days_filter_delivers_on_allowed_day(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "weekday only")
|
|
conn.execute(
|
|
"INSERT INTO schedule_random (reminder_id, times_per_day, window_start, window_end, days_filter) "
|
|
"VALUES (?, 1, 540, 1260, '1-5')",
|
|
(rid,),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
from random_times import compute_fire_times
|
|
wednesday = date(2026, 6, 10)
|
|
fires = compute_fire_times(wednesday, "weekday only", {"times_per_day": 1, "window": "09:00-21:00", "days": "1-5"})
|
|
assert len(fires) == 1 # 2026-06-10 is a Wednesday, allowed by 1-5
|
|
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, fires[0])
|
|
mock_send.assert_called_once_with("⏰ Reminder: weekday only", "token", "chat")
|
|
|
|
|
|
def test_disabled_not_sent(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "disabled reminder", enabled=0)
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T10:00:00"))
|
|
finally:
|
|
conn.close()
|
|
|
|
now = datetime(2026, 6, 10, 10, 0, 0)
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_not_called()
|
|
|
|
|
|
def test_deleted_not_sent(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "deleted reminder", deleted_at="now")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T10:00:00"))
|
|
finally:
|
|
conn.close()
|
|
|
|
now = datetime(2026, 6, 10, 10, 0, 0)
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_not_called()
|
|
|
|
|
|
def test_delivery_failure_logged(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "fail reminder")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T10:00:00"))
|
|
finally:
|
|
conn.close()
|
|
|
|
now = datetime(2026, 6, 10, 10, 0, 0)
|
|
with patch.object(remind_send, "_send_telegram", side_effect=RuntimeError("network down")) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_called_once()
|
|
|
|
conn = get_db(db_path)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT status, error_message FROM reminder_fires WHERE reminder_id = ?", (rid,)
|
|
).fetchone()
|
|
assert row["status"] == "failed"
|
|
assert "network down" in row["error_message"]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_failed_fire_retries(tmp_path):
|
|
"""A fire that failed (status='failed') is not deduped — the next run retries it."""
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid = _add_reminder(conn, "retry me")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T10:00:00"))
|
|
finally:
|
|
conn.close()
|
|
|
|
now = datetime(2026, 6, 10, 10, 0, 0)
|
|
with patch.object(remind_send, "_send_telegram", side_effect=RuntimeError("down")):
|
|
_run_send(db_path, now)
|
|
|
|
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
|
|
_run_send(db_path, now)
|
|
mock_send.assert_called_once_with("⏰ Reminder: retry me", "token", "chat")
|
|
|
|
|
|
def test_schedule_type_correct_despite_id_collision(tmp_path):
|
|
"""A cron fire must record schedule_type='cron' even when schedule_cron.id collides
|
|
with a schedule_at.id (each schedule table has its own AUTOINCREMENT sequence)."""
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
conn = get_db(db_path)
|
|
try:
|
|
rid_at = _add_reminder(conn, "at one")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid_at, "2030-01-01T00:00:00"))
|
|
rid_cron = _add_reminder(conn, "cron one")
|
|
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid_cron, "0 10 * * *"))
|
|
# schedule_at.id and schedule_cron.id both equal 1 here — the collision the fix guards against.
|
|
assert conn.execute("SELECT id FROM schedule_at").fetchone()["id"] == 1
|
|
assert conn.execute("SELECT id FROM schedule_cron").fetchone()["id"] == 1
|
|
finally:
|
|
conn.close()
|
|
|
|
now = datetime(2026, 6, 10, 10, 0, 0)
|
|
with patch.object(remind_send, "_send_telegram", return_value=None):
|
|
_run_send(db_path, now)
|
|
|
|
conn = get_db(db_path)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT schedule_type, status FROM reminder_fires WHERE reminder_id = ?", (rid_cron,)
|
|
).fetchone()
|
|
assert row["schedule_type"] == "cron"
|
|
assert row["status"] == "delivered"
|
|
finally:
|
|
conn.close()
|