frigate-notify/tests/test_api.py

"""API endpoint tests for frigate-notify v2."""

import sqlite3
import sys
from unittest.mock import MagicMock, patch

import pytest

from database import close_connection, get_connection, init_db, set_db_path


@pytest.fixture
def client(db_path):
    """Create a TestClient with a temporary database.

    Patches config.DB_PATH and mocks heavy dependencies (MQTT, anthropic,
    matrix_client, vision) so that main.py can be imported without them.
    """
    # Set DB path before anything imports the app
    set_db_path(db_path)
    init_db()

    # Mock modules that main.py imports at top level but aren't needed for API tests
    mock_modules = {}
    for mod_name in ("paho", "paho.mqtt", "paho.mqtt.client", "anthropic", "aiohttp"):
        if mod_name not in sys.modules:
            mock_modules[mod_name] = MagicMock()
            sys.modules[mod_name] = mock_modules[mod_name]

    # Ensure matrix_client and vision modules are available as mocks
    if "matrix_client" not in sys.modules:
        mock_modules["matrix_client"] = MagicMock()
        sys.modules["matrix_client"] = mock_modules["matrix_client"]
    if "vision" not in sys.modules:
        mock_modules["vision"] = MagicMock()
        sys.modules["vision"] = mock_modules["vision"]

    # Patch config DB_PATH
    from config import config

    original_db_path = config.DB_PATH
    config.DB_PATH = db_path

    # Now import the app (this will use the patched DB_PATH)
    # We need to patch the lifespan to avoid starting MQTT
    from fastapi.testclient import TestClient

    from main import app

    # Override the lifespan to just init DB without MQTT
    from contextlib import asynccontextmanager

    @asynccontextmanager
    async def test_lifespan(app):
        set_db_path(db_path)
        init_db()
        yield
        close_connection()

    app.router.lifespan_context = test_lifespan

    with TestClient(app) as tc:
        yield tc

    # Cleanup
    close_connection()
    config.DB_PATH = original_db_path

    # Remove mocked modules
    for mod_name in mock_modules:
        if mod_name in sys.modules and sys.modules[mod_name] is mock_modules[mod_name]:
            del sys.modules[mod_name]


# --- Helper to insert test data ---


def _insert_event(
    frigate_event_id="evt-001",
    camera="front_door",
    label="person",
    score=0.95,
    snapshot=None,
    embedding=None,
    match_status="unreviewed",
    matched_object_id=None,
    ai_description=None,
):
    """Insert a test event directly into the database. Returns the event id."""
    conn = get_connection()
    cursor = conn.execute(
        """INSERT INTO events
           (frigate_event_id, camera, label, score, snapshot, embedding,
            match_status, matched_object_id, ai_description, notified)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1)""",
        (
            frigate_event_id,
            camera,
            label,
            score,
            snapshot,
            embedding,
            match_status,
            matched_object_id,
            ai_description,
        ),
    )
    conn.commit()
    return cursor.lastrowid


def _insert_known_object(name="John", type_="person", action="quiet"):
    """Insert a test known object directly into the database. Returns the object id."""
    conn = get_connection()
    cursor = conn.execute(
        "INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
        (name, type_, action),
    )
    conn.commit()
    return cursor.lastrowid


# --- Health endpoint ---


class TestHealth:
    def test_health_returns_200(self, client):
        resp = client.get("/health")
        assert resp.status_code == 200
        data = resp.json()
        assert data["status"] == "ok"
        assert data["version"] == "2.0.0"


# --- Events endpoints ---


