180 lines
6.5 KiB
Python
180 lines
6.5 KiB
Python
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
SCRIPTS = Path(__file__).parent.parent / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
from db import get_db, init_db
|
|
from forecast import fires_in_window, format_upcoming, window_for
|
|
|
|
|
|
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]
|
|
|
|
|
|
@pytest.fixture
|
|
def conn(tmp_path):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
connection = get_db(db_path)
|
|
yield connection
|
|
connection.close()
|
|
|
|
|
|
# --- window_for ---------------------------------------------------------------
|
|
|
|
def test_window_default_is_rest_of_today():
|
|
now = datetime(2026, 6, 10, 14, 30, 0)
|
|
start, end = window_for(now, None, None)
|
|
assert start == now
|
|
assert end == datetime(2026, 6, 11, 0, 0, 0)
|
|
|
|
|
|
def test_window_date_is_whole_day():
|
|
start, end = window_for(datetime(2026, 6, 10, 14, 30), "2026-06-15", None)
|
|
assert start == datetime(2026, 6, 15, 0, 0, 0)
|
|
assert end == datetime(2026, 6, 16, 0, 0, 0)
|
|
|
|
|
|
def test_window_days_spans_n_calendar_days():
|
|
now = datetime(2026, 6, 10, 14, 30, 0)
|
|
start, end = window_for(now, None, 3)
|
|
assert start == now
|
|
assert end == datetime(2026, 6, 13, 0, 0, 0)
|
|
|
|
|
|
def test_window_date_and_days_mutually_exclusive():
|
|
with pytest.raises(ValueError):
|
|
window_for(datetime(2026, 6, 10), "2026-06-15", 3)
|
|
|
|
|
|
def test_window_days_must_be_positive():
|
|
with pytest.raises(ValueError):
|
|
window_for(datetime(2026, 6, 10), None, 0)
|
|
|
|
|
|
# --- at -----------------------------------------------------------------------
|
|
|
|
def test_at_inside_window(conn):
|
|
rid = _add_reminder(conn, "at one")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T18:00:00"))
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 0, 0), datetime(2026, 6, 11, 0, 0))
|
|
assert len(fires) == 1
|
|
assert fires[0]["fire_time"] == datetime(2026, 6, 10, 18, 0)
|
|
assert fires[0]["schedule_type"] == "at"
|
|
|
|
|
|
def test_at_outside_window_excluded(conn):
|
|
rid = _add_reminder(conn, "at later")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-20T18:00:00"))
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 0, 0), datetime(2026, 6, 11, 0, 0))
|
|
assert fires == []
|
|
|
|
|
|
def test_at_end_is_exclusive(conn):
|
|
rid = _add_reminder(conn, "at edge")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-11T00:00:00"))
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 0, 0), datetime(2026, 6, 11, 0, 0))
|
|
assert fires == []
|
|
|
|
|
|
# --- cron ---------------------------------------------------------------------
|
|
|
|
def test_cron_enumerates_each_day(conn):
|
|
rid = _add_reminder(conn, "daily 9am")
|
|
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid, "0 9 * * *"))
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 0, 0), datetime(2026, 6, 13, 0, 0))
|
|
times = [f["fire_time"] for f in fires]
|
|
assert times == [
|
|
datetime(2026, 6, 10, 9, 0),
|
|
datetime(2026, 6, 11, 9, 0),
|
|
datetime(2026, 6, 12, 9, 0),
|
|
]
|
|
|
|
|
|
def test_cron_includes_fire_exactly_at_start(conn):
|
|
rid = _add_reminder(conn, "at start")
|
|
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid, "0 9 * * *"))
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 9, 0), datetime(2026, 6, 11, 0, 0))
|
|
assert [f["fire_time"] for f in fires] == [datetime(2026, 6, 10, 9, 0)]
|
|
|
|
|
|
# --- random -------------------------------------------------------------------
|
|
|
|
def test_random_matches_compute_fire_times(conn):
|
|
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),
|
|
)
|
|
|
|
from random_times import compute_fire_times
|
|
from datetime import date
|
|
expected = compute_fire_times(date(2026, 6, 10), "random reminder", {"times_per_day": 2, "window": "09:00-21:00"})
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 0, 0), datetime(2026, 6, 11, 0, 0))
|
|
assert [f["fire_time"] for f in fires] == expected
|
|
|
|
|
|
def test_random_respects_days_filter(conn):
|
|
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,),
|
|
)
|
|
|
|
# 2026-06-13 is a Saturday, 2026-06-14 Sunday — both filtered out; 2026-06-15 Monday allowed.
|
|
weekend = fires_in_window(conn, datetime(2026, 6, 13, 0, 0), datetime(2026, 6, 15, 0, 0))
|
|
assert weekend == []
|
|
monday = fires_in_window(conn, datetime(2026, 6, 15, 0, 0), datetime(2026, 6, 16, 0, 0))
|
|
assert len(monday) == 1
|
|
|
|
|
|
# --- general ------------------------------------------------------------------
|
|
|
|
def test_disabled_and_deleted_excluded(conn):
|
|
rid_off = _add_reminder(conn, "disabled", enabled=0)
|
|
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid_off, "0 9 * * *"))
|
|
rid_del = _add_reminder(conn, "deleted", deleted_at="now")
|
|
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid_del, "0 9 * * *"))
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 0, 0), datetime(2026, 6, 11, 0, 0))
|
|
assert fires == []
|
|
|
|
|
|
def test_fires_sorted_across_types(conn):
|
|
rid_cron = _add_reminder(conn, "morning")
|
|
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid_cron, "0 9 * * *"))
|
|
rid_at = _add_reminder(conn, "evening")
|
|
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid_at, "2026-06-10T18:00:00"))
|
|
|
|
fires = fires_in_window(conn, datetime(2026, 6, 10, 0, 0), datetime(2026, 6, 11, 0, 0))
|
|
assert [f["text"] for f in fires] == ["morning", "evening"]
|
|
|
|
|
|
def test_format_empty_window():
|
|
assert format_upcoming([], {}) == ["(nothing scheduled in this window)"]
|
|
|
|
|
|
def test_format_line_shape_uses_display_id():
|
|
# Internal id 5 maps to display ID 2 — the line shows the display ID.
|
|
lines = format_upcoming(
|
|
[{"fire_time": datetime(2026, 6, 10, 9, 0), "id": 5, "text": "call mom", "schedule_type": "cron"}],
|
|
{5: 2},
|
|
)
|
|
assert lines == ["2026-06-10 09:00 #2 call mom (cron)"]
|