Trainer_LLM/scripts/test_plans_wp15.py

#!/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():
    # Wenn Qdrant nicht erreichbar ist, Test überspringen
    try:
        rq = requests.post(
            f"{QDRANT}/collections/{TPL_COLL}/points/scroll",
            json={"with_payload": True, "limit": 1},
            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 Punkte in plan_templates – überspringe")

    payload = pts[0].get("payload") or {}
    # Erwartet: materialisierte Listen vorhanden
    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"