tests/test_api.py

"""FastAPI surface: health, model info, metrics, validation, predict, feedback, personalise."""

from __future__ import annotations

import io

import numpy as np
import pytest

import config


# --------------------------------------------------------------------------
# health / metadata
# --------------------------------------------------------------------------
def test_health(api_client):
    r = api_client.get("/health")
    assert r.status_code == 200
    body = r.json()
    assert body["status"] in ("ok", "degraded")
    assert "model_loaded" in body and "uptime_s" in body
    assert "num_classes" in body, "health reports the class count under an unambiguous name"
    if body["model_loaded"]:
        assert body["num_classes"] == config.NUM_CLASSES


def test_live_session_starts_with_dynamic_time_axis_model(api_client):
    """The saved models declare a dynamic time axis but live sends 30 frames."""
    response = api_client.post("/live/start", json={
        "model": config.DEFAULT_MODEL, "threshold": 0.70, "smoothing_window": 3,
    })
    assert response.status_code == 200, response.text
    session_id = response.json()["session_id"]
    assert response.json()["sequence_length"] == config.SEQUENCE_LENGTH
    stopped = api_client.post("/live/stop", json={"session_id": session_id})
    assert stopped.status_code == 200


def test_root_serves_live_camera_dashboard(api_client):
    page = api_client.get("/")
    script = api_client.get("/static/app.js")
    assert page.status_code == 200 and 'id="startBtn"' in page.text
    assert script.status_code == 200 and "startLiveLoop" in script.text


def test_model_info(api_client):
    r = api_client.get("/model-info")
    assert r.status_code == 200
    body = r.json()
    assert body["num_classes"] == config.NUM_CLASSES
    assert len(body["classes"]) == config.NUM_CLASSES
    assert body["sequence_length"] == config.SEQUENCE_LENGTH
    assert body["split_protocol"] == config.SPLIT_PROTOCOL
    assert body["default_model"] == config.DEFAULT_MODEL
    assert "window" in body["smoothing"] and "speech" in body


def test_classes_endpoint_matches_vocabulary(api_client):
    body = api_client.get("/classes").json()
    assert body["num_classes"] == 50 and len(body["classes"]) == 50
    assert "INCLUDE" in body["source"].upper()


def test_config_endpoint_exposes_no_secrets(api_client):
    body = api_client.get("/config").json()
    text = str(body).lower()
    for secret in ("api_key", "password", "token", "secret"):
        assert secret not in text
    assert body["privacy"]["persist_uploads"] in (True, False)


def test_metrics_endpoint_shape(api_client):
    body = api_client.get("/metrics").json()
    for key in ("predictions_total", "feedback_total", "status_counts", "latency_ms",
                "uncertain_rate", "human_actions", "corrections_total"):
        assert key in body


# --------------------------------------------------------------------------
# validation
# --------------------------------------------------------------------------
def test_predict_rejects_empty_body(api_client):
    r = api_client.post("/predict", json={})
    assert r.status_code == 400


def test_predict_rejects_wrong_feature_width(api_client):
    r = api_client.post("/predict", json={"features": [[0.0] * 10] * 30})
    assert r.status_code == 422
    assert str(config.MODEL_INPUT_DIM) in r.text


def test_predict_rejects_too_few_frames(api_client):
    r = api_client.post("/predict", json={"features": [[0.0] * config.MODEL_INPUT_DIM] * 2})
    assert r.status_code == 422


def test_predict_rejects_both_sources(api_client):
    payload = {"features": [[0.0] * config.MODEL_INPUT_DIM] * 30, "dataset_clip": "x/y.MOV"}
    assert api_client.post("/predict", json=payload).status_code == 422


def test_predict_rejects_path_traversal(api_client):
    r = api_client.post("/predict", json={"dataset_clip": "../../etc/passwd"})
    assert r.status_code == 400


def test_predict_rejects_unknown_clip(api_client):
    r = api_client.post("/predict", json={"dataset_clip": "Nope/1. Fake/MVI_0001.MOV"})
    assert r.status_code == 404


def test_predict_rejects_bad_model_name(api_client):
    r = api_client.post("/predict", json={"features": [[0.0] * config.MODEL_INPUT_DIM] * 30,
                                          "model": "resnet"})
    assert r.status_code == 422


def test_upload_rejects_disallowed_extension(api_client):
    r = api_client.post("/predict", files={"file": ("evil.sh", b"#!/bin/sh\n", "text/plain")})
    assert r.status_code == 415


def test_upload_rejects_fake_video_content(api_client):
    r = api_client.post("/predict", files={"file": ("fake.mp4", b"not a video at all", "video/mp4")})
    assert r.status_code == 415


def test_upload_rejects_oversized_file(api_client, monkeypatch):
    monkeypatch.setattr(config, "MAX_UPLOAD_MB", 0)
    data = b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 64
    r = api_client.post("/predict", files={"file": ("big.mp4", data, "video/mp4")})
    assert r.status_code == 413


def test_feedback_validation(api_client):
    assert api_client.post("/feedback", json={"action": "MAYBE", "predicted_class": "HELLO"}).status_code == 422
    assert api_client.post("/feedback", json={"action": "CORRECT", "predicted_class": "HELLO"}).status_code == 422
    ok = api_client.post("/feedback", json={"action": "ACCEPT", "predicted_class": "HELLO",
                                            "confidence": 0.9, "model": "gru_mha"})
    assert ok.status_code == 200 and ok.json()["status"] == "recorded"


