#!/usr/bin/env python3 """Test fuer Mehrbenutzer, Workouts-API und Demo-Import. Lauf: python3 tests/test_auth.py """ import json import os import shutil import subprocess import sys import time import urllib.request import urllib.error BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) STATE = "/tmp/tp_state3" PORT = 8277 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=10) 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 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) # --- Offener Modus ------------------------------------------------- st, meta = req("GET", "/api/meta") check("offener Modus: auth_required false", meta["auth_required"] is False and meta["users"] == 0, meta) st, sess = req("POST", "/api/sessions", {"activity": "gym", "day": "Push"}) check("offener Modus: Session ohne Token", st == 200, sess) req("PATCH", f"/api/sessions/{sess['id']}", {"meta": {"status": "done"}}) # --- Registrierung (erster Benutzer) -------------------------------- st, reg = req("POST", "/api/register", {"username": "alex", "password": "geheim"}) check("Registrierung", st == 200 and reg.get("token"), reg) token_a = reg["token"] st, meta = req("GET", "/api/meta") check("nach Registrierung: auth_required true", meta["auth_required"] is True and meta["users"] == 1, meta) st, me = req("GET", "/api/me", token=token_a) check("me: username + Owner", me["username"] == "alex" and me["is_owner"] is True, me) st, _ = req("GET", "/api/me") check("me ohne Token -> 401", st == 401, st) st, _ = req("GET", "/api/config") check("config ohne Token -> 401", st == 401, st) st, _ = req("GET", "/api/sessions") check("sessions ohne Token -> 401", st == 401, st) # Migration: Alt-Daten (offener Modus) gehoeren dem ersten Benutzer st, done = req("GET", "/api/sessions?status=done", token=token_a) check("erster Benutzer uebernimmt Alt-Daten", len(done) == 1, done) # Zweite Registrierung blockiert st, _ = req("POST", "/api/register", {"username": "berta", "password": "geheim"}) check("zweite Registrierung -> 403", st == 403, st) # Login-Fehlerfaelle st, _ = req("POST", "/api/login", {"username": "alex", "password": "falsch"}) check("Login falsches Passwort -> 401", st == 401, st) st, login = req("POST", "/api/login", {"username": "alex", "password": "geheim"}) check("Login korrekt", st == 200 and login.get("token"), login) token_a2 = login["token"] st, me = req("GET", "/api/me", token=token_a2) check("neuer Token gueltig", me["username"] == "alex", me) # --- Benutzer-Verwaltung (Owner) ------------------------------------- st, created = req("POST", "/api/users", {"username": "berta", "password": "geheim"}, token=token_a) check("Owner legt Benutzer an", st == 200 and created["username"] == "berta", created) berta_id = created["id"] st, users = req("GET", "/api/users", token=token_a) check("Owner listet Benutzer", st == 200 and len(users) == 2, users) st, login = req("POST", "/api/login", {"username": "berta", "password": "geheim"}) token_b = login["token"] st, me = req("GET", "/api/me", token=token_b) check("Berta: nicht Owner", me["username"] == "berta" and me["is_owner"] is False, me) st, _ = req("GET", "/api/users", token=token_b) check("Berta darf keine Benutzer sehen -> 403", st == 403, st) # --- Demo-Daten importieren (in Berta, VOR eigener Session) ---------- demo_path = os.path.join(BASE, "demo-data.json") demo = json.load(open(demo_path, encoding="utf-8")) st, imp = req("POST", "/api/import", {"format": "trainingsplan-export-v2", "data": json.dumps(demo)}, token=token_b) demo_sessions = len(demo["sessions"]) demo_gym = sum(1 for s in demo["sessions"] if s["activity"] == "gym") demo_ex_dates = len({s["date"] for s in demo["sessions"] for e in s.get("entries", []) if e.get("exercise") == "Push/Bankdrücken KH (je Hand)" and e.get("weight")}) check("Demo-Import in Berta", st == 200 and imp["created"] == demo_sessions, imp) # --- Berta trainiert selbst ------------------------------------------- st, sess_b = req("POST", "/api/sessions", {"activity": "gym", "day": "Push"}, token=token_b) req("PATCH", f"/api/sessions/{sess_b['id']}", {"sets": [{"exercise": "Push/Bankdrücken KH (je Hand)", "sort": 0, "weight": "40", "reps": "8", "done": 1}]}, token=token_b) req("PATCH", f"/api/sessions/{sess_b['id']}", {"meta": {"status": "done"}}, token=token_b) st, done_b = req("GET", "/api/sessions?status=done", token=token_b) st, done_a = req("GET", "/api/sessions?status=done", token=token_a) check("Isolation: B sieht alle Demo-Sessions + 1 eigene", len(done_b) == demo_sessions + 1, len(done_b)) check("Isolation: A sieht seine 1 Session (nicht B)", len(done_a) == 1, len(done_a)) st, rv = req("GET", "/api/reset-values?exercise=" + urllib.parse.quote("Push/Bankdrücken KH (je Hand)") + "&set=0", token=token_b) check("Reset-Werte sind pro Benutzer (eigene heutige Session)", rv.get("weight") == "40", rv) # Einstellungen pro Benutzer req("PUT", "/api/settings", {"theme": "light"}, token=token_a) st, s_a = req("GET", "/api/settings", token=token_a) st, s_b = req("GET", "/api/settings", token=token_b) check("Einstellungen pro Benutzer", s_a["theme"] == "light" and s_b["theme"] == "dark", (s_a, s_b)) # Owner setzt Sprache/Akzent -> Server-Standard fuer andere Benutzer req("PUT", "/api/settings", {"language": "en", "accent": "green"}, token=token_a) st, s_b = req("GET", "/api/settings", token=token_b) check("Owner-Sprache/Akzent wird Server-Standard", s_b["language"] == "en" and s_b["accent"] == "green", s_b) req("PUT", "/api/settings", {"language": "de", "accent": "orange", "theme": "dark"}, token=token_a) # --- Workouts-API ----------------------------------------------------- st, wo_b = req("GET", "/api/workouts", token=token_b) check("Workouts-API: alle Demo-Workouts + eigene", st == 200 and len(wo_b) == demo_sessions + 1, len(wo_b)) w = wo_b[0] # neueste Session = Berta heute check("Workouts-API: Shape", w["activity_name"] == "Gym (Kraft)" and w["day"] == "Push" and w["duration_min"] is not None and w["exercises"][0]["name"] == "Bankdrücken KH (je Hand)" and w["exercises"][0]["sets"][0]["weight"] == "40", w) st, wo_b = req("GET", "/api/workouts?activity=gym", token=token_b) check("Workouts nach Import (gym): Demo + 1 eigene", len(wo_b) == demo_gym + 1, len(wo_b)) st, wo_a = req("GET", "/api/workouts", token=token_a) check("A bleibt unberuehrt vom Import", len(wo_a) == 1, len(wo_a)) st, stats = req("GET", "/api/stats?exercise=" + urllib.parse.quote("Push/Bankdrücken KH (je Hand)"), token=token_b) check("Statistik aus Demo-Daten + eigene Session", st == 200 and len(stats) == demo_ex_dates + 1 and stats[0]["weight"] < stats[-1]["weight"] and stats[-1]["weight"] == 40, stats) # --- Passwort ändern / zurücksetzen ------------------------------------ st, _ = req("POST", "/api/password", {"current_password": "falsch", "new_password": "neu123"}, token=token_a) check("Eigenes Passwort: falsches aktuelles -> 400", st == 400, st) st, _ = req("POST", "/api/password", {"current_password": "geheim", "new_password": "neu123"}, token=token_a) check("Eigenes Passwort aendern", st == 200, st) st, _ = req("POST", "/api/login", {"username": "alex", "password": "geheim"}) check("altes Passwort ungueltig", st == 401, st) st, login2 = req("POST", "/api/login", {"username": "alex", "password": "neu123"}) check("neues Passwort funktioniert", st == 200, st) token_a = login2["token"] st, _ = req("PUT", f"/api/users/{berta_id}/password", {"password": "neu456"}, token=token_a) check("Owner setzt Passwort von B", st == 200, st) st, _ = req("GET", "/api/sessions", token=token_b) check("B: alter Token nach Passwort-Reset invalidiert", st == 401, st) st, _ = req("POST", "/api/login", {"username": "berta", "password": "geheim"}) check("B: altes Passwort weg", st == 401, st) st, login_b2 = req("POST", "/api/login", {"username": "berta", "password": "neu456"}) check("B: neues Passwort funktioniert", st == 200, st) token_b = login_b2["token"] st, _ = req("PUT", f"/api/users/{berta_id}/password", {"password": "x"}, token=token_a) check("Passwort zu kurz -> 400", st == 400, st) st, _ = req("PUT", f"/api/users/{berta_id}/password", {"password": "neu789"}, token=token_b) check("Nicht-Owner darf Passwort nicht setzen -> 403", st == 403, st) # --- Logout ------------------------------------------------------------- st, _ = req("POST", "/api/logout", token=token_b) st, _ = req("GET", "/api/sessions", token=token_b) check("Logout invalidiert Token -> 401", st == 401, st) # --- Benutzer loeschen --------------------------------------------------- st, r = req("DELETE", f"/api/users/{berta_id}", token=token_a) check("Owner loescht Benutzer", st == 200 and r["deleted"] == 1, r) st, login = req("POST", "/api/login", {"username": "berta", "password": "geheim"}) check("geloeschter Benutzer kann nicht mehr einloggen", st == 401, st) st, users = req("GET", "/api/users", token=token_a) check("Benutzerliste nach Loeschung", len(users) == 1, users) st, done_a = req("GET", "/api/sessions?status=done", token=token_a) check("A-Daten nach Loeschung von B intakt", len(done_a) == 1, len(done_a)) 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()