feat: add backup alert background poller
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from media_library_viewer_api.services.settings_store import SettingsStore
|
||||
|
||||
|
||||
@@ -9,7 +10,9 @@ def test_backup_schema_created():
|
||||
store = SettingsStore(db_path)
|
||||
store.init_schema()
|
||||
with store.connect() as conn:
|
||||
tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'backup_%'").fetchall()
|
||||
tables = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'backup_%'"
|
||||
).fetchall()
|
||||
table_names = [t[0] for t in tables]
|
||||
assert "backup_jobs" in table_names
|
||||
assert "backup_runs" in table_names
|
||||
@@ -72,3 +75,34 @@ def test_alert_failed_status():
|
||||
assert len(alerts) == 1
|
||||
assert alerts[0]["alert_type"] == "failed_status"
|
||||
assert alerts[0]["severity"] == "critical"
|
||||
|
||||
|
||||
def test_backup_poller_detects_missed_schedule():
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from media_library_viewer_api.services.backup_poller import BackupAlertPoller
|
||||
from media_library_viewer_api.services.settings_store import SettingsStore
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = Path(tmpdir) / "test_settings.sqlite"
|
||||
store = SettingsStore(db_path)
|
||||
store.init_schema()
|
||||
|
||||
import time
|
||||
# Create job with created_at in the past so it's already overdue
|
||||
past_time = int(time.time()) - 200
|
||||
job = store.upsert_backup_job({
|
||||
"name": "old-job",
|
||||
"schedule_interval_seconds": 60,
|
||||
})
|
||||
# Override created_at to be in the past so missed schedule fires
|
||||
with store.connect() as conn:
|
||||
conn.execute("UPDATE backup_jobs SET created_at = ? WHERE id = ?", (past_time, job["id"]))
|
||||
|
||||
poller = BackupAlertPoller()
|
||||
poller._run_cycle(store)
|
||||
|
||||
alerts = store.list_backup_alerts(job_id=job["id"])
|
||||
assert len(alerts) == 1
|
||||
assert alerts[0]["alert_type"] == "missed_schedule"
|
||||
|
||||
Reference in New Issue
Block a user