migrace /remind na sqlite
This commit is contained in:
lachtan
124
skills/remind/tests/test_db.py
Normal file
124
skills/remind/tests/test_db.py
Normal file
@@ -0,0 +1,124 @@
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from db import get_db, init_db
|
||||
|
||||
|
||||
def test_init_db_creates_tables(tmp_path):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
tables = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
||||
).fetchall()
|
||||
names = [r["name"] for r in tables]
|
||||
assert "reminders" in names
|
||||
assert "schedule_at" in names
|
||||
assert "schedule_cron" in names
|
||||
assert "schedule_random" in names
|
||||
assert "reminder_fires" in names
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_foreign_keys_enforced(tmp_path):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
conn.execute(
|
||||
"INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (999, '2026-06-10T10:00:00')"
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_reminder_constraints(tmp_path):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES ('', 1, 'UTC', 'now', 'now')"
|
||||
)
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES ('x', 2, 'UTC', 'now', 'now')"
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_schedule_random_window_constraint(tmp_path):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES ('x', 1, 'UTC', 'now', 'now')"
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
conn.execute(
|
||||
"INSERT INTO schedule_random (reminder_id, times_per_day, window_start, window_end) VALUES (?, 1, 1200, 600)",
|
||||
(rid,),
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_reminder_crud(tmp_path):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES (?, 1, 'Europe/Prague', '2026-06-10T10:00:00', '2026-06-10T10:00:00')",
|
||||
("test reminder",),
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute(
|
||||
"INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)",
|
||||
(rid, "2026-06-11T09:00:00"),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)",
|
||||
(rid, "0 9 * * *"),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO schedule_random (reminder_id, times_per_day, window_start, window_end) VALUES (?, ?, ?, ?)",
|
||||
(rid, 2, 540, 1260),
|
||||
)
|
||||
|
||||
row = conn.execute("SELECT * FROM reminders WHERE id = ?", (rid,)).fetchone()
|
||||
assert row["text"] == "test reminder"
|
||||
|
||||
at_rows = conn.execute("SELECT * FROM schedule_at WHERE reminder_id = ?", (rid,)).fetchall()
|
||||
assert len(at_rows) == 1
|
||||
assert at_rows[0]["at_datetime"] == "2026-06-11T09:00:00"
|
||||
|
||||
cron_rows = conn.execute("SELECT * FROM schedule_cron WHERE reminder_id = ?", (rid,)).fetchall()
|
||||
assert len(cron_rows) == 1
|
||||
assert cron_rows[0]["cron_expr"] == "0 9 * * *"
|
||||
|
||||
random_rows = conn.execute("SELECT * FROM schedule_random WHERE reminder_id = ?", (rid,)).fetchall()
|
||||
assert len(random_rows) == 1
|
||||
assert random_rows[0]["times_per_day"] == 2
|
||||
|
||||
# soft delete
|
||||
conn.execute("UPDATE reminders SET deleted_at = '2026-06-10T11:00:00' WHERE id = ?", (rid,))
|
||||
row = conn.execute("SELECT deleted_at FROM reminders WHERE id = ?", (rid,)).fetchone()
|
||||
assert row["deleted_at"] is not None
|
||||
|
||||
# cascade delete
|
||||
conn.execute("DELETE FROM reminders WHERE id = ?", (rid,))
|
||||
assert conn.execute("SELECT 1 FROM schedule_at WHERE reminder_id = ?", (rid,)).fetchone() is None
|
||||
assert conn.execute("SELECT 1 FROM schedule_cron WHERE reminder_id = ?", (rid,)).fetchone() is None
|
||||
assert conn.execute("SELECT 1 FROM schedule_random WHERE reminder_id = ?", (rid,)).fetchone() is None
|
||||
finally:
|
||||
conn.close()
|
||||
184
skills/remind/tests/test_remind_edit.py
Normal file
184
skills/remind/tests/test_remind_edit.py
Normal file
@@ -0,0 +1,184 @@
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
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
|
||||
import remind_edit
|
||||
|
||||
|
||||
def _run(db_path, argv):
|
||||
"""Run remind_edit main with a temporary DB path."""
|
||||
original_db_path = remind_edit.DB_PATH
|
||||
try:
|
||||
remind_edit.DB_PATH = db_path
|
||||
# Patch the module-level DB_PATH used by functions
|
||||
sys.argv = ["remind_edit.py"] + argv
|
||||
return remind_edit.main()
|
||||
finally:
|
||||
remind_edit.DB_PATH = original_db_path
|
||||
|
||||
|
||||
def test_add_list(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
ret = _run(db_path, ["add", "--text", "drink water", "--cron", "0 9 * * *"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert data["added"]["text"] == "drink water"
|
||||
assert len(data["added"]["cron"]) == 1
|
||||
|
||||
ret = _run(db_path, ["list"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert len(data["reminders"]) == 1
|
||||
assert data["reminders"][0]["text"] == "drink water"
|
||||
|
||||
|
||||
def test_add_at_validation(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
ret = _run(db_path, ["add", "--text", "x", "--at", "not-a-date"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 1
|
||||
assert "invalid --at datetime" in captured.err
|
||||
|
||||
|
||||
def test_add_cron_validation(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
ret = _run(db_path, ["add", "--text", "x", "--cron", "bad"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 1
|
||||
assert "invalid cron expression" in captured.err
|
||||
|
||||
|
||||
def test_remove(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
_run(db_path, ["add", "--text", "stretch", "--cron", "0 10 * * *"])
|
||||
capsys.readouterr() # clear setup output
|
||||
ret = _run(db_path, ["remove", "--keyword", "stretch"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert data["removed"]["text"] == "stretch"
|
||||
|
||||
ret = _run(db_path, ["list"])
|
||||
captured = capsys.readouterr()
|
||||
data = json.loads(captured.out)
|
||||
assert len(data["reminders"]) == 0
|
||||
|
||||
|
||||
def test_remove_no_match(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
ret = _run(db_path, ["remove", "--keyword", "nonexistent"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 1
|
||||
assert "no match" in captured.err
|
||||
|
||||
|
||||
def test_remove_ambiguous(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
_run(db_path, ["add", "--text", "drink water", "--cron", "0 9 * * *"])
|
||||
_run(db_path, ["add", "--text", "drink tea", "--cron", "0 10 * * *"])
|
||||
ret = _run(db_path, ["remove", "--keyword", "drink"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 1
|
||||
assert "ambiguous" in captured.err
|
||||
|
||||
|
||||
def test_edit_text(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
_run(db_path, ["add", "--text", "old text", "--cron", "0 9 * * *"])
|
||||
capsys.readouterr() # clear setup output
|
||||
ret = _run(db_path, ["edit", "--keyword", "old", "--text", "new text"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert data["edited"]["text"] == "new text"
|
||||
|
||||
|
||||
def test_edit_replace_schedules(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
_run(db_path, ["add", "--text", "task", "--cron", "0 9 * * *"])
|
||||
capsys.readouterr() # clear setup output
|
||||
ret = _run(db_path, ["edit", "--keyword", "task", "--replace-schedules", "--at", "2026-06-15T10:00:00"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert len(data["edited"]["cron"]) == 0
|
||||
assert len(data["edited"]["at"]) == 1
|
||||
|
||||
|
||||
def test_enable_disable(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
_run(db_path, ["add", "--text", "toggle me", "--cron", "0 9 * * *"])
|
||||
capsys.readouterr() # clear setup output
|
||||
|
||||
ret = _run(db_path, ["disable", "--keyword", "toggle"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert data["disabled"]["enabled"] == 0
|
||||
|
||||
ret = _run(db_path, ["enable", "--keyword", "toggle"])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert data["enabled"]["enabled"] == 1
|
||||
|
||||
|
||||
def test_add_combined_schedules(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
ret = _run(db_path, [
|
||||
"add", "--text", "multi",
|
||||
"--at", "2026-06-15T10:00:00",
|
||||
"--cron", "0 9 * * *",
|
||||
"--random-times-per-day", "2",
|
||||
"--random-window", "08:00-20:00",
|
||||
])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 0
|
||||
data = json.loads(captured.out)
|
||||
assert len(data["added"]["at"]) == 1
|
||||
assert len(data["added"]["cron"]) == 1
|
||||
assert len(data["added"]["random"]) == 1
|
||||
|
||||
|
||||
def test_add_random_validation(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
|
||||
ret = _run(db_path, [
|
||||
"add", "--text", "bad",
|
||||
"--random-times-per-day", "2",
|
||||
"--random-window", "08:00-08:01",
|
||||
])
|
||||
captured = capsys.readouterr()
|
||||
assert ret == 1
|
||||
assert "window" in captured.err.lower() or "gap" in captured.err.lower()
|
||||
198
skills/remind/tests/test_remind_send.py
Normal file
198
skills/remind/tests/test_remind_send.py
Normal file
@@ -0,0 +1,198 @@
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
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 path and optional mocked now."""
|
||||
original_db_path = remind_send.DB_PATH
|
||||
try:
|
||||
remind_send.DB_PATH = db_path
|
||||
if now is not None:
|
||||
remind_send._now = lambda: now
|
||||
remind_send.main()
|
||||
finally:
|
||||
remind_send.DB_PATH = original_db_path
|
||||
remind_send._now = lambda: datetime.now(remind_send.TZ).replace(tzinfo=None)
|
||||
|
||||
|
||||
def test_due_at_delivers_once(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES (?, 1, 'Europe/Prague', 'now', 'now')",
|
||||
("at reminder",),
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
fire_time = "2026-06-10T10:00:00"
|
||||
conn.execute(
|
||||
"INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)",
|
||||
(rid, fire_time),
|
||||
)
|
||||
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")
|
||||
|
||||
# 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, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES (?, 1, 'Europe/Prague', 'now', 'now')",
|
||||
("cron reminder",),
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
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")
|
||||
|
||||
# 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, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES (?, 1, 'Europe/Prague', 'now', 'now')",
|
||||
("random reminder",),
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute(
|
||||
"INSERT INTO schedule_random (reminder_id, times_per_day, window_start, window_end) VALUES (?, ?, ?, ?)",
|
||||
(rid, 2, 540, 1260),
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# compute expected fire times for the date
|
||||
from random_times import compute_fire_times
|
||||
fires = compute_fire_times(datetime(2026, 6, 10).date(), "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")
|
||||
|
||||
# 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_disabled_not_sent(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES (?, 0, 'Europe/Prague', 'now', 'now')",
|
||||
("disabled reminder",),
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[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, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at, deleted_at) VALUES (?, 1, 'Europe/Prague', 'now', 'now', 'now')",
|
||||
("deleted reminder",),
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[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_delivery_failure_logged(tmp_path, capsys):
|
||||
db_path = tmp_path / "test.sqlite"
|
||||
init_db(db_path)
|
||||
conn = get_db(db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES (?, 1, 'Europe/Prague', 'now', 'now')",
|
||||
("fail reminder",),
|
||||
)
|
||||
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[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", 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()
|
||||
Reference in New Issue
Block a user