Dateien nach "tests" hochladen

tests/scripts_test_retriever_smoke.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+scripts/test_retriever_smoke.py - einfacher Smoke-Test für den mindnet-Retriever
+
+Zweck:
+    - Startet eine Query gegen die laufende FastAPI-Instanz
+    - Prüft, ob der /query-Endpunkt antwortet und sinnvolle Treffer liefert
+    - Gibt Top-K-Ergebnisse inkl. Scores auf stdout aus
+
+Verwendung:
+    1) Stelle sicher, dass der mindnet-Service läuft, z. B.:
+           uvicorn app.main:create_app --factory --reload
+
+    2) Führe dieses Skript aus:
+           python scripts/test_retriever_smoke.py --query "karate" --mode hybrid
+
+    3) Optional:
+           python scripts/test_retriever_smoke.py --query "embeddings" --mode semantic --top-k 5
+
+Hinweis:
+    - Standard-URL ist http://127.0.0.1:8000
+    - Endpoint ist /query (bestehender Router), der QueryRequest erwartet
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from typing import Any, Dict
+
+import requests
+
+
+def _build_payload(args: argparse.Namespace) -> Dict[str, Any]:
+    payload: Dict[str, Any] = {
+        "mode": args.mode,
+        "query": args.query,
+        "top_k": args.top_k,
+    }
+    if args.expand_depth is not None and args.expand_depth > 0:
+        expand: Dict[str, Any] = {"depth": args.expand_depth}
+        if args.expand_edge_types:
+            expand["edge_types"] = args.expand_edge_types
+        payload["expand"] = expand
+    return payload
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Smoke-Test für mindnet-Retriever (/query)")
+    parser.add_argument("--url", default="http://127.0.0.1:8000/query", help="Basis-URL des Query-Endpunkts")
+    parser.add_argument("--query", required=True, help="Text-Query für den Retriever")
+    parser.add_argument("--mode", choices=["semantic", "hybrid"], default="hybrid", help="Retrieval-Modus")
+    parser.add_argument("--top-k", type=int, default=5, help="Anzahl der gewünschten Treffer")
+    parser.add_argument("--expand-depth", type=int, default=0, help="Edge-Expansionstiefe (nur hybrid)")
+    parser.add_argument(
+        "--expand-edge-types",
+        nargs="*",
+        default=None,
+        help="Liste von Edge-Typen für die Expansion (z. B. references depends_on)",
+    )
+
+    args = parser.parse_args(argv)
+
+    payload = _build_payload(args)
+
+    try:
+        resp = requests.post(args.url, json=payload, timeout=10)
+    except Exception as exc:  # pragma: no cover - Netzwerkfehler
+        print(f"[ERROR] HTTP-Request fehlgeschlagen: {exc}", file=sys.stderr)
+        return 1
+
+    if resp.status_code != 200:
+        print(f"[ERROR] Unerwarteter Statuscode: {resp.status_code}", file=sys.stderr)
+        print(resp.text, file=sys.stderr)
+        return 1
+
+    try:
+        data = resp.json()
+    except Exception as exc:  # pragma: no cover - JSON-Fehler
+        print(f"[ERROR] Antwort ist kein gültiges JSON: {exc}", file=sys.stderr)
+        print(resp.text, file=sys.stderr)
+        return 1
+
+    print("=== mindnet Retriever Smoke-Test ===")
+    print(f"URL:    {args.url}")
+    print(f"Mode:   {args.mode}")
+    print(f"Query:  {args.query!r}")
+    print(f"Top-K:  {args.top_k}")
+    print()
+
+    results = data.get("results") or []
+    used_mode = data.get("used_mode")
+    latency_ms = data.get("latency_ms")
+
+    print(f"used_mode = {used_mode}, latency = {latency_ms} ms")
+    print(f"Treffer insgesamt: {len(results)}")
+    print()
+
+    for i, hit in enumerate(results, start=1):
+        note_id = hit.get("note_id")
+        node_id = hit.get("node_id")
+        total = hit.get("total_score")
+        sem = hit.get("semantic_score")
+        edge_bonus = hit.get("edge_bonus")
+        cent_bonus = hit.get("centrality_bonus")
+        src = hit.get("source") or {}
+        path = src.get("path")
+        section = src.get("section")
+
+        print(f"[{i}] note_id={note_id} node_id={node_id}")
+        print(f"     total={total:.4f} (semantic={sem:.4f} edge={edge_bonus:.4f} centrality={cent_bonus:.4f})")
+        if path or section:
+            print(f"     source: {path or '-'} :: {section or '-'}")
+        print()
+
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())