SQLite-Ereignisse und Personenprofile vorbereiten

tests/test_event_store.py

@@ -0,0 +1,129 @@
+import asyncio
+from datetime import datetime, timedelta, timezone
+import sqlite3
+import stat
+import tempfile
+import unittest
+from pathlib import Path
+
+import httpx
+
+from halloween.event_store import EventStore
+from halloween.web import create_app
+
+
+class EventStoreTest(unittest.TestCase):
+    def test_events_and_profiles_persist_and_delete_keeps_events(self):
+        async def scenario(folder: Path):
+            headers = {"X-Halloween-UI": "1"}
+            app = create_app(demo=True, state_dir=folder)
+            transport = httpx.ASGITransport(app=app)
+            async with httpx.AsyncClient(transport=transport, base_url="http://local") as client:
+                self.assertEqual((await client.get("/api/events")).json()["events"], [])
+                self.assertEqual((await client.post("/api/profiles", json={"label": "Gast"})).status_code, 403)
+                profile_response = await client.post(
+                    "/api/profiles", headers=headers, json={"label": "  Gast  "}
+                )
+                self.assertEqual(profile_response.status_code, 200)
+                profile_id = profile_response.json()["profile"]["id"]
+                self.assertEqual(profile_response.json()["profile"]["label"], "Gast")
+                await client.post("/api/zone-clear", headers=headers)
+                triggered = await client.post("/api/test-trigger", headers=headers)
+                self.assertTrue(triggered.json()["accepted"])
+                self.assertFalse((await client.post("/api/test-trigger", headers=headers)).json()["accepted"])
+                events = (await client.get("/api/events")).json()["events"]
+                self.assertEqual(len(events), 1)
+                self.assertEqual(events[0]["kind"], "test_trigger")
+                event_id = events[0]["id"]
+                self.assertEqual(
+                    (await client.patch(
+                        f"/api/events/{event_id}/profile", headers=headers,
+                        json={"profile_id": profile_id},
+                    )).status_code, 200,
+                )
+                self.assertEqual(
+                    (await client.get("/api/profiles")).json()["profiles"][0]["event_count"], 1
+                )
+
+            restarted = create_app(demo=True, state_dir=folder)
+            transport = httpx.ASGITransport(app=restarted)
+            async with httpx.AsyncClient(transport=transport, base_url="http://local") as client:
+                event = (await client.get("/api/events")).json()["events"][0]
+                self.assertEqual(event["profile_id"], profile_id)
+                self.assertEqual(event["profile_label"], "Gast")
+                self.assertEqual(
+                    (await client.delete(f"/api/profiles/{profile_id}", headers=headers)).status_code,
+                    200,
+                )
+                remaining = (await client.get("/api/events")).json()["events"]
+                self.assertEqual(len(remaining), 1)
+                self.assertIsNone(remaining[0]["profile_id"])
+
+        with tempfile.TemporaryDirectory() as folder:
+            asyncio.run(scenario(Path(folder)))
+            mode = stat.S_IMODE((Path(folder) / "events.sqlite3").stat().st_mode)
+            self.assertEqual(mode, 0o600)
+
+    def test_existing_event_table_is_extended_without_losing_rows(self):
+        with tempfile.TemporaryDirectory() as folder:
+            db_path = Path(folder) / "events.sqlite3"
+            with sqlite3.connect(db_path) as connection:
+                connection.execute(
+                    "CREATE TABLE events (id INTEGER PRIMARY KEY, occurred_at TEXT NOT NULL, "
+                    "kind TEXT NOT NULL, mode TEXT NOT NULL)"
+                )
+                connection.execute(
+                    "INSERT INTO events (occurred_at, kind, mode) VALUES (?, ?, ?)",
+                    ("2026-10-01T00:00:00+00:00", "test_trigger", "voll"),
+                )
+            store = EventStore(Path(folder))
+            self.assertEqual(len(store.recent()), 1)
+            self.assertIsNone(store.recent()[0]["profile_id"])
+
+    def test_auto_profiles_expire_after_30_days_and_manual_profiles_remain(self):
+        with tempfile.TemporaryDirectory() as folder:
+            store = EventStore(Path(folder))
+            manual = store.create_profile("Bekannter Gast")
+            now = datetime.now(timezone.utc)
+            vector = [1.0] + [0.0] * 31
+            auto = store.create_auto_profile(vector, "test-model", now=now)
+            self.assertEqual(
+                store.best_template_match(vector, "test-model", now=now)[0], auto["id"]
+            )
+            self.assertIsNone(store.best_template_match(vector, "other-model", now=now)[0])
+            event_id = store.record_test_trigger("voll")
+            store.assign_profile(event_id, auto["id"])
+            self.assertEqual(store.cleanup_expired(now + timedelta(days=29)), 0)
+            self.assertEqual(store.cleanup_expired(now + timedelta(days=31)), 1)
+            self.assertEqual([profile["id"] for profile in store.profiles()], [manual["id"]])
+            self.assertIsNone(store.recent()[0]["profile_id"])
+            self.assertIsNone(
+                store.best_template_match(vector, "test-model", now=now + timedelta(days=31))[0]
+            )
+
+    def test_expired_profiles_are_removed_during_read_without_restart(self):
+        with tempfile.TemporaryDirectory() as folder:
+            store = EventStore(Path(folder))
+            past = datetime.now(timezone.utc) - timedelta(days=31)
+            auto = store.create_auto_profile([1.0] + [0.0] * 31, "test-model", now=past)
+            self.assertEqual(store.profiles(), [])
+            with sqlite3.connect(store.path) as connection:
+                self.assertEqual(
+                    connection.execute(
+                        "SELECT COUNT(*) FROM profile_templates WHERE profile_id = ?",
+                        (auto["id"],),
+                    ).fetchone()[0], 0,
+                )
+
+    def test_repeat_sighting_extends_retention_from_last_seen(self):
+        with tempfile.TemporaryDirectory() as folder:
+            store = EventStore(Path(folder))
+            now = datetime.now(timezone.utc)
+            auto = store.create_auto_profile([1.0] + [0.0] * 31, "test-model", now=now)
+            store.touch_auto_profile(auto["id"], now=now + timedelta(days=20))
+            self.assertEqual(store.cleanup_expired(now + timedelta(days=31)), 0)
+            self.assertEqual(store.cleanup_expired(now + timedelta(days=50)), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()