diff --git a/scripts/wp04_smoketest.py b/scripts/wp04_smoketest.py new file mode 100644 index 0000000..83b4b40 --- /dev/null +++ b/scripts/wp04_smoketest.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +scripts/wp04_smoketest.py — E2E-Schnelltest der WP-04 Endpunkte + +Zweck: + - Holt exemplarisch einen Vektor aus {prefix}_chunks (falls verfügbar), + - ruft POST /query mit diesem Vektor auf, + - ruft GET /graph/ für die Top-Note auf, + - druckt Kurzresultate auf STDOUT. + +Kompatibilität: + Python 3.12+, requests, qdrant-client +Version: + 0.1.0 (Erstanlage) +Stand: + 2025-10-07 +Bezug: + WP-04 /query & /graph +Nutzung: + export QDRANT_URL="http://localhost:6333" + export MINDNET_PREFIX="mindnet" + python3 scripts/wp04_smoketest.py --api http://localhost:8000 +Änderungsverlauf: + 0.1.0 (2025-10-07) – Erstanlage. +""" + +from __future__ import annotations +import os +import sys +import json +import argparse +import requests +from qdrant_client import QdrantClient + + +def get_any_chunk_vector(client: QdrantClient, prefix: str): + """Holt eine beliebige Chunk-Vector/Note-ID-Kombi aus Qdrant (falls vorhanden).""" + col = f"{prefix}_chunks" + pts, _ = client.scroll(collection_name=col, limit=1, with_vectors=True, with_payload=True) + if not pts: + return None, None, None + p = pts[0] + vec = p.vector + payload = p.payload or {} + node_id = str(p.id) + note_id = payload.get("note_id") + return vec, node_id, note_id + + +def run(api_base: str, qdrant_url: str, prefix: str) -> int: + print(f"[i] API: {api_base}") + print(f"[i] Qdrant: {qdrant_url}") + client = QdrantClient(url=qdrant_url) + + vec, node_id, note_id = get_any_chunk_vector(client, prefix) + if vec is None: + print("[!] Keine Chunks mit Vektoren gefunden. Bitte Import (WP-03) prüfen.") + return 2 + + print(f"[i] Beispiel-Chunk-ID: {node_id}, Note-ID: {note_id}") + print("[i] Rufe /query (hybrid) mit Beispielvektor auf ...") + + q_payload = { + "mode": "hybrid", + "query_vector": vec, # 384d + "top_k": 5, + "expand": {"depth": 1, "edge_types": ["references", "belongs_to", "prev", "next"]}, + } + r = requests.post(f"{api_base.rstrip('/')}/query", json=q_payload, timeout=60) + print(f"[i] /query Status: {r.status_code}") + if r.status_code != 200: + print(r.text) + return 3 + + data = r.json() + hits = data.get("results", []) + print(f"[i] Treffer: {len(hits)}") + for i, h in enumerate(hits, 1): + print(f" {i:02d}. node_id={h['node_id']} note_id={h.get('note_id')} total={h['total_score']:.3f}") + + # Graph-Test (wenn Note-ID vorhanden) + gid = note_id or (hits[0].get("note_id") if hits else None) + if gid: + print(f"[i] Rufe /graph/{gid} ...") + r2 = requests.get(f"{api_base.rstrip('/')}/graph/{gid}?depth=1", timeout=60) + print(f"[i] /graph Status: {r2.status_code}") + if r2.status_code == 200: + g = r2.json() + print(f" Nodes: {len(g.get('nodes', []))}, Edges: {len(g.get('edges', []))}") + else: + print(r2.text) + + print("[✔] Smoke-Test abgeschlossen.") + return 0 + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--api", default="http://localhost:8000", help="Basis-URL der FastAPI.") + args = ap.parse_args() + + qdrant_url = os.getenv("QDRANT_URL", "http://127.0.0.1:6333") + prefix = os.getenv("MINDNET_PREFIX", "mindnet") + sys.exit(run(args.api, qdrant_url, prefix)) + + +if __name__ == "__main__": + main()