Dateien nach "tests" hochladen

tests/test_retriever_smoke.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+#
+# test_retriever_smoke.py – End-to-End Smoke-Test für den mindnet Retriever
+#
+# Dieser Smoke-Test ist für manuelle End-to-End-Tests gedacht.
+# Er wird NICHT automatisch von pytest ausgeführt.
+#
+# Empfohlener Speicherort: tests/test_retriever_smoke.py
+#
+# Aufrufbeispiele:
+#
+#   python tests/test_retriever_smoke.py --query "embeddings" --mode semantic
+#
+#   python tests/test_retriever_smoke.py \
+#       --query "embeddings" \
+#       --mode hybrid \
+#       --expand-depth 1 \
+#       --expand-edge-types references depends_on
+#
+
+from __future__ import annotations
+
+import argparse
+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="URL des Query-Endpunkts")
+    parser.add_argument("--query", required=True, help="Text-Query")
+    parser.add_argument("--mode", choices=["semantic", "hybrid"], default="hybrid")
+    parser.add_argument("--top-k", type=int, default=5)
+    parser.add_argument("--expand-depth", type=int, default=0)
+    parser.add_argument(
+        "--expand-edge-types",
+        nargs="*",
+        default=None,
+        help="Edge-Typen (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:
+        print(f"[ERROR] HTTP-Request fehlgeschlagen: {exc}", file=sys.stderr)
+        return 1
+
+    if resp.status_code != 200:
+        print(f"[ERROR] Statuscode: {resp.status_code}", file=sys.stderr)
+        print(resp.text, file=sys.stderr)
+        return 1
+
+    try:
+        data = resp.json()
+    except Exception:
+        print("[ERROR] Ungültiges JSON:", 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 []
+    print(f"used_mode:  {data.get('used_mode')}")
+    print(f"latency:    {data.get('latency_ms')} ms")
+    print(f"Treffer:    {len(results)}")
+    print()
+
+    for i, hit in enumerate(results, start=1):
+        nid = hit.get("note_id")
+        cid = hit.get("node_id")
+        tot = hit.get("total_score")
+        sem = hit.get("semantic_score")
+        edge = hit.get("edge_bonus")
+        cen = hit.get("centrality_bonus")
+        src = hit.get("source") or {}
+        path = src.get("path")
+        sec = src.get("section")
+
+        print(f"[{i}] note_id={nid} node_id={cid}")
+        try:
+            print(
+                f"     total={float(tot):.4f} "
+                f"(semantic={float(sem):.4f} edge={float(edge):.4f} centrality={float(cen):.4f})"
+            )
+        except Exception:
+            print(f"     total={tot}, semantic={sem}, edge={edge}, centrality={cen}")
+
+        if path or sec:
+            print(f"     source: {path or '-'} :: {sec or '-'}")
+        print()
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())