From 75b257bb150e5d85046c3d284999b49648b7effb Mon Sep 17 00:00:00 2001 From: Lars Date: Mon, 11 Aug 2025 19:44:39 +0200 Subject: [PATCH] scripts/test_llm_api.py aktualisiert --- scripts/test_llm_api.py | 166 ++++++++++++++-------------------------- 1 file changed, 57 insertions(+), 109 deletions(-) diff --git a/scripts/test_llm_api.py b/scripts/test_llm_api.py index a26cfb1..7d6cc4c 100644 --- a/scripts/test_llm_api.py +++ b/scripts/test_llm_api.py @@ -1,38 +1,37 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -Erweiterter API-Schnelltest – v2.0 +Erweiterter API-Schnelltest – v2.1 -Deckt ab: -- OpenAPI erreichbar -- Wiki-Routen: /import/wiki/health, /semantic/pages, /info, /parsepage, optional /login -- Exercise-Routen: /exercise (Upsert & Idempotenz), /exercise/by-external-id, /exercise/search (Filter-only & Vector), - /exercise/delete-by-external-id -- Embed-Routen: /embed, /search, /prompt, /delete-source, /delete-collection +Änderungen ggü. v2.0: +- **Login VOR** /semantic/pages (viele Wikis liefern Ask-Ergebnisse nur authentifiziert) +- Robuster Fallback: Wenn /semantic/pages {} liefert → optionaler ENV-Fallback `TEST_WIKI_TITLE` +- Bessere Diagnoseausgaben (zeigt Basis-URL, Category, gesetzte Login-ENV) -Konfiguration via ENV: +ENV: BASE_URL (default http://127.0.0.1:8000) - QDRANT_HTTP (default http://127.0.0.1:6333) – optional, hier nicht zwingend genutzt TEST_WIKI_CATEGORY (default "Übungen") - WIKI_BOT_USER, WIKI_BOT_PASSWORD – optional; wenn gesetzt, wird /import/wiki/login getestet - -Exit-Code != 0 bei Fehler. Ausgabe mit ✓/✗. + TEST_WIKI_TITLE (optional Fallback-Titel, z. B. "Affenklatschen") + WIKI_BOT_USER, WIKI_BOT_PASSWORD (optional; empfohlen) """ import os import sys import json import uuid -import time import requests BASE = os.getenv("BASE_URL", "http://127.0.0.1:8000").rstrip("/") TEST_WIKI_CATEGORY = os.getenv("TEST_WIKI_CATEGORY", "Übungen") +TEST_WIKI_TITLE = os.getenv("TEST_WIKI_TITLE") WIKI_USER = os.getenv("WIKI_BOT_USER") WIKI_PASS = os.getenv("WIKI_BOT_PASSWORD") -COL = "test_collection" -SRC = "unit-test-src" +print("\nConfig:") +print(" BASE_URL=", BASE) +print(" TEST_WIKI_CATEGORY=", TEST_WIKI_CATEGORY) +print(" TEST_WIKI_TITLE=", TEST_WIKI_TITLE or "") +print(" WIKI_BOT_USER set=", bool(WIKI_USER)) # ---- helpers ---- @@ -69,21 +68,45 @@ def test_wiki_health(): ok("Wiki /health ok") -def test_wiki_semantic_pages_pick_one(): +def test_wiki_login_if_env(): + if not (WIKI_USER and WIKI_PASS): + print("⚠️ Wiki /login übersprungen (WIKI_BOT_USER/PASSWORD nicht gesetzt)") + return False + r = requests.post(f"{BASE}/import/wiki/login", json={"username": WIKI_USER, "password": WIKI_PASS}) + if r.status_code != 200 or r.json().get("status") != "success": + fail("/import/wiki/login fehlgeschlagen", r) + ok("Wiki /login ok") + return True + + +def get_wiki_title_and_pageid() -> tuple[str, int]: + # 1) Versuch über semantic/pages r = requests.get(f"{BASE}/import/wiki/semantic/pages", params={"category": TEST_WIKI_CATEGORY}) if r.status_code != 200: fail("/import/wiki/semantic/pages fehlgeschlagen", r) data = r.json() - if not isinstance(data, dict) or not data: - fail("/semantic/pages lieferte keine Titel", r) - # Beliebigen Titel nehmen (stabil: erster Key) - title = next(iter(data.keys())) - entry = data[title] - pageid = entry.get("pageid") - if not pageid: - fail("/semantic/pages ohne pageid nach Enrichment", r) - ok(f"Wiki /semantic/pages ok – Beispiel: '{title}' (pageid={pageid})") - return title, pageid + if isinstance(data, dict) and data: + title = next(iter(data.keys())) + entry = data[title] + pid = entry.get("pageid") + if not pid: + fail("/semantic/pages ohne pageid nach Enrichment", r) + ok(f"Wiki /semantic/pages ok – Beispiel: '{title}' (pageid={pid})") + return title, int(pid) + + print("ℹ️ /semantic/pages lieferte keine Titel ({}). Versuche Fallback...") + # 2) Fallback über TEST_WIKI_TITLE + if not TEST_WIKI_TITLE: + fail("Kein TEST_WIKI_TITLE gesetzt und /semantic/pages leer – kann keinen Titel testen.") + r2 = requests.get(f"{BASE}/import/wiki/info", params={"title": TEST_WIKI_TITLE}) + if r2.status_code != 200: + fail("/import/wiki/info(Fallback) fehlgeschlagen", r2) + js = r2.json() + pid = js.get("pageid") + if not pid: + fail("/info(Fallback) ohne pageid", r2) + ok(f"Wiki /info Fallback ok – '{TEST_WIKI_TITLE}' (pageid={pid})") + return TEST_WIKI_TITLE, int(pid) def test_wiki_info(title: str): @@ -106,16 +129,6 @@ def test_wiki_parse(pageid: int, title: str): ok("Wiki /parsepage ok") -def test_wiki_login_if_env(): - if not (WIKI_USER and WIKI_PASS): - ok("Wiki /login übersprungen (ENV nicht gesetzt)") - return - r = requests.post(f"{BASE}/import/wiki/login", json={"username": WIKI_USER, "password": WIKI_PASS}) - if r.status_code != 200 or r.json().get("status") != "success": - fail("/import/wiki/login fehlgeschlagen", r) - ok("Wiki /login ok") - - # ---- exercise ---- def make_exercise_payload(external_id: str): @@ -146,6 +159,7 @@ def make_exercise_payload(external_id: str): def test_exercise_upsert_and_idempotence(): + import time ext = f"ut:{uuid.uuid4()}" payload = make_exercise_payload(ext) @@ -178,7 +192,6 @@ def test_exercise_upsert_and_idempotence(): def test_exercise_search_filter(ext_id: str): - # Treffer mit Level >=2 in Reaktionsfähigkeit und Ausrüstung Bälle req = { "discipline": "Karate", "equipment_all": ["Bälle"], @@ -192,9 +205,8 @@ def test_exercise_search_filter(ext_id: str): js = r.json(); hits = js.get("hits", []) if not isinstance(hits, list) or not hits: fail("/exercise/search liefert keine Treffer (Filter)", r) - # Optional prüfen, ob unser Testpunkt in den Treffern ist if not any(h.get("payload", {}).get("external_id") == ext_id for h in hits): - ok("Exercise Search (Filter) ok – Testpunkt nicht zwingend unter Top-N") + ok("Exercise Search (Filter) ok – Testpunkt nicht zwingend in Top-N") else: ok("Exercise Search (Filter) ok – Testpunkt gefunden") @@ -221,75 +233,18 @@ def test_exercise_delete(ext_id: str): ok("Exercise Delete-by-external-id ok") -# ---- embed pipeline (bestehend) ---- - -def test_embed_ingest(): - payload = { - "collection": COL, - "chunks": [ - {"text": "Das ist ein Testtext für Embed.", "source": SRC} - ] - } - r = requests.post(f"{BASE}/embed", json=payload) - if r.status_code != 200: - fail("/embed fehlgeschlagen", r) - js = r.json() - if js.get("count") != 1: - fail("/embed count != 1", r) - ok("Embed ingest ok") - - -def test_embed_search(): - params = {"query": "Testtext", "collection": COL} - r = requests.get(f"{BASE}/search", params=params) - if r.status_code != 200: - fail("/search fehlgeschlagen", r) - results = r.json() - if not any("score" in item for item in results): - fail("/search keine Treffer") - ok("Embed search ok") - - -def test_prompt(): - payload = {"query": "Wie lautet dieser Testtext?", "context_limit": 1, "collection": COL} - r = requests.post(f"{BASE}/prompt", json=payload) - if r.status_code != 200: - fail("/prompt fehlgeschlagen", r) - data = r.json() - if "answer" not in data: - fail("/prompt ohne 'answer'") - ok("Prompt ok") - - -def test_delete_source(): - params = {"collection": COL, "source": SRC} - r = requests.delete(f"{BASE}/delete-source", params=params) - if r.status_code != 200: - fail("/delete-source fehlgeschlagen", r) - data = r.json() - if data.get("count") != 1: - fail("/delete-source count != 1") - ok("Delete-source ok") - - -def test_delete_collection(): - params = {"collection": COL} - r = requests.delete(f"{BASE}/delete-collection", params=params) - if r.status_code != 200: - fail("/delete-collection fehlgeschlagen", r) - ok("Delete-collection ok") - - if __name__ == "__main__": print("\nStarte API-Tests...\n") test_openapi() - - # Wiki test_wiki_health() - title, pageid = test_wiki_semantic_pages_pick_one() + + # Login nun VOR Semantic Query + did_login = test_wiki_login_if_env() + + # Titel ermitteln (semantic/pages oder Fallback) + title, pageid = get_wiki_title_and_pageid() test_wiki_info(title) test_wiki_parse(pageid, title) - test_wiki_login_if_env() # Exercise ext, _id = test_exercise_upsert_and_idempotence() @@ -297,11 +252,4 @@ if __name__ == "__main__": test_exercise_search_vector() test_exercise_delete(ext) - # Embed - test_embed_ingest() - test_embed_search() - test_prompt() - test_delete_source() - test_delete_collection() - print("\n🎉 Alle Tests erfolgreich durchlaufen!\n")