diff --git a/scripts/test_llm_api.py b/scripts/test_llm_api.py new file mode 100644 index 0000000..a26cfb1 --- /dev/null +++ b/scripts/test_llm_api.py @@ -0,0 +1,307 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Erweiterter API-Schnelltest – v2.0 + +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 + +Konfiguration via 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 ✓/✗. +""" + +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") +WIKI_USER = os.getenv("WIKI_BOT_USER") +WIKI_PASS = os.getenv("WIKI_BOT_PASSWORD") + +COL = "test_collection" +SRC = "unit-test-src" + +# ---- helpers ---- + +def fail(msg, resp: requests.Response | None = None): + print("✗", msg) + if resp is not None: + try: + print(" → status:", resp.status_code) + print(" → body:", resp.text[:800]) + except Exception: + pass + sys.exit(1) + + +def ok(msg): + print("✓", msg) + + +# ---- basic ---- + +def test_openapi(): + r = requests.get(f"{BASE}/openapi.json") + if r.status_code != 200: + fail("/openapi.json nicht erreichbar", r) + ok("OpenAPI erreichbar") + + +# ---- wiki ---- + +def test_wiki_health(): + r = requests.get(f"{BASE}/import/wiki/health", params={"verbose": 1}) + if r.status_code != 200 or r.json().get("status") != "ok": + fail("/import/wiki/health fehlgeschlagen", r) + ok("Wiki /health ok") + + +def test_wiki_semantic_pages_pick_one(): + 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 + + +def test_wiki_info(title: str): + r = requests.get(f"{BASE}/import/wiki/info", params={"title": title}) + if r.status_code != 200: + fail("/import/wiki/info fehlgeschlagen", r) + js = r.json() + if not js.get("pageid"): + fail("/info ohne pageid", r) + ok("Wiki /info ok") + + +def test_wiki_parse(pageid: int, title: str): + r = requests.get(f"{BASE}/import/wiki/parsepage", params={"pageid": pageid, "title": title}) + if r.status_code != 200: + fail("/import/wiki/parsepage fehlgeschlagen", r) + js = r.json() + if not isinstance(js.get("wikitext"), str): + fail("/parsepage ohne wikitext", r) + 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): + return { + "external_id": external_id, + "fingerprint": "unit-test-sha", + "source": "UnitTest", + "title": "Testübung Reaktion", + "summary": "Kurzbeschreibung für Test.", + "short_description": "Kurzbeschreibung für Test.", + "keywords": ["Reaktion", "Bälle", "Bälle"], + "link": "http://example.local", + "discipline": "Karate", + "group": "5", + "age_group": "Teenager", + "target_group": "Breitensport", + "min_participants": 1, + "duration_minutes": 10, + "capabilities": {"Reaktionsfähigkeit": 2, "Kopplungsfähigkeit": 1}, + "category": "Übungen", + "purpose": "Aufwärmen", + "execution": "Einfacher Ablauf.", + "notes": "Hinweise.", + "preparation": "Bälle holen.", + "method": "Frontale Methode", + "equipment": ["Bälle", "Pratze"] + } + + +def test_exercise_upsert_and_idempotence(): + ext = f"ut:{uuid.uuid4()}" + payload = make_exercise_payload(ext) + + # create + r1 = requests.post(f"{BASE}/exercise", json=payload) + if r1.status_code != 200: + fail("POST /exercise (create) fehlgeschlagen", r1) + id1 = r1.json().get("id") + if not id1: + fail("POST /exercise lieferte keine id", r1) + + # update (idempotent) + r2 = requests.post(f"{BASE}/exercise", json=payload) + if r2.status_code != 200: + fail("POST /exercise (update) fehlgeschlagen", r2) + id2 = r2.json().get("id") + if id2 != id1: + fail("Idempotenz verletzt: id hat sich geändert") + + # by-external-id + r3 = requests.get(f"{BASE}/exercise/by-external-id", params={"external_id": ext}) + if r3.status_code != 200: + fail("GET /exercise/by-external-id fehlgeschlagen", r3) + id3 = r3.json().get("id") + if id3 != id1: + fail("Lookup by external_id liefert andere id") + + ok("Exercise Upsert & Idempotenz ok") + return ext, id1 + + +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"], + "capability_names": ["Reaktionsfähigkeit"], + "capability_ge_level": 2, + "limit": 10, + } + r = requests.post(f"{BASE}/exercise/search", json=req) + if r.status_code != 200: + fail("POST /exercise/search (Filter) fehlgeschlagen", r) + 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") + else: + ok("Exercise Search (Filter) ok – Testpunkt gefunden") + + +def test_exercise_search_vector(): + req = { + "query": "Aufwärmen 10min, Reaktionsfähigkeit, Teenager, Bälle", + "discipline": "Karate", + "limit": 5 + } + r = requests.post(f"{BASE}/exercise/search", json=req) + if r.status_code != 200: + fail("POST /exercise/search (Vector) fehlgeschlagen", r) + js = r.json(); hits = js.get("hits", []) + if not isinstance(hits, list): + fail("/exercise/search (Vector) liefert ungültige Struktur", r) + ok(f"Exercise Search (Vector) ok – {len(hits)} Treffer") + + +def test_exercise_delete(ext_id: str): + r = requests.delete(f"{BASE}/exercise/delete-by-external-id", params={"external_id": ext_id}) + if r.status_code != 200: + fail("DELETE /exercise/delete-by-external-id fehlgeschlagen", r) + 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() + test_wiki_info(title) + test_wiki_parse(pageid, title) + test_wiki_login_if_env() + + # Exercise + ext, _id = test_exercise_upsert_and_idempotence() + test_exercise_search_filter(ext) + 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")