Halloween/tests/test_event_store.py

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()