125 lines
4.5 KiB
Python
125 lines
4.5 KiB
Python
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()
|