84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Verifica P7: RAG Irrigazione + Turni."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, "/srv" if os.path.isdir("/srv/app") else os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
|
|
def main() -> int:
|
|
from app.db import init_db
|
|
from app.knowledge import apps_indexer, indexer
|
|
from app.mcp import tools
|
|
|
|
init_db()
|
|
user = os.environ.get("MCP_TEST_USER", "daniele")
|
|
claims = {
|
|
"sub": user,
|
|
"scope": "knowledge:read irrigation:read turni:read admin",
|
|
}
|
|
errors = []
|
|
|
|
print("=== P7 apps RAG ===")
|
|
existing = apps_indexer.list_indexed_records(limit=5)
|
|
if len(existing) >= 2:
|
|
print(f"skip re-index: {len(existing)} records already present")
|
|
result = {"irrigazione": {"indexed": len(existing), "skipped_reindex": True}, "turni": {}}
|
|
else:
|
|
result = {
|
|
"irrigazione": apps_indexer.index_irrigazione(history_limit=0, events_limit=0),
|
|
"turni": apps_indexer.index_turni(assignments_limit=3),
|
|
}
|
|
print("index_all:", json.dumps(result, ensure_ascii=False)[:500])
|
|
|
|
irr = result.get("irrigazione", {})
|
|
turn = result.get("turni", {})
|
|
if irr.get("skipped") and turn.get("skipped"):
|
|
print("P7 FAILED: both sources skipped")
|
|
return 1
|
|
if irr.get("indexed", 0) + turn.get("indexed", 0) < 2 and not irr.get("skipped_reindex"):
|
|
errors.append("insufficient-indexed-records")
|
|
|
|
records = apps_indexer.list_indexed_records(limit=10)
|
|
print(f"indexed records: {len(records)}")
|
|
if not records:
|
|
errors.append("no-records")
|
|
|
|
try:
|
|
hits = apps_indexer.search_apps_knowledge("irrigazione zona prato", limit=3)
|
|
print(f"search_apps_knowledge irrigazione: {len(hits)} hits")
|
|
if not hits:
|
|
errors.append("search-irrigazione-empty")
|
|
except Exception as exc:
|
|
print(f"FAIL search_apps: {exc}")
|
|
errors.append("search_apps")
|
|
|
|
try:
|
|
out = tools.call_tool("search_apps_knowledge", {"query": "turno guardia", "limit": 3}, claims)
|
|
data = json.loads(out["content"][0]["text"])
|
|
print(f"OK tool search_apps_knowledge: {len(data.get('results', []))} hits")
|
|
except Exception as exc:
|
|
print(f"FAIL tool search_apps_knowledge: {exc}")
|
|
errors.append("tool-search")
|
|
|
|
try:
|
|
combined = indexer.search_knowledge(user, "irrigazione valvola", limit=5, is_admin=True)
|
|
apps_hits = [h for h in combined if h.get("source") in ("irrigazione", "turni")]
|
|
print(f"search_knowledge includes apps: {len(apps_hits)} app hits / {len(combined)} total")
|
|
except Exception as exc:
|
|
print(f"FAIL search_knowledge: {exc}")
|
|
errors.append("search_knowledge")
|
|
|
|
if errors:
|
|
print("P7 FAILED:", errors)
|
|
return 1
|
|
print("P7 OK")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|