ops: consistent rotated db backups + cron, consecutive-api-error staff alert

tests/test_spec_ops_backup.py

@@ -0,0 +1,100 @@
+"""Unit coverage for SPEC-012 ops hardening (OPS-13/14/16)."""
+
+import gzip
+import sqlite3
+import stat
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+from fjerkroa_bot.persistence import PersistentStore
+
+from .test_spec_ops import OpsBase
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "deploy"))
+import backup_db  # noqa: E402
+
+
+class TestSnapshotConsistency(unittest.TestCase):
+    def test_snapshot_roundtrips(self):
+        """OPS-13: a gzipped snapshot restores to a readable DB with the same rows, 0600."""
+        with tempfile.TemporaryDirectory() as tmp:
+            db = Path(tmp) / "bot.db"
+            store = PersistentStore(db)
+            store.save_history("chat", [{"role": "user", "content": "hei"}])
+            store.add_user_fact("alice", "likes espresso", "self")
+
+            dest = Path(tmp) / "snap.db.gz"
+            backup_db.snapshot(db, dest)
+            self.assertEqual(stat.S_IMODE(dest.stat().st_mode), 0o600)
+
+            restored = Path(tmp) / "restored.db"
+            with gzip.open(dest, "rb") as gz, open(restored, "wb") as out:
+                out.write(gz.read())
+            conn = sqlite3.connect(restored)
+            try:
+                rows = conn.execute("SELECT content FROM history WHERE channel='chat'").fetchall()
+                facts = conn.execute("SELECT fact FROM user_facts").fetchall()
+            finally:
+                conn.close()
+            self.assertEqual(rows, [("hei",)])
+            self.assertEqual(facts, [("likes espresso",)])
+
+    def test_snapshot_during_writes(self):
+        """OPS-13: snapshot succeeds while another connection holds the DB open (WAL)."""
+        with tempfile.TemporaryDirectory() as tmp:
+            db = Path(tmp) / "bot.db"
+            store = PersistentStore(db)
+            store.save_history("chat", [{"role": "user", "content": "x"}])
+            live = sqlite3.connect(db)  # simulate the running bot's open handle
+            live.execute("PRAGMA journal_mode=WAL")
+            try:
+                dest = Path(tmp) / "snap.db.gz"
+                backup_db.snapshot(db, dest)  # must not raise
+                self.assertTrue(dest.exists())
+            finally:
+                live.close()
+
+
+class TestRotation(unittest.TestCase):
+    def test_victims_keeps_newest(self):
+        """OPS-14: only the oldest beyond `keep` are selected for deletion."""
+        names = [f"bot-2026070{d}-000000.db.gz" for d in range(1, 8)]  # 7 chronological
+        victims = backup_db.victims(list(reversed(names)), keep=3)
+        self.assertEqual(victims, names[:4])  # oldest 4 removed, newest 3 kept
+
+    def test_victims_under_keep_deletes_nothing(self):
+        """OPS-14: fewer than `keep` backups -> nothing deleted."""
+        self.assertEqual(backup_db.victims(["bot-20260701-000000.db.gz"], keep=14), [])
+
+    def test_rotate_on_disk(self):
+        """OPS-14: rotate removes the right files from a real dir."""
+        with tempfile.TemporaryDirectory() as tmp:
+            for d in range(1, 6):
+                (Path(tmp) / f"bot-2026070{d}-000000.db.gz").write_bytes(b"x")
+            removed = backup_db.rotate(Path(tmp), keep=2)
+            self.assertEqual(removed, 3)
+            self.assertEqual(len(list(Path(tmp).glob("bot-*.db.gz"))), 2)
+
+
+class TestApiErrorAlert(OpsBase):
+    async def test_threshold_alert_and_reset(self):
+        """OPS-16: N consecutive failures fire one staff alert; success resets."""
+        self.bot.config["api-error-alert-threshold"] = 3
+        self.bot.send_message_with_typing = AsyncMock(side_effect=RuntimeError("boom"))
+        origin = MagicMock()
+        from fjerkroa_bot.ai_responder import AIMessage
+
+        for _ in range(3):
+            await self.bot.respond(AIMessage("alice", "hei", "chat"), origin)
+        self.assertEqual(self.bot.staff_channel.send.await_count, 1)  # exactly one alert at threshold
+        self.assertEqual(self.bot._consecutive_api_errors, 3)
+
+        # a success resets the counter
+        from fjerkroa_bot.ai_responder import AIResponse
+
+        self.bot.send_message_with_typing = AsyncMock(return_value=AIResponse(None, False, "chat", None, None, False, False))
+        await self.bot.respond(AIMessage("alice", "hei", "chat"), origin)
+        self.assertEqual(self.bot._consecutive_api_errors, 0)