feat: add REST API endpoints for events, known objects, and settings
Routes package with CRUD endpoints: event listing/filtering/tagging/dismissal, known object management with thumbnails, settings CRUD, and stats. 64 tests passing. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,619 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user