pi-qmem/gateway/tests/test_api.py

"""Test API del Memory Gateway: validazione, auth, idempotency, supersede, ricerca."""

import pytest
from conftest import auth_headers, make_record


# ---------------------------------------------------------------------------
# Validazione input
# ---------------------------------------------------------------------------
def test_project_id_obbligatorio(client):
    body = make_record()
    del body["project_id"]
    r = client.post("/v1/memories", json=body, headers=auth_headers())
    assert r.status_code == 422


def test_kind_invalido(client):
    r = client.post("/v1/memories", json=make_record(kind="boh"), headers=auth_headers())
    assert r.status_code == 422


def test_expires_at_invalido(client):
    r = client.post("/v1/memories", json=make_record(expires_at="non-una-data"), headers=auth_headers())
    assert r.status_code == 422
    assert "ISO 8601" in r.text


def test_expires_at_valido(client):
    r = client.post("/v1/memories", json=make_record(expires_at="2026-09-01T00:00:00Z"), headers=auth_headers())
    assert r.status_code == 200


def test_confidence_invalido(client):
    r = client.post("/v1/memories", json=make_record(confidence="super"), headers=auth_headers())
    assert r.status_code == 422


def test_confidence_default_medium(client):
    r = client.post("/v1/memories", json=make_record(), headers=auth_headers())
    assert r.status_code == 200
    mid = r.json()["memory_id"]
    g = client.get(f"/v1/memories/{mid}", headers=auth_headers())
    assert g.json()["confidence"] == "medium"


def test_text_troppo_lungo(client):
    r = client.post("/v1/memories", json=make_record(text="x" * 9000), headers=auth_headers())
    assert r.status_code == 422


# ---------------------------------------------------------------------------
# Auth e rate limit
# ---------------------------------------------------------------------------
def test_senza_chiave_422(client):
    # Header X-API-Key mancante → 422 (header richiesto da FastAPI)
    r = client.post("/v1/memories", json=make_record())
    assert r.status_code == 422


def test_chiave_invalida_401(client):
    r = client.post("/v1/memories", json=make_record(), headers={"X-API-Key": "sbagliata"})
    assert r.status_code == 401


def test_rate_limit_429(client, monkeypatch):
    monkeypatch.setattr("main.RATE_LIMIT_PER_MIN", 3)
    for _ in range(3):
        r = client.post("/v1/memories", json=make_record(), headers=auth_headers())
        assert r.status_code == 200
    r = client.post("/v1/memories", json=make_record(), headers=auth_headers())
    assert r.status_code == 429


# ---------------------------------------------------------------------------
# Idempotency
# ---------------------------------------------------------------------------
def test_idempotency_replay_stessa_risposta(client):
    h = {**auth_headers(), "Idempotency-Key": "k-1"}
    r1 = client.post("/v1/memories", json=make_record(), headers=h)
    r2 = client.post("/v1/memories", json=make_record(), headers=h)
    assert r1.status_code == 200 and r2.status_code == 200
    assert r1.json()["memory_id"] == r2.json()["memory_id"]
    assert client.fake_qdrant.upsert_calls == 1


def test_idempotency_payload_diverso_409(client):
    h = {**auth_headers(), "Idempotency-Key": "k-2"}
    client.post("/v1/memories", json=make_record(), headers=h)
    r = client.post("/v1/memories", json=make_record(text="diverso"), headers=h)
    assert r.status_code == 409


def test_idempotency_key_diverse_record_distinti(client):
    r1 = client.post("/v1/memories", json=make_record(), headers={**auth_headers(), "Idempotency-Key": "k-a"})
    r2 = client.post("/v1/memories", json=make_record(), headers={**auth_headers(), "Idempotency-Key": "k-b"})
    assert r1.json()["memory_id"] != r2.json()["memory_id"]


# ---------------------------------------------------------------------------
# Supersede
# ---------------------------------------------------------------------------
def test_supersede_target_inesistente_404(client):
    r = client.post(
        "/v1/memories",
        json=make_record(supersedes_id="00000000-0000-0000-0000-000000000000"),
        headers=auth_headers(),
    )
    assert r.status_code == 404


