Lars
1dbcf33540
All checks were successful
Deploy Trainer_LLM to llm-node / deploy (push) Successful in 1s
241 lines
8.1 KiB
Python
241 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""Unit-, Integrations- und E2E-Tests für WP-15 (v1.2.0)."""
|
||
import os
|
||
import json
|
||
import requests
|
||
import pytest
|
||
from datetime import datetime, timezone
|
||
|
||
BASE = os.getenv("BASE_URL", "http://127.0.0.1:8000").rstrip("/")
|
||
QDRANT = os.getenv("QDRANT_BASE", "http://127.0.0.1:6333").rstrip("/")
|
||
TPL_COLL = os.getenv("PLAN_TEMPLATE_COLLECTION", "plan_templates")
|
||
|
||
# ---------- Helpers ----------
|
||
|
||
def _fp_local(plan_payload: dict) -> str:
|
||
import hashlib
|
||
core = {
|
||
"title": plan_payload["title"],
|
||
"total_minutes": int(plan_payload["total_minutes"]),
|
||
"items": [
|
||
{"exercise_external_id": it["exercise_external_id"], "duration": int(it["duration"])}
|
||
for sec in plan_payload["sections"]
|
||
for it in sec.get("items", [])
|
||
],
|
||
}
|
||
raw = json.dumps(core, sort_keys=True, ensure_ascii=False)
|
||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _unique_len_lower(xs):
|
||
seen = set()
|
||
out = []
|
||
for x in xs:
|
||
k = x.casefold()
|
||
if k not in seen:
|
||
seen.add(k)
|
||
out.append(x)
|
||
return len(out)
|
||
|
||
# ---------- Unit ----------
|
||
|
||
def test_fingerprint_unit_v12():
|
||
p = {
|
||
"title": "Montag – Reaktion",
|
||
"total_minutes": 90,
|
||
"sections": [
|
||
{"name": "Warmup", "minutes": 15, "items": [
|
||
{"exercise_external_id": "ex:001", "duration": 10, "why": "Aufwärmen"}
|
||
]}
|
||
],
|
||
}
|
||
fp1 = _fp_local(p)
|
||
p2 = json.loads(json.dumps(p, ensure_ascii=False))
|
||
fp2 = _fp_local(p2)
|
||
assert fp1 == fp2
|
||
|
||
# ---------- Integration: Templates mit mehreren Sections + ideal/supplement ----------
|
||
|
||
def test_template_sections_ideal_supplement_roundtrip():
|
||
tpl = {
|
||
"name": "Std 90 v1.1",
|
||
"discipline": "Karate",
|
||
"age_group": "Teenager",
|
||
"target_group": "Breitensport",
|
||
"total_minutes": 90,
|
||
"sections": [
|
||
{
|
||
"name": "Warmup",
|
||
"target_minutes": 15,
|
||
"must_keywords": ["Reaktion", "reaktion"], # Duplikat in anderer Schreibweise
|
||
"ideal_keywords": ["Koordination", "koordination"], # Duplikat in anderer Schreibweise
|
||
"supplement_keywords": ["Teamspiel", " Teamspiel "], # Duplikat mit Whitespace
|
||
"forbid_keywords": [],
|
||
"capability_targets": {"Reaktionsfähigkeit": 2, "Mobilität": 1}
|
||
},
|
||
{
|
||
"name": "Technikblock",
|
||
"target_minutes": 30,
|
||
"must_keywords": ["Mae-Geri"],
|
||
"ideal_keywords": ["Timing"],
|
||
"supplement_keywords": ["Partnerarbeit"],
|
||
"forbid_keywords": ["Bodenarbeit"],
|
||
"capability_targets": {"Technikpräzision": 2, "Schnelligkeit": 1}
|
||
}
|
||
],
|
||
"goals": ["Technik", "Kondition"],
|
||
"equipment_allowed": ["Bälle"],
|
||
"created_by": "tester",
|
||
"version": "1.1"
|
||
}
|
||
|
||
r = requests.post(f"{BASE}/plan_templates", json=tpl)
|
||
assert r.status_code == 200, r.text
|
||
tpl_id = r.json()["id"]
|
||
|
||
r2 = requests.get(f"{BASE}/plan_templates/{tpl_id}")
|
||
assert r2.status_code == 200, r2.text
|
||
got = r2.json()
|
||
|
||
# Prüfen: Beide Sections vorhanden
|
||
assert len(got["sections"]) == 2
|
||
|
||
s1 = got["sections"][0]
|
||
# Normalisierung: Duplikate entfernt (case-insensitive), Whitespace getrimmt
|
||
assert _unique_len_lower(s1["must_keywords"]) == len(s1["must_keywords"]) == 1
|
||
assert _unique_len_lower(s1["ideal_keywords"]) == len(s1["ideal_keywords"]) == 1
|
||
assert _unique_len_lower(s1["supplement_keywords"]) == len(s1["supplement_keywords"]) == 1
|
||
|
||
# ---------- Optional: Qdrant-Payload-Check (materialisierte Felder) ----------
|
||
|
||
def test_qdrant_materialized_fields_template():
|
||
"""
|
||
Robust: Erst frisches Template anlegen (mit neuen Feldern), dann Qdrant-Scroll
|
||
mit Filter auf genau diese Template-ID. So treffen wir sicher einen Punkt,
|
||
der die materialisierten Felder enthält – unabhängig von älteren Datensätzen.
|
||
"""
|
||
# 1) Frisches Template erzeugen (mit ideal/supplement)
|
||
tpl = {
|
||
"name": "Std 90 v1.1 – materialized-check",
|
||
"discipline": "Karate",
|
||
"age_group": "Teenager",
|
||
"target_group": "Breitensport",
|
||
"total_minutes": 90,
|
||
"sections": [
|
||
{
|
||
"name": "Warmup",
|
||
"target_minutes": 15,
|
||
"must_keywords": ["Reaktion"],
|
||
"ideal_keywords": ["Koordination"],
|
||
"supplement_keywords": ["Teamspiel"],
|
||
"forbid_keywords": [],
|
||
"capability_targets": {"Reaktionsfähigkeit": 2}
|
||
}
|
||
],
|
||
"goals": ["Technik"],
|
||
"equipment_allowed": ["Bälle"],
|
||
"created_by": "tester",
|
||
"version": "1.1"
|
||
}
|
||
r = requests.post(f"{BASE}/plan_templates", json=tpl)
|
||
assert r.status_code == 200, r.text
|
||
tpl_id = r.json()["id"]
|
||
|
||
# 2) Qdrant gezielt nach genau diesem Punkt scrollen (Payload enthält id)
|
||
try:
|
||
rq = requests.post(
|
||
f"{QDRANT}/collections/{TPL_COLL}/points/scroll",
|
||
json={
|
||
"with_payload": True,
|
||
"limit": 1,
|
||
"filter": {"must": [{"key": "id", "match": {"value": tpl_id}}]}
|
||
},
|
||
timeout=2.0,
|
||
)
|
||
except Exception:
|
||
pytest.skip("Qdrant nicht erreichbar – überspringe materialisierte Feldprüfung")
|
||
|
||
if rq.status_code != 200:
|
||
pytest.skip(f"Qdrant-Scroll liefert {rq.status_code}")
|
||
|
||
js = rq.json()
|
||
pts = (js.get("result") or {}).get("points") or []
|
||
if not pts:
|
||
pytest.skip("Keine Übereinstimmung in plan_templates – überspringe")
|
||
|
||
payload = pts[0].get("payload") or {}
|
||
for key in [
|
||
"section_names",
|
||
"section_must_keywords",
|
||
"section_ideal_keywords",
|
||
"section_supplement_keywords",
|
||
"section_forbid_keywords",
|
||
]:
|
||
assert key in payload
|
||
assert isinstance(payload[key], list)
|
||
|
||
# ---------- E2E: Plan anlegen + Idempotenz trotz variierender Nebenfelder ----------
|
||
|
||
def test_plan_e2e_idempotence_same_fingerprint():
|
||
# Template minimal für Bezug
|
||
tpl = {
|
||
"name": "Std 90 for plan",
|
||
"discipline": "Karate",
|
||
"age_group": "Teenager",
|
||
"target_group": "Breitensport",
|
||
"total_minutes": 90,
|
||
"sections": [
|
||
{"name": "Warmup", "target_minutes": 15, "must_keywords": [], "forbid_keywords": [], "capability_targets": {}}
|
||
],
|
||
"goals": [],
|
||
"equipment_allowed": [],
|
||
"created_by": "tester",
|
||
"version": "1.0"
|
||
}
|
||
r1 = requests.post(f"{BASE}/plan_templates", json=tpl)
|
||
assert r1.status_code == 200
|
||
tpl_id = r1.json()["id"]
|
||
|
||
plan_base = {
|
||
"template_id": tpl_id,
|
||
"title": "KW32 – Montag",
|
||
"discipline": "Karate",
|
||
"age_group": "Teenager",
|
||
"target_group": "Breitensport",
|
||
"total_minutes": 90,
|
||
"sections": [
|
||
{"name": "Warmup", "minutes": 15, "items": [
|
||
{"exercise_external_id": "ex:001", "duration": 10, "why": "Aufwärmen"}
|
||
]}
|
||
],
|
||
"created_by": "tester",
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
"source": "test"
|
||
}
|
||
|
||
# Plan A mit goals/capability_summary
|
||
plan_a = dict(plan_base)
|
||
plan_a.update({
|
||
"goals": ["Technik"],
|
||
"capability_summary": {"Reaktionsfähigkeit": 2}
|
||
})
|
||
r2 = requests.post(f"{BASE}/plan", json=plan_a)
|
||
assert r2.status_code == 200, r2.text
|
||
plan_id = r2.json()["id"]
|
||
|
||
# Plan B – gleicher Fingerprint (gleiche items), aber andere Nebenfelder
|
||
plan_b = dict(plan_base)
|
||
plan_b.update({
|
||
"goals": ["Kondition"],
|
||
"capability_summary": {"Reaktionsfähigkeit": 3}
|
||
})
|
||
r3 = requests.post(f"{BASE}/plan", json=plan_b)
|
||
assert r3.status_code == 200
|
||
assert r3.json()["id"] == plan_id, "Idempotenz verletzt: gleicher Fingerprint muss gleiche ID liefern"
|
||
|
||
# GET prüfen
|
||
r4 = requests.get(f"{BASE}/plan/{plan_id}")
|
||
assert r4.status_code == 200
|
||
js = r4.json()
|
||
assert js["title"] == "KW32 – Montag" |