Lars
c0bb562a8d
All checks were successful
Deploy Trainer_LLM to llm-node / deploy (push) Successful in 2s
308 lines
9.5 KiB
Python
308 lines
9.5 KiB
Python
#!/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")
|