Matteo Benedetto
4776b4b496
- 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
112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
"""Test del guardrail di similarità pre-scrittura (Memory Gateway).
|
|
|
|
Casi: duplicato esatto (hash) -> BLOCK 409; similarità alta -> BLOCK 409;
|
|
similarità moderata -> WARN (salva con flag); nessun candidato -> ALLOW;
|
|
supersede esplicito bypassa il guardrail.
|
|
"""
|
|
|
|
import pytest
|
|
from conftest import auth_headers, make_record
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def enable_guardrail(monkeypatch):
|
|
"""Abilita il guardrail per i test di questa suite (il conftest lo disabilita di default)."""
|
|
import config
|
|
|
|
monkeypatch.setattr(config, "GUARDRAIL_ENABLED", True)
|
|
|
|
|
|
def test_duplicato_esatto_bloccato_409(client):
|
|
"""Stesso testo normalizzato -> hash uguale -> BLOCK (409)."""
|
|
body = make_record(text="Il cliente non accede al portale COSMO-SkyMed")
|
|
r1 = client.post("/v1/memories", json=body, headers=auth_headers())
|
|
assert r1.status_code == 200
|
|
|
|
# Stesso testo con maiuscole/spazi diversi -> stesso hash normalizzato
|
|
body2 = make_record(text=" IL CLIENTE NON ACCEDE al portale COSMO-SkyMed ")
|
|
r2 = client.post("/v1/memories", json=body2, headers=auth_headers())
|
|
assert r2.status_code == 409
|
|
detail = r2.json()["detail"]
|
|
assert detail["error"] == "duplicate_memory"
|
|
assert detail["reason"] == "EXACT_DUPLICATE"
|
|
|
|
|
|
def test_similarita_alta_bloccata_409(client):
|
|
"""Score top-1 >= soglia BLOCK (0.85) -> 409 KNOWN_SOLUTION."""
|
|
client.fake_qdrant.query_score = 0.9
|
|
r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers())
|
|
assert r1.status_code == 200
|
|
|
|
r2 = client.post("/v1/memories", json=make_record(text="secondo record simile"), headers=auth_headers())
|
|
assert r2.status_code == 409
|
|
assert r2.json()["detail"]["reason"] == "KNOWN_SOLUTION"
|
|
|
|
|
|
def test_similarita_moderata_warn_salva(client):
|
|
"""Score top-1 tra 0.70 e 0.85 -> WARN: salva con flag guardrail."""
|
|
client.fake_qdrant.query_score = 0.75
|
|
r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers())
|
|
assert r1.status_code == 200
|
|
|
|
r2 = client.post("/v1/memories", json=make_record(text="secondo record simile"), headers=auth_headers())
|
|
assert r2.status_code == 200
|
|
memory_id = r2.json()["memory_id"]
|
|
# Il record salvato deve avere il flag guardrail WARN
|
|
saved = client.fake_qdrant.points[memory_id].payload
|
|
assert saved["guardrail"]["decision"] == "WARN"
|
|
assert saved["guardrail"]["reason"] == "MODERATE_SIMILARITY"
|
|
|
|
|
|
def test_nessun_candidato_allow(client):
|
|
"""Score top-1 sotto soglia WARN -> ALLOW, nessun flag guardrail."""
|
|
client.fake_qdrant.query_score = 0.5
|
|
r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers())
|
|
assert r1.status_code == 200
|
|
|
|
r2 = client.post("/v1/memories", json=make_record(text="secondo record"), headers=auth_headers())
|
|
assert r2.status_code == 200
|
|
memory_id = r2.json()["memory_id"]
|
|
saved = client.fake_qdrant.points[memory_id].payload
|
|
assert saved["guardrail"]["decision"] == "ALLOW"
|
|
assert saved["guardrail"]["reason"] == "NEW_SOLUTION"
|
|
|
|
|
|
def test_supersede_bypassa_guardrail(client):
|
|
"""Il supersede esplicito è una correzione intenzionale: bypassa il guardrail."""
|
|
client.fake_qdrant.query_score = 0.9
|
|
r1 = client.post("/v1/memories", json=make_record(text="record originale"), headers=auth_headers())
|
|
assert r1.status_code == 200
|
|
old_id = r1.json()["memory_id"]
|
|
|
|
# Supersede con testo molto simile -> deve passare (correzione intenzionale)
|
|
r2 = client.post(
|
|
"/v1/memories",
|
|
json=make_record(text="record originale corretto", supersedes_id=old_id, supersede_reason="correzione"),
|
|
headers=auth_headers(),
|
|
)
|
|
assert r2.status_code == 200
|
|
assert r2.json()["supersedes_id"] == old_id
|
|
|
|
|
|
def test_guardrail_disabilitato_salva_sempre(client, monkeypatch):
|
|
"""Con GUARDRAIL_ENABLED=false non si blocca nulla."""
|
|
import config
|
|
|
|
monkeypatch.setattr(config, "GUARDRAIL_ENABLED", False)
|
|
client.fake_qdrant.query_score = 0.9
|
|
r1 = client.post("/v1/memories", json=make_record(text="primo"), headers=auth_headers())
|
|
assert r1.status_code == 200
|
|
r2 = client.post("/v1/memories", json=make_record(text="primo"), headers=auth_headers())
|
|
assert r2.status_code == 200
|
|
|
|
|
|
def test_text_hash_salvato_nel_payload(client):
|
|
"""Ogni record salvato deve avere text_hash (per lo strato 1 del guardrail)."""
|
|
r = client.post("/v1/memories", json=make_record(text="record con hash"), headers=auth_headers())
|
|
assert r.status_code == 200
|
|
memory_id = r.json()["memory_id"]
|
|
saved = client.fake_qdrant.points[memory_id].payload
|
|
assert "text_hash" in saved
|
|
assert len(saved["text_hash"]) == 64 # SHA-256 hex
|