Trainer_LLM/scripts/test_llm_api.py

#!/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")