Matteo Benedetto
eccb2cb870
Guardrail deterministico FUORI dall'LLM (stessa architettura di egeos-copilot): - Strato 1: text_hash SHA-256 normalizzato -> BLOCK 409 (EXACT_DUPLICATE) - Strato 2: similarità semantica top-3 BGE-M3 cosine -> BLOCK/WARN/ALLOW (soglie configurabili: GUARDRAIL_BLOCK_THRESHOLD 0.85, WARN 0.70) - Supersede esplicito bypassa il guardrail (correzione intenzionale) - text_hash e flag guardrail nel payload; audit create_blocked - Indice payload su text_hash - Test: 6 nuovi (duplicato esatto, similarità alta/moderata, nessun candidato, supersede bypass, disabilitato, text_hash) — 28/28 passano
137 lines
4.3 KiB
Python
137 lines
4.3 KiB
Python
"""Fixtures pytest per il Memory Gateway: FakeQdrant in-memory + mock di embed.
|
|
|
|
I test non richiedono Qdrant né Ollama: il gateway viene importato con le
|
|
dipendenze reali (fastapi/pydantic/qdrant-client/httpx) ma qdrant e embed
|
|
sono sostituiti da fake deterministici.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Disabilita il push metriche nei test
|
|
os.environ["METRICS_ENABLED"] = "false"
|
|
os.environ["API_KEYS"] = "test-key"
|
|
# Guardrail disabilitato di default nei test esistenti (abilitato nei test del guardrail)
|
|
os.environ["GUARDRAIL_ENABLED"] = "false"
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import pytest # noqa: E402
|
|
import main as gateway # noqa: E402
|
|
|
|
|
|
class FakePoint:
|
|
def __init__(self, point_id, vector=None, payload=None):
|
|
self.id = point_id
|
|
self.vector = vector or {}
|
|
self.payload = payload or {}
|
|
|
|
|
|
class FakeQdrant:
|
|
"""Implementazione in-memory dei metodi Qdrant usati dal gateway."""
|
|
|
|
def __init__(self):
|
|
self.points: dict[str, FakePoint] = {}
|
|
self.collection_exists = False
|
|
self.upsert_calls = 0
|
|
self.query_score = 0.9 # score di default per query_points (configurabile nei test)
|
|
|
|
def get_collections(self):
|
|
class _C:
|
|
def __init__(self, names):
|
|
self.collections = [type("X", (), {"name": n})() for n in names]
|
|
|
|
return _C(["memories"] if self.collection_exists else [])
|
|
|
|
def create_collection(self, **kw):
|
|
self.collection_exists = True
|
|
|
|
def create_payload_index(self, **kw):
|
|
pass
|
|
|
|
def create_vector_name(self, *a, **kw):
|
|
pass
|
|
|
|
def upsert(self, collection_name, points, **kw):
|
|
self.upsert_calls += 1
|
|
for p in points:
|
|
self.points[p.id] = FakePoint(p.id, p.vector, p.payload)
|
|
|
|
def retrieve(self, collection_name, ids, with_payload=True):
|
|
return [self.points[i] for i in ids if i in self.points]
|
|
|
|
def set_payload(self, collection_name, payload, points, **kw):
|
|
for pid in points:
|
|
if pid in self.points:
|
|
self.points[pid].payload.update(payload)
|
|
|
|
def delete(self, collection_name, points_selector, **kw):
|
|
for pid in points_selector:
|
|
self.points.pop(pid, None)
|
|
|
|
def scroll(self, collection_name, **kw):
|
|
return list(self.points.values()), None
|
|
|
|
def get_collection(self, collection_name):
|
|
class _Info:
|
|
points_count = len(self.points)
|
|
|
|
return _Info()
|
|
|
|
def query_points(self, collection_name, query=None, query_filter=None, limit=5,
|
|
score_threshold=None, with_payload=True, prefetch=None, **kw):
|
|
# Ritorna tutti i punti (score fisso); i filtri metadata sono applicati
|
|
# in modo semplice per testare kind/project_id/scope/superseded.
|
|
results = []
|
|
for p in self.points.values():
|
|
pl = p.payload
|
|
if query_filter and query_filter.must:
|
|
ok = True
|
|
for cond in query_filter.must:
|
|
if hasattr(cond, "key") and hasattr(cond, "match"):
|
|
if pl.get(cond.key) != cond.match.value:
|
|
ok = False
|
|
elif hasattr(cond, "is_empty"):
|
|
if pl.get(cond.is_empty.key):
|
|
ok = False
|
|
if not ok:
|
|
continue
|
|
results.append(type("H", (), {"id": p.id, "score": self.query_score, "payload": pl})())
|
|
return type("R", (), {"points": results[:limit]})()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
"""TestClient con qdrant e embed finti."""
|
|
fake = FakeQdrant()
|
|
monkeypatch.setattr(gateway, "qdrant", fake)
|
|
monkeypatch.setattr(gateway, "API_KEYS", {"test-key"})
|
|
monkeypatch.setattr(gateway, "_ratelimit", {}) # rate limit pulito per test
|
|
|
|
async def fake_embed(text):
|
|
return [0.0] * 1024
|
|
|
|
monkeypatch.setattr(gateway, "embed", fake_embed)
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
with TestClient(gateway.app) as c:
|
|
c.fake_qdrant = fake
|
|
yield c
|
|
|
|
|
|
def auth_headers():
|
|
return {"X-API-Key": "test-key"}
|
|
|
|
|
|
def make_record(**overrides):
|
|
base = {
|
|
"text": "record di test",
|
|
"kind": "fact",
|
|
"project_id": "test-proj",
|
|
"scope": "agent",
|
|
}
|
|
base.update(overrides)
|
|
return base
|