Blame for backend/tests/test_api.py

"""API contract tests — run with CEPHEUS_CLOUD=1 to avoid heavy ML imports."""
import os
import sys
from unittest.mock import MagicMock

os.environ.setdefault("CEPHEUS_CLOUD", "1")
os.environ.setdefault("CEPHEUS_API_KEY", "test-key")
os.environ.pop("CEPHEUS_AUTH_DEV_MODE", None)
os.environ.pop("CEPHEUS_JWT_SECRET", None)

# Stub Google ADK so tests run without the full agent stack installed.
for mod in (
    "google.adk",
    "google.adk.agents",
    "google.adk.runners",
    "google.adk.sessions",
):
    sys.modules.setdefault(mod, MagicMock())

BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if BACKEND_DIR not in sys.path:
    sys.path.insert(0, BACKEND_DIR)

from fastapi.testclient import TestClient  # noqa: E402

import main  # noqa: E402

API_HEADERS = {"X-API-Key": "test-key"}
client = TestClient(main.app)


def test_health():
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"


def test_health_live_wake():
    r = client.get("/health/live")
    assert r.status_code == 200
    body = r.json()
    assert body["status"] in ("warming", "ready")
    assert "message" in body


def test_gossip_start_accepts_payload():
    r = client.post(
        "/gossip/start",
        headers=API_HEADERS,
        json={"staffId": "STAFF-42", "personName": "Alex", "cause": "Suspicious contact"},
    )
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "started"
    assert body["tracking"]["staffId"] == "STAFF-42"
    assert body["tracking"]["personName"] == "Alex"
    assert body["tracking"]["cause"] == "Suspicious contact"
    assert body["root_person"] == "Alex"

    gossip = client.get("/gossip", headers=API_HEADERS).json()
    assert gossip["is_tracking"] is True
    assert gossip["tracking"]["staffId"] == "STAFF-42"
    assert gossip["root_person"] == "Alex"


def test_gossip_stop_clears_tracking_meta():
    client.post("/gossip/start", headers=API_HEADERS, json={"personName": "Pat"})
    r = client.post("/gossip/stop", headers=API_HEADERS)
    assert r.status_code == 200
    gossip = client.get("/gossip", headers=API_HEADERS).json()
    assert gossip["is_tracking"] is False
    assert gossip["tracking"] == {}


def test_sos_persists_to_get():
    before = len(client.get("/sos", headers=API_HEADERS).json())
    r = client.post(
        "/sos",
        headers=API_HEADERS,
        json={
            "guest_id": "guest-test-1",
            "lat": 12.97,
            "lng": 77.59,
            "location_label": "Gate A",
            "message": "Need help",
        },
    )
    assert r.status_code == 200
    assert r.json()["status"] == "received"
    after = client.get("/sos", headers=API_HEADERS).json()
    assert len(after) == before + 1
    assert after[-1]["guest_id"] == "guest-test-1"


def test_create_issue_defaults_title_and_broadcast_shape():
    r = client.post(
        "/issues",
        headers=API_HEADERS,
        json={"desc": "Smoke in corridor B", "type": "fire"},
    )
    assert r.status_code == 200
    issue = r.json()
    assert issue["title"] == "Smoke in corridor B"
    assert issue["status"] == "ONGOING"

    listed = client.get("/issues", headers=API_HEADERS).json()
    assert any(i["id"] == issue["id"] for i in listed)


def test_unauthorized_without_api_key():
    r = client.post("/issues", json={"title": "x", "desc": "y"})
    assert r.status_code == 401


def test_signage_toggle_sets_explicit_active_state():
    r = client.post(
        "/signage/s1/toggle",
        headers=API_HEADERS,
        json={"active": True},
    )
    assert r.status_code == 200
    body = r.json()
    assert body["id"] == "s1"
    assert body["active"] is True

    state = client.get("/signage", headers=API_HEADERS).json()
    assert state["s1"] is True


def test_files_upload_returns_normalized_path():
    png = (
        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01"
        b"\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\x00\x01"
        b"\x00\x00\x05\x00\x01\x0d\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
    )
    r = client.post(
        "/files/upload",
        headers=API_HEADERS,
        files={"file": ("tiny.png", png, "image/png")},
    )
    assert r.status_code == 200
    body = r.json()
    assert body["url"].startswith("/files/uploads/")
    assert body["path"] == body["url"]


def test_emergency_nearby_returns_services(monkeypatch):
    import types

    fake_payload = {
        "elements": [
            {"id": 1, "lat": 12.98, "lon": 77.60, "tags": {"amenity": "hospital", "name": "Test Hospital"}},
            {"id": 2, "lat": 12.96, "lon": 77.58, "tags": {"amenity": "police", "name": "Test PS"}},
            {"id": 3, "lat": 12.97, "lon": 77.59, "tags": {"amenity": "cafe", "name": "Ignore Me"}},
        ]
    }

    class FakeResp:
        def raise_for_status(self):
            return None

        def json(self):
            return fake_payload

    class FakeClient:
        def __init__(self, *a, **k):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def post(self, *a, **k):
            return FakeResp()

    monkeypatch.setattr(main, "httpx", types.SimpleNamespace(AsyncClient=FakeClient))
    main._nearby_cache.clear()

    r = client.get("/emergency/nearby?lat=12.9716&lng=77.5946", headers=API_HEADERS)
    assert r.status_code == 200
    body = r.json()
    assert body["source"] == "overpass"
    assert body["services"]["hospital"][0]["name"] == "Test Hospital"
    assert body["services"]["police"][0]["name"] == "Test PS"
    assert "cafe" not in body["services"]
    assert isinstance(body["services"]["hospital"][0]["distKm"], (int, float))


def test_gossip_ingest_frame_co_presence_only(monkeypatch):
    """Single-person frames must NOT create interaction edges; pairs in one frame do."""
    class FakeEngine:
        def reload_db(self):
            return None

        def match_all_faces(self, frame, threshold=None):
            return [
                {"name": "ContactA", "confidence": 0.9, "bbox": [0, 0, 1, 1], "found": True},
                {"name": "ContactB", "confidence": 0.85, "bbox": [2, 2, 3, 3], "found": True},
            ]

    monkeypatch.setattr(main.vision_engine, "face_engine", FakeEngine())
    client.post("/gossip/start", headers=API_HEADERS, json={"personName": "RootGuy"})

    import cv2
    import numpy as np

    frame = np.zeros((16, 16, 3), dtype=np.uint8)
    ok, buf = cv2.imencode(".jpg", frame)
    assert ok
    jpeg_bytes = buf.tobytes()

    r = client.post(
        "/gossip/ingest_frame",
        headers=API_HEADERS,
        data={"cam_id": "cam-01"},
        files={"file": ("frame.jpg", jpeg_bytes, "image/jpeg")},
    )
    assert r.status_code == 200
    body = r.json()
    assert "ContactA" in body["names"]
    assert "ContactB" in body["names"]
    pairs = {(l["source"], l["target"]) for l in body["graph"]["links"]}
    assert ("ContactA", "ContactB") in pairs or ("ContactB", "ContactA") in pairs


def test_tracking_session_reset_clears_history():
    client.post("/gossip/start", headers=API_HEADERS, json={"personName": "ResetTest", "staffId": "S1"})
    r = client.post("/tracking/session/reset", headers=API_HEADERS, json={"broadcast": False})
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "reset"
    assert body.get("broadcast") is False
    gossip = client.get("/gossip", headers=API_HEADERS).json()
    assert gossip.get("total_interactions", len(gossip.get("links", []))) == 0 or len(gossip.get("links", [])) == 0
    assert gossip.get("tracking") == {}