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