253 lines
11 KiB
Python
253 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""API-Test fuer den Trainingsplan-Server (End-to-End, offener Modus).
|
|
|
|
Lauf: python3 tests/test_api.py
|
|
Voraussetzung: Server-Code liegt eine Ebene ueber tests/.
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import urllib.parse
|
|
|
|
BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
STATE = "/tmp/tp_state"
|
|
PORT = 8211
|
|
BASE_URL = f"http://127.0.0.1:{PORT}"
|
|
|
|
PASS = 0
|
|
FAIL = 0
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print(f" ok {name}")
|
|
else:
|
|
FAIL += 1
|
|
print(f"FAIL {name} {detail}")
|
|
|
|
|
|
def req(method, path, body=None, token=None):
|
|
url = BASE_URL + path
|
|
data = None
|
|
headers = {}
|
|
if body is not None:
|
|
data = json.dumps(body).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
if token:
|
|
headers["Authorization"] = "Bearer " + token
|
|
r = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(r, timeout=5) as resp:
|
|
return resp.status, json.loads(resp.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
try:
|
|
return e.code, json.loads(e.read().decode("utf-8"))
|
|
except Exception:
|
|
return e.code, {}
|
|
|
|
|
|
def raw(method, path):
|
|
r = urllib.request.Request(BASE_URL + path, method=method)
|
|
with urllib.request.urlopen(r, timeout=5) as resp:
|
|
return resp.status, resp.read().decode("utf-8")
|
|
|
|
|
|
def main():
|
|
if os.path.exists(STATE):
|
|
shutil.rmtree(STATE)
|
|
os.makedirs(STATE, exist_ok=True)
|
|
env = dict(os.environ, TRAINING_STATE_DIR=STATE, PORT=str(PORT))
|
|
proc = subprocess.Popen(
|
|
[sys.executable, os.path.join(BASE, "server.py")],
|
|
env=env, cwd=BASE,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
for _ in range(40):
|
|
time.sleep(0.25)
|
|
try:
|
|
urllib.request.urlopen(BASE_URL + "/api/meta", timeout=1)
|
|
break
|
|
except Exception:
|
|
pass
|
|
else:
|
|
print("Server nicht erreichbar")
|
|
sys.exit(1)
|
|
|
|
st, meta = req("GET", "/api/meta")
|
|
check("meta", st == 200 and "today" in meta and "now" in meta, meta)
|
|
|
|
st, cfg = req("GET", "/api/config")
|
|
acts = cfg["activities"]
|
|
check("config hat 3 Aktivitaeten", len(acts) == 3,
|
|
[a["id"] for a in acts])
|
|
gym = next(a for a in acts if a["id"] == "gym")
|
|
check("gym normalisiert (Push/Pull/Legs)",
|
|
len(gym["cycle"]["days"]) == 3,
|
|
[d["key"] for d in gym["cycle"]["days"]])
|
|
push = gym["cycle"]["days"][0]
|
|
ex = push["exercises"][0]
|
|
check("Uebung hat key + sets", ex["key"] == "Push/Bankdrücken KH (je Hand)"
|
|
and ex["sets"]["count"] == 4, ex)
|
|
run = next(a for a in acts if a["id"] == "run")
|
|
check("run fieldTemplates", len(run["fieldTemplates"]) == 5,
|
|
run["fieldTemplates"])
|
|
|
|
st, static = raw("GET", "/")
|
|
check("index.html", st == 200 and "<html" in static)
|
|
st, css = raw("GET", "/styles.css")
|
|
check("styles.css", st == 200 and "--accent" in css)
|
|
st, js = raw("GET", "/app.js")
|
|
check("app.js", st == 200 and "makeModel" in js)
|
|
|
|
st, sess = req("POST", "/api/sessions", {"activity": "gym", "day": "Push"})
|
|
check("session starten", st == 200 and sess["status"] == "active"
|
|
and sess["day"] == "Push" and sess["start_time"], sess)
|
|
sid = sess["id"]
|
|
|
|
st, act = req("GET", "/api/sessions?status=active")
|
|
check("aktive Session sichtbar", st == 200 and len(act) == 1)
|
|
|
|
st, done = req("GET", "/api/sessions?status=done")
|
|
check("kein Geist im Verlauf (vor Abschluss)", st == 200 and len(done) == 0)
|
|
|
|
setsA = [
|
|
{"exercise": ex["key"], "sort": 0, "weight": "18", "reps": "10", "done": 1},
|
|
{"exercise": ex["key"], "sort": 1, "weight": "18,5", "reps": "8", "done": 1},
|
|
{"exercise": ex["key"], "sort": 2, "weight": "20", "reps": "6", "done": 0},
|
|
]
|
|
st, r = req("PATCH", f"/api/sessions/{sid}", {"sets": setsA})
|
|
check("sets Uebung A schreiben", st == 200, r)
|
|
entriesA = r["entries"]
|
|
check("3 Sets gespeichert", len(entriesA) == 3, entriesA)
|
|
idsA = [e["id"] for e in entriesA]
|
|
check("stabile IDs", all(x for x in idsA))
|
|
check("Komma-Gewicht bleibt erhalten",
|
|
next(e for e in entriesA if e["sort"] == 1)["weight"] == "18,5")
|
|
|
|
setsA2 = [
|
|
{"exercise": ex["key"], "sort": 0, "weight": "18", "reps": "10", "done": 1},
|
|
{"exercise": ex["key"], "sort": 1, "weight": "20", "reps": "8", "done": 1},
|
|
{"exercise": ex["key"], "sort": 2, "weight": "20", "reps": "6", "done": 0},
|
|
]
|
|
st, r = req("PATCH", f"/api/sessions/{sid}", {"sets": setsA2})
|
|
idsA2 = [e["id"] for e in r["entries"]]
|
|
check("Upsert haelt IDs stabil", idsA2 == idsA, (idsA, idsA2))
|
|
|
|
exB = push["exercises"][1]
|
|
setsB = [{"exercise": exB["key"], "sort": 0, "weight": "14", "reps": "12", "done": 0}]
|
|
st, r = req("PATCH", f"/api/sessions/{sid}", {"sets": setsB})
|
|
eA = next(e for e in r["entries"] if e["exercise"] == ex["key"] and e["sort"] == 1)
|
|
eB = next(e for e in r["entries"] if e["exercise"] == exB["key"])
|
|
check("Uebung A unberuehrt (ID + Werte)", eA["id"] == idsA2[1]
|
|
and eA["weight"] == "20" and eA["reps"] == "8", eA)
|
|
check("Uebung B angelegt", eB["weight"] == "14", eB)
|
|
|
|
setsA3 = [
|
|
{"exercise": ex["key"], "sort": 0, "weight": "18", "reps": "10", "done": 1},
|
|
{"exercise": ex["key"], "sort": 1, "weight": "20", "reps": "8", "done": 1},
|
|
]
|
|
st, r = req("PATCH", f"/api/sessions/{sid}", {"sets": setsA3})
|
|
check("Set 3 entfernt", len([e for e in r["entries"]
|
|
if e["exercise"] == ex["key"]]) == 2)
|
|
check("Uebung B nach Entfernen unberuehrt",
|
|
any(e["exercise"] == exB["key"] for e in r["entries"]))
|
|
|
|
st, r = req("PATCH", f"/api/sessions/{sid}",
|
|
{"meta": {"note": "gutes Training", "status": "done"}})
|
|
check("abschliessen setzt Endzeit", r["end_time"] != ""
|
|
and r["status"] == "done" and r["note"] == "gutes Training", r)
|
|
|
|
st, done = req("GET", "/api/sessions?status=done")
|
|
check("Session sofort im Verlauf", st == 200 and len(done) == 1)
|
|
|
|
st, act = req("GET", "/api/sessions?status=active")
|
|
check("nicht mehr aktiv", len(act) == 0)
|
|
|
|
st, rv = req("GET", f"/api/reset-values?exercise={urllib.parse.quote(ex['key'])}&set=1")
|
|
check("reset-values liefert letzte Messung", rv.get("weight") == "20", rv)
|
|
|
|
st, sess2 = req("POST", "/api/sessions", {"activity": "gym", "day": "Push"})
|
|
sid2 = sess2["id"]
|
|
st, r = req("PATCH", f"/api/sessions/{sid2}",
|
|
{"sets": [{"exercise": ex["key"], "sort": 1, "weight": "99", "reps": "5", "done": 1}]})
|
|
st, rv = req("GET", f"/api/reset-values?exercise={urllib.parse.quote(ex['key'])}&set=1")
|
|
check("Reset ignoriert heute-laufende Session",
|
|
rv.get("weight") == "20", rv)
|
|
|
|
st, r = req("PATCH", f"/api/sessions/{sid}",
|
|
{"meta": {"date": "2026-09-20", "start_time": "17:30",
|
|
"end_time": "18:40"}})
|
|
check("Verlauf: Datum/Zeit editierbar",
|
|
r["date"] == "2026-09-20" and r["start_time"] == "17:30"
|
|
and r["end_time"] == "18:40", r)
|
|
|
|
st, cs = req("POST", "/api/sessions", {"activity": "run"})
|
|
cid = cs["id"]
|
|
st, r = req("PATCH", f"/api/sessions/{cid}",
|
|
{"meta": {"fields": {"distanceKm": "5,2", "minutes": "32",
|
|
"pacePerKm": "6:09", "heartRate": "148",
|
|
"elevation": "45"},
|
|
"status": "done"}})
|
|
check("Cardio-Felder gespeichert",
|
|
r["fields"]["distanceKm"] == "5,2" and r["fields"]["pacePerKm"] == "6:09", r["fields"])
|
|
|
|
st, prev = req("POST", "/api/sessions",
|
|
{"activity": "gym", "day": "Push", "date": "2026-09-15"})
|
|
pid = prev["id"]
|
|
req("PATCH", f"/api/sessions/{pid}",
|
|
{"sets": [{"exercise": ex["key"], "sort": 0, "weight": "22", "reps": "8", "done": 1}],
|
|
"meta": {"status": "done"}})
|
|
st, stats = req("GET", f"/api/stats?exercise={urllib.parse.quote(ex['key'])}")
|
|
check("Statistik: 2 Termine", st == 200 and len(stats) == 2, stats)
|
|
s0, s1 = stats[0], stats[1]
|
|
check("Statistik: Delta zur ersten Messung",
|
|
s0["delta"] == 0 and abs(s1["delta"]) > 0, stats)
|
|
check("Statistik: bestes Gewicht je Termin",
|
|
s0["weight"] == 22 and s1["weight"] == 20, stats)
|
|
|
|
st, exp = req("GET", "/api/export?format=json")
|
|
check("Export JSON Format", st == 200
|
|
and exp["format"] == "trainingsplan-export-v2"
|
|
and len(exp["sessions"]) == 4, exp.get("format"))
|
|
st, body = raw("GET", "/api/export?format=csv")
|
|
check("Export CSV Header", st == 200 and "session_id" in body
|
|
and "distanceKm" in body, body[:120])
|
|
|
|
st, imp = req("POST", "/api/import",
|
|
{"format": "trainingsplan-export-v2",
|
|
"data": json.dumps(exp)})
|
|
check("Import JSON (Merge) ok", st == 200 and imp["created"] == 0
|
|
and imp["updated"] == 4, imp)
|
|
st, csvimp = req("POST", "/api/import",
|
|
{"format": "csv", "data": body})
|
|
check("Import CSV ok", st == 200 and csvimp["updated"] == 4, csvimp)
|
|
|
|
st, r = req("DELETE", f"/api/sessions/{cid}")
|
|
check("Session loeschen", st == 200 and r["deleted"] == 1, r)
|
|
st, r = req("DELETE", f"/api/sessions/{sid2}")
|
|
check("Session loeschen (sid2)", r.get("deleted") == 1, r)
|
|
|
|
st, r = req("POST", "/api/sessions", {})
|
|
check("POST ohne activity -> 400", st == 400, (st, r))
|
|
st, r = req("PATCH", "/api/sessions/99999", {"meta": {"note": "x"}})
|
|
check("PATCH unbekannte Session -> 404", st == 404, (st, r))
|
|
st, r = req("GET", "/api/nonexistent")
|
|
check("unbekannter Pfad -> 404", st == 404)
|
|
|
|
finally:
|
|
proc.terminate()
|
|
proc.wait(timeout=5)
|
|
|
|
print(f"\n{PASS} ok, {FAIL} fail")
|
|
sys.exit(1 if FAIL else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |