260 lines
7.9 KiB
Python
260 lines
7.9 KiB
Python
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
|
|
assert "drink water" in captured.out
|
|
assert "cron: 0 9 * * *" in captured.out
|
|
|
|
|
|
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()
|
|
assert captured.out.strip() == ""
|
|
|
|
|
|
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()
|
|
|
|
|
|
def test_remove_by_id_disambiguates_duplicates(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 water", "--cron", "0 10 * * *"])
|
|
capsys.readouterr()
|
|
|
|
ret = _run(db_path, ["remove", "--keyword", "drink water"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
assert "ambiguous" in captured.err
|
|
|
|
ret = _run(db_path, ["remove", "--id", "1"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 0
|
|
assert json.loads(captured.out)["removed"]["id"] == 1
|
|
|
|
# the second duplicate is still there
|
|
ret = _run(db_path, ["list"])
|
|
captured = capsys.readouterr()
|
|
assert "drink water" in captured.out
|
|
|
|
|
|
def test_resolve_requires_id_or_keyword(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
ret = _run(db_path, ["remove"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
assert "id or --keyword" in captured.err
|
|
|
|
|
|
def test_edit_replace_schedules_requires_new_schedule(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
_run(db_path, ["add", "--text", "task", "--cron", "0 9 * * *"])
|
|
capsys.readouterr()
|
|
ret = _run(db_path, ["edit", "--keyword", "task", "--replace-schedules"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
assert "replace-schedules" in captured.err
|
|
|
|
|
|
def test_delivered_lists_deliveries(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 ('took pills', 1, 'Europe/Prague', 'now', 'now')"
|
|
)
|
|
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute(
|
|
"INSERT INTO reminder_fires (reminder_id, schedule_id, schedule_type, fire_time, delivered_at, status) "
|
|
"VALUES (?, 1, 'cron', '2026-06-01T09:00:00', '2026-06-01T09:00:01', 'delivered')",
|
|
(rid,),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO reminder_fires (reminder_id, schedule_id, schedule_type, fire_time, delivered_at, status, error_message) "
|
|
"VALUES (?, 1, 'cron', '2026-06-01T10:00:00', NULL, 'failed', 'boom')",
|
|
(rid,),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
ret = _run(db_path, ["delivered", "--since", "2026-06-01"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 0
|
|
assert "took pills" in captured.out
|
|
assert "2026-06-01T09:00:01" in captured.out
|
|
# failed fire is not reported as delivered
|
|
assert captured.out.count("took pills") == 1
|