lachtan/nanobot-runtime
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dalsi vlna cisteni /remind od Claude

Browse Source
This commit is contained in:
lachtan
2026-06-10 08:48:26 +02:00
parent edf823fc72
commit abdfd8c67d
9 changed files with 237 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
skills/remind/tests/test_remind_edit.py
View File

@@ -1,11 +1,8 @@
import json
import os
import sqlite3
import sys
from datetime import datetime
from pathlib import Path
import pytest
SCRIPTS = Path(__file__).parent.parent / "scripts"
sys.path.insert(0, str(SCRIPTS))
@@ -39,7 +36,7 @@ def test_add_list(tmp_path, capsys):
ret = _run(db_path, ["list"])
captured = capsys.readouterr()
assert ret == 0
assert "drink water" in captured.out
assert "#1 drink water [enabled]" in captured.out
assert "cron: 0 9 * * *" in captured.out
@@ -77,7 +74,7 @@ def test_remove(tmp_path, capsys):
ret = _run(db_path, ["list"])
captured = capsys.readouterr()
assert captured.out.strip() == ""
assert "(no active reminders)" in captured.out
def test_remove_no_match(tmp_path, capsys):
@@ -257,3 +254,35 @@ def test_delivered_lists_deliveries(tmp_path, capsys):
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_edit.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

169
skills/remind/tests/test_remind_send.py
View File

@@ -1,12 +1,7 @@
import json
import os
import sqlite3
import sys
from datetime import datetime, timedelta
from datetime import date, datetime
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch
SCRIPTS = Path(__file__).parent.parent / "scripts"
sys.path.insert(0, str(SCRIPTS))
@@ -16,40 +11,45 @@ import remind_send
def _run_send(db_path, now=None):
"""Run remind_send main with a temporary DB path and optional mocked now."""
"""Run remind_send.main with a temporary DB, a stubbed config, and an optional fixed now."""
original_db_path = remind_send.DB_PATH
original_now = remind_send._now_prague
original_config = remind_send._telegram_config
try:
remind_send.DB_PATH = db_path
remind_send._telegram_config = lambda: ("token", "chat")
if now is not None:
remind_send._now = lambda: now
remind_send._now_prague = 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)
remind_send._now_prague = original_now
remind_send._telegram_config = original_config
def test_due_at_delivers_once(tmp_path, capsys):
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]
def test_due_at_delivers_once(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', '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),
)
rid = _add_reminder(conn, "at reminder")
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_called_once_with("⏰ Reminder: at reminder")
mock_send.assert_called_once_with("⏰ Reminder: at reminder", "token", "chat")
# second run — dedup
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
@@ -57,27 +57,20 @@ def test_due_at_delivers_once(tmp_path, capsys):
mock_send.assert_not_called()
def test_due_cron_delivers_once(tmp_path, capsys):
def test_due_cron_delivers_once(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', '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 * * *"),
)
rid = _add_reminder(conn, "cron reminder")
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")
mock_send.assert_called_once_with("⏰ Reminder: cron reminder", "token", "chat")
# second run — dedup
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
@@ -85,16 +78,12 @@ def test_due_cron_delivers_once(tmp_path, capsys):
mock_send.assert_not_called()
def test_due_random_delivers_once(tmp_path, capsys):
def test_due_random_delivers_once(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', 'now', 'now')",
("random reminder",),
)
rid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
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),
@@ -102,15 +91,14 @@ def test_due_random_delivers_once(tmp_path, capsys):
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"})
fires = compute_fire_times(date(2026, 6, 10), "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")
mock_send.assert_called_once_with("⏰ Reminder: random reminder", "token", "chat")
# all deduped now
for ft in fires:
@@ -119,43 +107,53 @@ def test_due_random_delivers_once(tmp_path, capsys):
mock_send.assert_not_called()
def test_disabled_not_sent(tmp_path, capsys):
def test_due_random_with_days_filter_delivers_on_allowed_day(tmp_path):
db_path = tmp_path / "test.sqlite"
init_db(db_path)
conn = get_db(db_path)
try:
rid = _add_reminder(conn, "weekday only")
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"),
"INSERT INTO schedule_random (reminder_id, times_per_day, window_start, window_end, days_filter) "
"VALUES (?, 1, 540, 1260, '1-5')",
(rid,),
)
finally:
conn.close()
from random_times import compute_fire_times
wednesday = date(2026, 6, 10)
fires = compute_fire_times(wednesday, "weekday only", {"times_per_day": 1, "window": "09:00-21:00", "days": "1-5"})
assert len(fires) == 1 # 2026-06-10 is a Wednesday, allowed by 1-5
with patch.object(remind_send, "_send_telegram", return_value=None) as mock_send:
_run_send(db_path, fires[0])
mock_send.assert_called_once_with("⏰ Reminder: weekday only", "token", "chat")
def test_disabled_not_sent(tmp_path):
db_path = tmp_path / "test.sqlite"
init_db(db_path)
conn = get_db(db_path)
try:
rid = _add_reminder(conn, "disabled reminder", enabled=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):
def test_deleted_not_sent(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, 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"),
)
rid = _add_reminder(conn, "deleted reminder", deleted_at="now")
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T10:00:00"))
finally:
conn.close()
@@ -165,20 +163,13 @@ def test_deleted_not_sent(tmp_path, capsys):
mock_send.assert_not_called()
def test_delivery_failure_logged(tmp_path, capsys):
def test_delivery_failure_logged(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', '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"),
)
rid = _add_reminder(conn, "fail reminder")
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid, "2026-06-10T10:00:00"))
finally:
conn.close()
@@ -198,6 +189,26 @@ def test_delivery_failure_logged(tmp_path, capsys):
conn.close()
def test_failed_fire_retries(tmp_path):
"""A fire that failed (status='failed') is not deduped — the next run retries it."""
db_path = tmp_path / "test.sqlite"
init_db(db_path)
conn = get_db(db_path)
try:
rid = _add_reminder(conn, "retry me")
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("down")):
_run_send(db_path, now)
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: retry me", "token", "chat")
def test_schedule_type_correct_despite_id_collision(tmp_path):
"""A cron fire must record schedule_type='cron' even when schedule_cron.id collides
with a schedule_at.id (each schedule table has its own AUTOINCREMENT sequence)."""
@@ -205,20 +216,10 @@ def test_schedule_type_correct_despite_id_collision(tmp_path):
init_db(db_path)
conn = get_db(db_path)
try:
conn.execute(
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES ('at one', 1, 'Europe/Prague', 'now', 'now')"
)
rid_at = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
conn.execute(
"INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid_at, "2030-01-01T00:00:00")
)
conn.execute(
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES ('cron one', 1, 'Europe/Prague', 'now', 'now')"
)
rid_cron = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
conn.execute(
"INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid_cron, "0 10 * * *")
)
rid_at = _add_reminder(conn, "at one")
conn.execute("INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)", (rid_at, "2030-01-01T00:00:00"))
rid_cron = _add_reminder(conn, "cron one")
conn.execute("INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)", (rid_cron, "0 10 * * *"))
# schedule_at.id and schedule_cron.id both equal 1 here — the collision the fix guards against.
assert conn.execute("SELECT id FROM schedule_at").fetchone()["id"] == 1
assert conn.execute("SELECT id FROM schedule_cron").fetchone()["id"] == 1