425 lines
14 KiB
Python
425 lines
14 KiB
Python
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
SCRIPTS = Path(__file__).parent.parent / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
from db import get_db, init_db
|
|
import remind_cli
|
|
import store
|
|
|
|
|
|
def _run(db_path, argv):
|
|
"""Run remind_cli main with a temporary DB path."""
|
|
original_db_path = remind_cli.DB_PATH
|
|
try:
|
|
remind_cli.DB_PATH = db_path
|
|
# Patch the module-level DB_PATH used by functions
|
|
sys.argv = ["remind_cli.py"] + argv
|
|
return remind_cli.main()
|
|
finally:
|
|
remind_cli.DB_PATH = original_db_path
|
|
|
|
|
|
def _seed_duplicate(db_path, text, cron):
|
|
"""Insert a reminder directly via store, bypassing the CLI duplicate guard.
|
|
|
|
Used to construct pre-existing duplicate texts that the disambiguation code
|
|
(`--id`, ambiguous keyword) must still handle even though `add` now blocks them.
|
|
"""
|
|
with store.transaction(db_path) as conn:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
rid = store.insert_reminder(conn, text, now)
|
|
store.insert_schedules(conn, rid, None, [cron], None)
|
|
return rid
|
|
|
|
|
|
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 "#1 drink water [enabled]" 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 "(no active reminders)" in captured.out
|
|
|
|
|
|
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_text_by_id(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
_run(db_path, ["add", "--text", "original text", "--cron", "0 9 * * *"])
|
|
capsys.readouterr() # clear setup output
|
|
ret = _run(db_path, ["edit", "--id", "1", "--text", "renamed"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 0
|
|
data = json.loads(captured.out)
|
|
assert data["edited"]["text"] == "renamed"
|
|
assert len(data["edited"]["cron"]) == 1
|
|
|
|
|
|
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_add_rejects_duplicate_text(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
ret = _run(db_path, ["add", "--text", "call mom", "--at", "2026-12-01T08:00:00"])
|
|
capsys.readouterr()
|
|
assert ret == 0
|
|
|
|
ret = _run(db_path, ["add", "--text", "call mom", "--at", "2026-12-01T20:00:00"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
err = json.loads(captured.err)
|
|
assert err["error"] == "duplicate text"
|
|
assert err["display_id"] == 1
|
|
|
|
# The first reminder is untouched — no second record was created.
|
|
ret = _run(db_path, ["list"])
|
|
captured = capsys.readouterr()
|
|
assert captured.out.count("call mom") == 1
|
|
|
|
|
|
def test_add_duplicate_case_and_diacritics_insensitive(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
_run(db_path, ["add", "--text", "Čaj", "--cron", "0 9 * * *"])
|
|
capsys.readouterr()
|
|
|
|
ret = _run(db_path, ["add", "--text", " čaj ", "--cron", "0 10 * * *"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
assert json.loads(captured.err)["error"] == "duplicate text"
|
|
|
|
|
|
def test_edit_text_collision_rejected(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
_run(db_path, ["add", "--text", "buy milk", "--cron", "0 9 * * *"])
|
|
_run(db_path, ["add", "--text", "buy bread", "--cron", "0 10 * * *"])
|
|
capsys.readouterr()
|
|
|
|
ret = _run(db_path, ["edit", "--id", "2", "--text", "buy milk"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
assert json.loads(captured.err)["error"] == "duplicate text"
|
|
|
|
# Editing a reminder's text to itself (no real change) must still work.
|
|
ret = _run(db_path, ["edit", "--id", "1", "--text", "buy milk"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 0
|
|
|
|
|
|
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 * * *"])
|
|
_seed_duplicate(db_path, "drink water", "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_display_id_renumbers_after_remove(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
for text in ("first", "second", "third"):
|
|
_run(db_path, ["add", "--text", text, "--cron", "0 9 * * *"])
|
|
capsys.readouterr()
|
|
|
|
# Display IDs follow insertion order: #1 first, #2 second, #3 third.
|
|
_run(db_path, ["remove", "--id", "1"]) # removes "first"
|
|
capsys.readouterr()
|
|
|
|
ret = _run(db_path, ["list"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 0
|
|
assert "#1 second [enabled]" in captured.out
|
|
assert "#2 third [enabled]" in captured.out
|
|
assert "first" not in captured.out
|
|
|
|
# After renumbering, display #1 is now "second".
|
|
ret = _run(db_path, ["remove", "--id", "1"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 0
|
|
assert json.loads(captured.out)["removed"]["text"] == "second"
|
|
|
|
|
|
def test_id_out_of_range_reports_display_id(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
_run(db_path, ["add", "--text", "only one", "--cron", "0 9 * * *"])
|
|
capsys.readouterr()
|
|
|
|
ret = _run(db_path, ["remove", "--id", "5"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
assert json.loads(captured.err) == {"error": "no match", "display_id": 5}
|
|
|
|
|
|
def test_ambiguous_keyword_returns_display_ids(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
|
|
_run(db_path, ["add", "--text", "drink water", "--cron", "0 9 * * *"])
|
|
_seed_duplicate(db_path, "drink water", "0 10 * * *")
|
|
capsys.readouterr()
|
|
|
|
ret = _run(db_path, ["remove", "--keyword", "drink"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 1
|
|
err = json.loads(captured.err)
|
|
assert err["error"] == "ambiguous"
|
|
assert sorted(m["display_id"] for m in err["matches"]) == [1, 2]
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_delivered_defaults_to_today(tmp_path, capsys):
|
|
db_path = tmp_path / "test.sqlite"
|
|
init_db(db_path)
|
|
today = datetime.now(remind_cli.PRAGUE).date().isoformat()
|
|
conn = get_db(db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) "
|
|
"VALUES ('today 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', ?, ?, 'delivered')",
|
|
(rid, f"{today}T09:00:00", f"{today}T09:00:01"),
|
|
)
|
|
# an older delivery must not show up when defaulting to today
|
|
conn.execute(
|
|
"INSERT INTO reminder_fires (reminder_id, schedule_id, schedule_type, fire_time, delivered_at, status) "
|
|
"VALUES (?, 1, 'cron', '2020-01-01T09:00:00', '2020-01-01T09:00:01', 'delivered')",
|
|
(rid,),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
ret = _run(db_path, ["delivered"])
|
|
captured = capsys.readouterr()
|
|
assert ret == 0
|
|
assert captured.out.count("today pills") == 1
|
|
assert "2020-01-01" not in captured.out
|