def test_supersede_ok_e_lineage(client):
    r1 = client.post("/v1/memories", json=make_record(text="fatto falso"), headers=auth_headers())
    old_id = r1.json()["memory_id"]
    r2 = client.post(
        "/v1/memories",
        json=make_record(text="fatto corretto", supersedes_id=old_id, supersede_reason="evidenza"),
        headers=auth_headers(),
    )
    assert r2.status_code == 200
    new_id = r2.json()["memory_id"]
    # il vecchio è marcato superseded_by
    old = client.get(f"/v1/memories/{old_id}", headers=auth_headers()).json()
    assert old["superseded_by"] == new_id
    # la ricerca di default esclude i superseduti
    s = client.post("/v1/memories:search", json={"query": "fatto", "top_k": 10, "min_score": 0.0}, headers=auth_headers())
    ids = [x["memory_id"] for x in s.json()["results"]]
    assert old_id not in ids
    # include_superseded li mostra
    s2 = client.post(
        "/v1/memories:search",
        json={"query": "fatto", "top_k": 10, "min_score": 0.0, "include_superseded": True},
        headers=auth_headers(),
    )
    ids2 = [x["memory_id"] for x in s2.json()["results"]]
    assert old_id in ids2


def test_supersede_doppio_409(client):
    r1 = client.post("/v1/memories", json=make_record(text="falso"), headers=auth_headers())
    old_id = r1.json()["memory_id"]
    client.post("/v1/memories", json=make_record(text="corretto", supersedes_id=old_id), headers=auth_headers())
    r = client.post("/v1/memories", json=make_record(text="ancora", supersedes_id=old_id), headers=auth_headers())
    assert r.status_code == 409


# ---------------------------------------------------------------------------
# Ricerca e filtri
# ---------------------------------------------------------------------------
def test_search_filtro_project_id(client):
    client.post("/v1/memories", json=make_record(text="uno", project_id="proj-a"), headers=auth_headers())
    client.post("/v1/memories", json=make_record(text="due", project_id="proj-b"), headers=auth_headers())
    s = client.post(
        "/v1/memories:search",
        json={"query": "test", "project_id": "proj-a", "top_k": 10, "min_score": 0.0},
        headers=auth_headers(),
    )
    results = s.json()["results"]
    assert len(results) == 1
    assert results[0]["project_id"] == "proj-a"


def test_search_hybrid_param_accettato(client):
    client.post("/v1/memories", json=make_record(text="codice XYZ-123"), headers=auth_headers())
    s = client.post(
        "/v1/memories:search",
        json={"query": "XYZ-123", "top_k": 5, "min_score": 0.0, "hybrid": True},
        headers=auth_headers(),
    )
    assert s.status_code == 200
    assert "results" in s.json()


def test_meta_overview(client):
    client.post("/v1/memories", json=make_record(project_id="proj-a"), headers=auth_headers())
    r = client.get("/v1/meta/overview", headers=auth_headers())
    assert r.status_code == 200
    data = r.json()
    assert data["total"] >= 1
    assert any(p["project_id"] == "proj-a" for p in data["projects"])


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


def test_metrics_auth(client):
    # Header mancante → 422; chiave invalida → 401; chiave valida → 200
    assert client.get("/v1/metrics").status_code == 422
    assert client.get("/v1/metrics", headers={"X-API-Key": "sbagliata"}).status_code == 401
    r2 = client.get("/v1/metrics", headers=auth_headers())
    assert r2.status_code == 200
    assert "requests" in r2.json()


# ---------------------------------------------------------------------------
# Versione / metadata del codice
# ---------------------------------------------------------------------------
def test_version_endpoint_pubblico(client):
    """GET /v1/version è pubblico e espone git_commit e guardrail_version."""
    r = client.get("/v1/version")
    assert r.status_code == 200
    data = r.json()
    assert "git_commit" in data
    assert "version" in data
    assert "guardrail_version" in data
    assert data["guardrail_version"] == "similarity-v1"


def test_status_espone_git_commit(client):
    """/v1/status include version, git_commit e guardrail_version."""
    r = client.get("/v1/status")
    assert r.status_code == 200
    data = r.json()
    assert "git_commit" in data
    assert "version" in data
    assert "guardrail_version" in data