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(): lines = format_upcoming([ {"fire_time": datetime(2026, 6, 10, 9, 0), "id": 1, "text": "call mom", "schedule_type": "cron"}, ]) assert lines == ["2026-06-10 09:00 #1 call mom (cron)"]