class TestEventsAPI:
    def test_list_events_empty(self, client):
        resp = client.get("/api/events")
        assert resp.status_code == 200
        assert resp.json() == []

    def test_list_events_returns_events(self, client):
        _insert_event(frigate_event_id="evt-list-1", camera="cam1", label="person")
        _insert_event(frigate_event_id="evt-list-2", camera="cam2", label="car")

        resp = client.get("/api/events")
        assert resp.status_code == 200
        data = resp.json()
        assert len(data) == 2

    def test_list_events_filter_by_camera(self, client):
        _insert_event(frigate_event_id="evt-cam1", camera="front_door")
        _insert_event(frigate_event_id="evt-cam2", camera="back_yard")

        resp = client.get("/api/events?camera=front_door")
        assert resp.status_code == 200
        data = resp.json()
        assert len(data) == 1
        assert data[0]["camera"] == "front_door"

    def test_list_events_filter_by_status(self, client):
        _insert_event(frigate_event_id="evt-s1", match_status="unreviewed")
        _insert_event(frigate_event_id="evt-s2", match_status="known")

        resp = client.get("/api/events?status=known")
        data = resp.json()
        assert len(data) == 1
        assert data[0]["match_status"] == "known"

    def test_list_events_has_snapshot_field(self, client):
        _insert_event(
            frigate_event_id="evt-snap", snapshot=b"\xff\xd8\xff\xe0fake-jpeg"
        )
        _insert_event(frigate_event_id="evt-nosnap", snapshot=None)

        resp = client.get("/api/events")
        data = resp.json()
        snap_event = next(e for e in data if e["frigate_event_id"] == "evt-snap")
        nosnap_event = next(e for e in data if e["frigate_event_id"] == "evt-nosnap")
        assert snap_event["has_snapshot"] is True
        assert nosnap_event["has_snapshot"] is False

    def test_list_events_with_matched_object_name(self, client):
        obj_id = _insert_known_object(name="Alice")
        _insert_event(
            frigate_event_id="evt-matched",
            match_status="known",
            matched_object_id=obj_id,
        )

        resp = client.get("/api/events")
        data = resp.json()
        assert data[0]["matched_object_name"] == "Alice"

    def test_get_event_not_found(self, client):
        resp = client.get("/api/events/99999")
        assert resp.status_code == 404

    def test_get_event_detail(self, client):
        event_id = _insert_event(frigate_event_id="evt-detail", camera="pool")

        resp = client.get(f"/api/events/{event_id}")
        assert resp.status_code == 200
        data = resp.json()
        assert data["frigate_event_id"] == "evt-detail"
        assert data["camera"] == "pool"

    def test_get_event_detail_includes_snapshot_base64(self, client):
        jpeg_bytes = b"\xff\xd8\xff\xe0fake-jpeg-detail"
        event_id = _insert_event(
            frigate_event_id="evt-detail-snap", camera="pool", snapshot=jpeg_bytes
        )

        resp = client.get(f"/api/events/{event_id}")
        assert resp.status_code == 200
        data = resp.json()
        assert data["has_snapshot"] is True
        assert data["snapshot_base64"] is not None

        import base64

        decoded = base64.b64decode(data["snapshot_base64"])
        assert decoded == jpeg_bytes

    def test_get_event_detail_no_snapshot_base64_when_missing(self, client):
        event_id = _insert_event(
            frigate_event_id="evt-detail-nosnap", camera="pool", snapshot=None
        )

        resp = client.get(f"/api/events/{event_id}")
        assert resp.status_code == 200
        data = resp.json()
        assert data["has_snapshot"] is False
        assert data["snapshot_base64"] is None

    def test_get_event_snapshot(self, client):
        jpeg_bytes = b"\xff\xd8\xff\xe0fake-jpeg-data"
        event_id = _insert_event(
            frigate_event_id="evt-snap-get", snapshot=jpeg_bytes
        )

        resp = client.get(f"/api/events/{event_id}/snapshot")
        assert resp.status_code == 200
        assert resp.headers["content-type"] == "image/jpeg"
        assert resp.content == jpeg_bytes

    def test_get_event_snapshot_not_found(self, client):
        resp = client.get("/api/events/99999/snapshot")
        assert resp.status_code == 404

    def test_get_event_snapshot_no_snapshot(self, client):
        event_id = _insert_event(frigate_event_id="evt-nosnap2", snapshot=None)

        resp = client.get(f"/api/events/{event_id}/snapshot")
        assert resp.status_code == 404

    def test_dismiss_event(self, client):
        event_id = _insert_event(frigate_event_id="evt-dismiss")

        resp = client.post(f"/api/events/{event_id}/dismiss")
        assert resp.status_code == 200

        # Verify status changed
        resp2 = client.get(f"/api/events/{event_id}")
        assert resp2.json()["match_status"] == "unknown"

    def test_dismiss_event_not_found(self, client):
        resp = client.post("/api/events/99999/dismiss")
        assert resp.status_code == 404

    def test_tag_event_with_existing_object(self, client):
        obj_id = _insert_known_object(name="Bob")
        event_id = _insert_event(frigate_event_id="evt-tag1")

        resp = client.post(
            f"/api/events/{event_id}/tag",
            json={"known_object_id": obj_id},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["matched_object_id"] == obj_id

        # Verify event updated
        resp2 = client.get(f"/api/events/{event_id}")
        event_data = resp2.json()
        assert event_data["match_status"] == "known"
        assert event_data["matched_object_id"] == obj_id
        assert event_data["match_confidence"] == 1.0

    def test_tag_event_creates_new_object(self, client):
        event_id = _insert_event(frigate_event_id="evt-tag2")

        resp = client.post(
            f"/api/events/{event_id}/tag",
            json={"name": "NewPerson", "type": "person"},
        )
        assert resp.status_code == 200
        new_obj_id = resp.json()["matched_object_id"]

        # Verify the known object was created
        conn = get_connection()
        cursor = conn.execute(
            "SELECT name, type FROM known_objects WHERE id = ?", (new_obj_id,)
        )
        row = cursor.fetchone()
        assert row["name"] == "NewPerson"
        assert row["type"] == "person"

    def test_tag_event_stores_embedding(self, client):
        fake_embedding = b"\x00\x01\x02\x03\x04"
        obj_id = _insert_known_object(name="EmbedTest")
        event_id = _insert_event(
            frigate_event_id="evt-embed", embedding=fake_embedding
        )

        client.post(
            f"/api/events/{event_id}/tag",
            json={"known_object_id": obj_id},
        )

        # Verify embedding was stored
        conn = get_connection()
        cursor = conn.execute(
            "SELECT embedding FROM embeddings WHERE known_object_id = ?", (obj_id,)
        )
        row = cursor.fetchone()
        assert row is not None
        assert bytes(row["embedding"]) == fake_embedding

    def test_tag_event_bad_request(self, client):
        event_id = _insert_event(frigate_event_id="evt-bad-tag")

        resp = client.post(
            f"/api/events/{event_id}/tag",
            json={},
        )
        assert resp.status_code == 400

    def test_tag_event_not_found(self, client):
        resp = client.post(
            "/api/events/99999/tag",
            json={"name": "Nobody", "type": "person"},
        )
        assert resp.status_code == 404

    def test_tag_event_known_object_not_found(self, client):
        event_id = _insert_event(frigate_event_id="evt-bad-obj")

        resp = client.post(
            f"/api/events/{event_id}/tag",
            json={"known_object_id": 99999},
        )
        assert resp.status_code == 404

    def test_list_events_pagination(self, client):
        for i in range(5):
            _insert_event(frigate_event_id=f"evt-page-{i}", camera="cam1")

        resp = client.get("/api/events?limit=2&offset=0")
        assert len(resp.json()) == 2

        resp2 = client.get("/api/events?limit=2&offset=2")
        assert len(resp2.json()) == 2

        resp3 = client.get("/api/events?limit=2&offset=4")
        assert len(resp3.json()) == 1


# --- Known Objects endpoints ---


class TestKnownObjectsAPI:
    def test_list_known_objects_empty(self, client):
        resp = client.get("/api/known-objects")
        assert resp.status_code == 200
        assert resp.json() == []

    def test_list_known_objects(self, client):
        _insert_known_object(name="Alice", type_="person")
        _insert_known_object(name="Rex", type_="pet")

        resp = client.get("/api/known-objects")
        data = resp.json()
        assert len(data) == 2
        # Should include embedding_count
        assert all("embedding_count" in obj for obj in data)

    def test_list_known_objects_filter_by_type(self, client):
        _insert_known_object(name="Alice", type_="person")
        _insert_known_object(name="Rex", type_="pet")

        resp = client.get("/api/known-objects?type=person")
        data = resp.json()
        assert len(data) == 1
        assert data[0]["name"] == "Alice"

    def test_get_known_object(self, client):
        obj_id = _insert_known_object(name="Charlie")

        resp = client.get(f"/api/known-objects/{obj_id}")
        assert resp.status_code == 200
        data = resp.json()
        assert data["name"] == "Charlie"
        assert "embeddings" in data
        assert data["embeddings"] == []

    def test_get_known_object_not_found(self, client):
        resp = client.get("/api/known-objects/99999")
        assert resp.status_code == 404

    def test_update_known_object(self, client):
        obj_id = _insert_known_object(name="OldName", type_="person", action="quiet")

        resp = client.put(
            f"/api/known-objects/{obj_id}",
            json={"name": "NewName", "action": "alert"},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["name"] == "NewName"
        assert data["action"] == "alert"
        assert data["type"] == "person"  # unchanged

    def test_update_known_object_not_found(self, client):
        resp = client.put(
            "/api/known-objects/99999",
            json={"name": "Nobody"},
        )
        assert resp.status_code == 404

    def test_update_known_object_no_fields(self, client):
        obj_id = _insert_known_object(name="NoChange")

        resp = client.put(
            f"/api/known-objects/{obj_id}",
            json={},
        )
        assert resp.status_code == 400

    def test_delete_known_object(self, client):
        obj_id = _insert_known_object(name="DeleteMe")

        resp = client.delete(f"/api/known-objects/{obj_id}")
        assert resp.status_code == 200

        # Verify deleted
        resp2 = client.get(f"/api/known-objects/{obj_id}")
        assert resp2.status_code == 404

    def test_delete_known_object_not_found(self, client):
        resp = client.delete("/api/known-objects/99999")
        assert resp.status_code == 404

    def test_get_embedding_thumbnail(self, client):
        obj_id = _insert_known_object(name="ThumbTest")
        thumbnail_data = b"\xff\xd8\xff\xe0thumb-jpeg"

        conn = get_connection()
        conn.execute(
            "INSERT INTO embeddings (known_object_id, embedding, thumbnail) VALUES (?, ?, ?)",
            (obj_id, b"\x00\x01", thumbnail_data),
        )
        conn.commit()
        emb_id = conn.execute(
            "SELECT id FROM embeddings WHERE known_object_id = ?", (obj_id,)
        ).fetchone()[0]

        resp = client.get(f"/api/known-objects/{obj_id}/thumbnails/{emb_id}")
        assert resp.status_code == 200
        assert resp.headers["content-type"] == "image/jpeg"
        assert resp.content == thumbnail_data

    def test_get_embedding_thumbnail_not_found(self, client):
        resp = client.get("/api/known-objects/99999/thumbnails/99999")
        assert resp.status_code == 404

    def test_known_objects_embedding_count(self, client):
        obj_id = _insert_known_object(name="CountTest")

        conn = get_connection()
        for i in range(3):
            conn.execute(
                "INSERT INTO embeddings (known_object_id, embedding) VALUES (?, ?)",
                (obj_id, b"\x00"),
            )
        conn.commit()

        resp = client.get("/api/known-objects")
        data = resp.json()
        obj = next(o for o in data if o["name"] == "CountTest")
        assert obj["embedding_count"] == 3


# --- Settings endpoints ---


class TestSettingsAPI:
    def test_get_settings_defaults(self, client):
        resp = client.get("/api/settings")
        assert resp.status_code == 200
        data = resp.json()
        # DB defaults
        assert data["vision_backend"] == "haiku"
        assert data["recognition_enabled"] == "false"
        assert data["event_retention_days"] == "90"
        # Runtime config values
        assert "recognition_threshold_high" in data
        assert "recognition_threshold_low" in data
        assert "rate_limit_seconds" in data
        assert "allowed_labels" in data
        assert "min_score" in data

    def test_update_settings(self, client):
        resp = client.put(
            "/api/settings",
            json={"vision_backend": "ollama", "event_retention_days": "30"},
        )
        assert resp.status_code == 200

        # Verify changes
        resp2 = client.get("/api/settings")
        data = resp2.json()
        assert data["vision_backend"] == "ollama"
        assert data["event_retention_days"] == "30"

    def test_update_settings_upsert_new_key(self, client):
        resp = client.put(
            "/api/settings",
            json={"custom_key": "custom_value"},
        )
        assert resp.status_code == 200

        resp2 = client.get("/api/settings")
        data = resp2.json()
        assert data["custom_key"] == "custom_value"


# --- Stats endpoint ---


class TestStatsAPI:
    def test_stats_returns_zeros(self, client):
        resp = client.get("/api/stats")
        assert resp.status_code == 200
        data = resp.json()
        assert data["total_events"] == 0
        assert data["known_matches"] == 0
        assert data["unknown_events"] == 0
        assert data["unreviewed"] == 0
        assert data["known_objects_count"] == 0
        assert data["events_today"] == 0

    def test_stats_counts_events(self, client):
        _insert_event(frigate_event_id="evt-stat-1", match_status="unreviewed")
        _insert_event(frigate_event_id="evt-stat-2", match_status="known")
        _insert_event(frigate_event_id="evt-stat-3", match_status="unknown")
        _insert_known_object(name="StatObj")

        resp = client.get("/api/stats")
        data = resp.json()
        assert data["total_events"] == 3
        assert data["known_matches"] == 1
        assert data["unknown_events"] == 1
        assert data["unreviewed"] == 1
        assert data["known_objects_count"] == 1
        assert data["events_today"] == 3  # All inserted today


# --- Full tag event flow ---


class TestTagEventFlow:
    def test_full_tag_flow(self, client):
        """Create event in DB, tag it with a new object, verify everything."""
        # 1. Create an event with an embedding
        fake_embedding = b"\x00\x01\x02\x03"
        event_id = _insert_event(
            frigate_event_id="evt-flow-1",
            camera="driveway",
            label="person",
            score=0.92,
            embedding=fake_embedding,
        )

        # 2. Verify it shows as unreviewed
        resp = client.get(f"/api/events/{event_id}")
        assert resp.json()["match_status"] == "unreviewed"

        # 3. Tag the event with a new person
        resp = client.post(
            f"/api/events/{event_id}/tag",
            json={"name": "FlowPerson", "type": "person"},
        )
        assert resp.status_code == 200
        new_obj_id = resp.json()["matched_object_id"]

        # 4. Verify event is now 'known'
        resp = client.get(f"/api/events/{event_id}")
        event_data = resp.json()
        assert event_data["match_status"] == "known"
        assert event_data["matched_object_id"] == new_obj_id
        assert event_data["matched_object_name"] == "FlowPerson"
        assert event_data["match_confidence"] == 1.0

        # 5. Verify known object was created with embedding
        resp = client.get(f"/api/known-objects/{new_obj_id}")
        obj_data = resp.json()
        assert obj_data["name"] == "FlowPerson"
        assert obj_data["type"] == "person"
        assert len(obj_data["embeddings"]) == 1

        # 6. Verify stats updated
        resp = client.get("/api/stats")
        stats = resp.json()
        assert stats["total_events"] == 1
        assert stats["known_matches"] == 1
        assert stats["known_objects_count"] == 1