enne2/pi-qmem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4776b4b4969286c49736192e3a4d8e54de0db157
pi-qmem/gateway/tests/test_api.py
T
Matteo Benedetto 4776b4b496 refactor: split gateway and pi-qmem extension into modules 
- gateway: separate config, models, state, audit, guardrail, embeddings,
  store, metrics, cleanup and routes; keep main.py as FastAPI bootstrap
- extension: split client/config, six tools, config command and rules;
  preserve jiti entrypoint and registrations
- Dockerfile copies the complete gateway module set
- tests: update monkeypatch boundaries for modular config/state
2026-08-24 16:19:19 +02:00

400 lines
14 KiB
Python
Raw Blame History

"""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):
import config
monkeypatch.setattr(config, "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
def test_supersede_root_ri_parenta_figli_attivi(client):
# L1 root + figlio L2
r1 = client.post(
"/v1/memories",
json=make_record(text="root L1", level="L1_ROOT", topic="TEST-TOPIC/ROOT"),
headers=auth_headers(),
)
root_id = r1.json()["memory_id"]
r2 = client.post(
"/v1/memories",
json=make_record(text="figlio L2", level="L2_SUBTOPIC", topic="TEST-TOPIC/SUB", parent_id=root_id),
headers=auth_headers(),
)
child_id = r2.json()["memory_id"]
# supersede il root
r3 = client.post(
"/v1/memories",
json=make_record(
text="root L1 corretto",
level="L1_ROOT",
topic="TEST-TOPIC/ROOT",
supersedes_id=root_id,
supersede_reason="aggiornamento",
),
headers=auth_headers(),
)
assert r3.status_code == 200
new_root_id = r3.json()["memory_id"]
assert r3.json()["reparented"] == 1
# il figlio attivo ora punta al nuovo root
child = client.get(f"/v1/memories/{child_id}", headers=auth_headers()).json()
assert child["parent_id"] == new_root_id
# search per parent_id sul nuovo root trova il figlio
s = client.post(
"/v1/memories:search",
json={"query": "*", "parent_id": new_root_id, "top_k": 10, "min_score": 0.0},
headers=auth_headers(),
)
ids = [x["memory_id"] for x in s.json()["results"]]
assert child_id in ids
def test_supersede_root_ri_parenta_solo_figli_attivi(client):
# L1 root + figlio L2 + figlio L2 già superseduto (versione attiva C1')
r1 = client.post(
"/v1/memories",
json=make_record(text="root", level="L1_ROOT", topic="T2/ROOT"),
headers=auth_headers(),
)
root_id = r1.json()["memory_id"]
c1 = client.post(
"/v1/memories",
json=make_record(text="figlio vecchio", level="L2_SUBTOPIC", topic="T2/SUB", parent_id=root_id),
headers=auth_headers(),
)
c1_id = c1.json()["memory_id"]
c1p = client.post(
"/v1/memories",
json=make_record(
text="figlio nuovo",
level="L2_SUBTOPIC",
topic="T2/SUB",
parent_id=root_id,
supersedes_id=c1_id,
),
headers=auth_headers(),
)
c1p_id = c1p.json()["memory_id"]
r2 = client.post(
"/v1/memories",
json=make_record(text="root corretto", level="L1_ROOT", topic="T2/ROOT", supersedes_id=root_id),
headers=auth_headers(),
)
new_root_id = r2.json()["memory_id"]
assert r2.json()["reparented"] == 1 # solo C1' (attivo)
# C1' ri-parentato al nuovo root; C1 storico resta ancorato al vecchio
assert client.get(f"/v1/memories/{c1p_id}", headers=auth_headers()).json()["parent_id"] == new_root_id
assert client.get(f"/v1/memories/{c1_id}", headers=auth_headers()).json()["parent_id"] == root_id
# ---------------------------------------------------------------------------
# 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
# ---------------------------------------------------------------------------
# Struttura Gerarchica e Relazionale
# ---------------------------------------------------------------------------
def test_create_and_retrieve_hierarchical_record(client):
"""Crea un nodo Root L1 e un nodo Figlio L2 con links, parent_id, level, topic."""
# 1. Crea Root L1
r_root = client.post(
"/v1/memories",
json=make_record(
text="Master Topic Alfa Romeo",
level="L1_ROOT",
topic="ALFA-ROMEO/ROOT",
),
headers=auth_headers(),
)
assert r_root.status_code == 200
root_id = r_root.json()["memory_id"]
# 2. Crea Figlio L2 collegato
r_child = client.post(
"/v1/memories",
json=make_record(
text="Scheda Tecnica Bialbero 1.3",
parent_id=root_id,
level="L2_SUBTOPIC",
topic="ALFA-ROMEO/SPECS",
links=[{"target_id": root_id, "predicate": "part_of", "weight": 1.0}],
),
headers=auth_headers(),
)
assert r_child.status_code == 200
child_id = r_child.json()["memory_id"]
# 3. Recupera e verifica payload strutturato
g = client.get(f"/v1/memories/{child_id}", headers=auth_headers())
assert g.status_code == 200
data = g.json()
assert data["parent_id"] == root_id
assert data["level"] == "L2_SUBTOPIC"
assert data["topic"] == "ALFA-ROMEO/SPECS"
assert len(data["links"]) == 1
assert data["links"][0]["target_id"] == root_id
def test_search_filters_hierarchical(client):
"""Filtra per parent_id, level e topic."""
r_root = client.post(
"/v1/memories",
json=make_record(text="Root doc", level="L1_ROOT", topic="TOPIC/ROOT"),
headers=auth_headers(),
)
root_id = r_root.json()["memory_id"]
client.post(
"/v1/memories",
json=make_record(text="Child A", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/A"),
headers=auth_headers(),
)
client.post(
"/v1/memories",
json=make_record(text="Child B", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/B"),
headers=auth_headers(),
)
# Cerca solo L1_ROOT
s1 = client.post(
"/v1/memories:search",
json={"query": "doc", "level": "L1_ROOT", "top_k": 5, "min_score": 0.0},
headers=auth_headers(),
)
assert s1.status_code == 200
assert len(s1.json()["results"]) == 1
assert s1.json()["results"][0]["level"] == "L1_ROOT"
# Cerca per parent_id
s2 = client.post(
"/v1/memories:search",
json={"query": "Child", "parent_id": root_id, "top_k": 5, "min_score": 0.0},
headers=auth_headers(),
)
assert s2.status_code == 200
assert len(s2.json()["results"]) == 2
# Cerca per topic specifico
s3 = client.post(
"/v1/memories:search",
json={"query": "Child", "topic": "TOPIC/A", "top_k": 5, "min_score": 0.0},
headers=auth_headers(),
)
assert s3.status_code == 200
assert len(s3.json()["results"]) == 1
assert s3.json()["results"][0]["topic"] == "TOPIC/A"
Reference in New Issue View Git Blame Copy Permalink