def test_personalize_validation(api_client):
    bad_vocab = api_client.post("/personalize", json={
        "profile": "t1", "action": "create", "vocabulary": ["NOT_A_REAL_SIGN"]})
    assert bad_vocab.status_code == 422
    bad_name = api_client.post("/personalize", json={"profile": "../etc", "action": "create"})
    assert bad_name.status_code == 422
    bad_preset = api_client.post("/personalize", json={"profile": "t1", "action": "create",
                                                       "preset": "made_up"})
    assert bad_preset.status_code == 422
    calibrate_without_data = api_client.post("/personalize", json={"profile": "t1",
                                                                   "action": "calibrate"})
    assert calibrate_without_data.status_code == 422


# --------------------------------------------------------------------------
# real predictions (skipped when no trained model is on disk)
# --------------------------------------------------------------------------
@pytest.fixture()
def require_model(api_client):
    info = api_client.get("/health").json()
    if not info["model_loaded"]:
        pytest.skip(f"no trained model: {info.get('reason')}")
    return info


def test_predict_from_dataset_clip(api_client, require_model):
    classes = api_client.get("/classes").json()["classes"]
    clip = api_client.get("/dataset/clips", params={"split": "test", "limit": 1}).json()[0]
    r = api_client.post("/predict", json={"dataset_clip": clip["path"], "log": False})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["sign"] in classes or body["sign"] == config.UNCERTAIN_LABEL
    assert 0.0 <= body["confidence"] <= 1.0
    assert body["status"] in ("CONFIDENT", "UNCERTAIN", "PENDING")
    assert body["latency_ms"] > 0
    assert body["model"] == config.DEFAULT_MODEL
    assert len(body["top_k"]) == 5
    probs = [t["probability"] for t in body["top_k"]]
    assert probs == sorted(probs, reverse=True)


def test_predict_logs_an_event(api_client, require_model, temp_logs):
    from backend.inference import audit

    clip = api_client.get("/dataset/clips", params={"split": "test", "limit": 1}).json()[0]
    body = api_client.post("/predict", json={"dataset_clip": clip["path"], "log": True}).json()
    assert body["event_id"]
    rows = audit.read_events(limit=5)
    assert rows and rows[0]["event_id"] == body["event_id"]
    assert rows[0]["source"] == "dataset_clip"


def test_feedback_links_to_prediction_event(api_client, require_model, temp_logs):
    from backend.inference import audit

    clips = api_client.get("/dataset/clips", params={"split": "test", "limit": 2}).json()
    pred = api_client.post("/predict", json={"dataset_clip": clips[0]["path"]}).json()
    api_client.post("/feedback", json={"action": "CORRECT", "predicted_class": pred["sign"],
                                       "corrected_class": clips[1]["label"],
                                       "event_id": pred["event_id"],
                                       "confidence": pred["confidence"], "model": pred["model"]})
    rows = audit.read_events(only_corrected=True)
    assert rows and rows[0]["corrected_class"] == clips[1]["label"]
    pairs = api_client.get("/audit/pairs").json()
    assert pairs and pairs[0]["predicted"] == pred["sign"]


def test_predict_threshold_override_is_reported(api_client, require_model):
    clip = api_client.get("/dataset/clips", params={"split": "test", "limit": 1}).json()[0]
    body = api_client.post("/predict", json={"dataset_clip": clip["path"], "threshold": 0.3,
                                             "log": False}).json()
    assert body["threshold"] == 0.3


def test_live_camera_frame_opens_mediapipe_session(api_client, require_model):
    import cv2

    started = api_client.post("/live/start", json={"model": config.DEFAULT_MODEL})
    assert started.status_code == 200, started.text
    session_id = started.json()["session_id"]
    ok, encoded = cv2.imencode(".jpg", np.zeros((360, 640, 3), dtype=np.uint8))
    assert ok
    try:
        response = api_client.post("/live/frame", data={
            "session_id": session_id, "timestamp_ms": "100",
        }, files={"file": ("camera-frame.jpg", encoded.tobytes(), "image/jpeg")})
        assert response.status_code == 200, response.text
        body = response.json()
        assert body["ready"] is False
        assert len(body["landmarks"]) == config.NUM_LANDMARKS
        assert body["processing_ms"] > 0
        assert body["landmark_state"] == {"pose": False, "left_hand": False, "right_hand": False}
    finally:
        assert api_client.post("/live/stop", json={"session_id": session_id}).status_code == 200


def test_zero_threshold_is_preserved_by_engine_cache(api_client, require_model):
    from backend import main as api

    api.clear_engine_cache()
    engine = api.get_engine(threshold=0.0)
    assert engine.threshold == 0.0


def test_audit_events_filters_via_api(api_client, require_model, temp_logs):
    clip = api_client.get("/dataset/clips", params={"split": "test", "limit": 1}).json()[0]
    api_client.post("/predict", json={"dataset_clip": clip["path"], "log": True})
    assert len(api_client.get("/audit/events", params={"limit": 10}).json()) >= 1
    assert api_client.get("/audit/events", params={"only_rejected": True}).json() == []