Lars
75b257bb15
All checks were successful
Deploy Trainer_LLM to llm-node / deploy (push) Successful in 2s
256 lines
8.3 KiB
Python
256 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Erweiterter API-Schnelltest – v2.1
|
||
|
||
Ä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)
|
||
|
||
ENV:
|
||
BASE_URL (default http://127.0.0.1:8000)
|
||
TEST_WIKI_CATEGORY (default "Übungen")
|
||
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 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")
|
||
|
||
print("\nConfig:")
|
||
print(" BASE_URL=", BASE)
|
||
print(" TEST_WIKI_CATEGORY=", TEST_WIKI_CATEGORY)
|
||
print(" TEST_WIKI_TITLE=", TEST_WIKI_TITLE or "<none>")
|
||
print(" WIKI_BOT_USER set=", bool(WIKI_USER))
|
||
|
||
# ---- 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_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 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):
|
||
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")
|
||
|
||
|
||
# ---- 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():
|
||
import time
|
||
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):
|
||
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)
|
||
if not any(h.get("payload", {}).get("external_id") == ext_id for h in hits):
|
||
ok("Exercise Search (Filter) ok – Testpunkt nicht zwingend in 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")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("\nStarte API-Tests...\n")
|
||
test_openapi()
|
||
test_wiki_health()
|
||
|
||
# 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)
|
||
|
||
# Exercise
|
||
ext, _id = test_exercise_upsert_and_idempotence()
|
||
test_exercise_search_filter(ext)
|
||
test_exercise_search_vector()
|
||
test_exercise_delete(ext)
|
||
|
||
print("\n🎉 Alle Tests erfolgreich durchlaufen!\n")
|