#!/usr/bin/env python3 """Small API client to test all Campus Safe endpoints. Usage examples: python scripts/api_test_client.py list --base-url http://127.0.0.1:8000 python scripts/api_test_client.py login --email admin@inphb.ci --password admin123 python scripts/api_test_client.py call --method GET --path /api/v1/postes/ --auth python scripts/api_test_client.py smoke --auth """ from __future__ import annotations import argparse import json import os import sys from dataclasses import dataclass from pathlib import Path from typing import Any import httpx @dataclass class Context: base_url: str timeout: float token: str | None @dataclass class RoleAccount: name: str email: str password: str SEED = { "user_admin_email": "admin@inphb.ci", "user_admin_password": os.getenv("CAMPUS_SAFE_TEST_PASSWORD", "admin123"), "user_manager_id": "750e8400-e29b-41d4-a716-446655440002", "user_admin_id": "750e8400-e29b-41d4-a716-446655440001", "signalement_id": "950e8400-e29b-41d4-a716-446655440001", "discussion_id": "c50e8400-e29b-41d4-a716-446655440001", "poste_id": "e50e8400-e29b-41d4-a716-446655440001", "commentaire_id": "f50e8400-e29b-41d4-a716-446655440001", "reaction_id": "a50e8400-e29b-41d4-a716-446655440001", "calendrier_id": "b50e8400-e29b-41d4-a716-446655440001", "participant_id": "c50e8400-e29b-41d4-a716-446655440001", "compte_id": "750e8400-e29b-41d4-a716-446655440003", } ROLE_ACCOUNTS: dict[str, RoleAccount] = { "admin": RoleAccount( name="admin", email=os.getenv("CAMPUS_SAFE_ADMIN_EMAIL", "admin@inphb.ci"), password=os.getenv("CAMPUS_SAFE_ADMIN_PASSWORD", "admin123"), ), "manager": RoleAccount( name="manager", email=os.getenv("CAMPUS_SAFE_MANAGER_EMAIL", "manager@inphb.ci"), password=os.getenv("CAMPUS_SAFE_MANAGER_PASSWORD", "staff123"), ), "user": RoleAccount( name="user", email=os.getenv("CAMPUS_SAFE_USER_EMAIL", "alice.bernard@inphb.ci"), password=os.getenv("CAMPUS_SAFE_USER_PASSWORD", "user123"), ), } RBAC_CASES: list[dict[str, Any]] = [ { "id": "roles_list", "method": "GET", "path": "/api/v1/roles/", "expected": {"admin": {200}, "manager": {403}, "user": {403}}, }, { "id": "stats_dashboard", "method": "GET", "path": "/api/v1/stats/dashboard", "expected": {"admin": {200}, "manager": {200}, "user": {403}}, }, { "id": "signalements_non_traites", "method": "GET", "path": "/api/v1/signalements/non-traites", "query": {"page": 0, "size": 5}, "expected": {"admin": {200}, "manager": {200}, "user": {403}}, }, { "id": "signalements_me", "method": "GET", "path": "/api/v1/signalements/me", "query": {"page": 0, "size": 5}, "expected": {"admin": {200}, "manager": {200}, "user": {200}}, }, { "id": "postes_epingle", "method": "PUT", "path": "/api/v1/postes/{poste_id}/epingle", "path_params": {"poste_id": SEED["poste_id"]}, "expected": {"admin": {200, 403}, "manager": {200}, "user": {403}}, }, { "id": "aide_faqs", "method": "GET", "path": "/api/v1/aide/faqs", "expected": {"admin": {200}, "manager": {200}, "user": {200}}, }, { "id": "calendriers_list", "method": "GET", "path": "/api/v1/calendriers/", "query": {"page": 0, "size": 3}, "expected": {"admin": {200}, "manager": {200}, "user": {200}}, }, { "id": "calendriers_participants", "method": "GET", "path": "/api/v1/calendriers/{evenement_id}/participants", "path_params": {"evenement_id": SEED["calendrier_id"]}, "query": {"page": 0, "size": 10}, "expected": {"admin": {200, 403}, "manager": {200}, "user": {403}}, }, ] # Optional presets used by smoke mode. PRESET_CASES: dict[tuple[str, str], dict[str, Any]] = { ("GET", "/api/v1/auth/validate"): {}, ("POST", "/api/v1/auth/refresh"): { "body": { "refreshToken": "__FROM_LOGIN__", "sessionId": "__FROM_LOGIN__", } }, ("POST", "/api/v1/auth/forgot-password"): {"body": {"email": SEED["user_admin_email"]}}, ("GET", "/api/v1/comptes/me"): {}, ("PUT", "/api/v1/comptes/me"): {"body": {"campus": "Campus Principal"}}, ("PUT", "/api/v1/comptes/me/password"): { "body": {"ancienMotDePasse": "admin123", "nouveauMotDePasse": "admin123"} }, ("GET", "/api/v1/comptes/"): {"query": {"page": 0, "size": 5}}, ("GET", "/api/v1/comptes/{compte_id}"): {"path": {"compte_id": SEED["compte_id"]}}, ("PUT", "/api/v1/comptes/{compte_id}"): { "path": {"compte_id": SEED["compte_id"]}, "body": {"filiere": "Informatique"}, }, ("PUT", "/api/v1/comptes/{compte_id}/reset-password"): { "path": {"compte_id": SEED["compte_id"]}, "body": {"nouveauMotDePasse": "user123"}, }, ("PUT", "/api/v1/comptes/{compte_id}/deactivate"): { "path": {"compte_id": "750e8400-e29b-41d4-a716-446655440006"} }, ("PUT", "/api/v1/comptes/{compte_id}/reactivate"): { "path": {"compte_id": "750e8400-e29b-41d4-a716-446655440006"} }, ("POST", "/api/v1/comptes/invitations"): { "body": {"email": "demo.invite@inphb.ci", "typeUtilisateur": "UTILISATEUR"} }, ("POST", "/api/v1/signalements/"): { "body": { "titre": "Test API client", "description": "Signalement de test", "type": "AUTRE", "modeIdentification": "NOMINATIF", } }, ("GET", "/api/v1/signalements/me"): {"query": {"page": 0, "size": 5}}, ("GET", "/api/v1/signalements/non-traites"): {"query": {"page": 0, "size": 5}}, ("GET", "/api/v1/signalements/{signalement_id}"): { "path": {"signalement_id": SEED["signalement_id"]} }, ("PUT", "/api/v1/signalements/{signalement_id}/statut"): { "path": {"signalement_id": SEED["signalement_id"]}, "body": {"statut": "EN_COURS"}, }, ("PUT", "/api/v1/signalements/{signalement_id}/assigner"): { "path": {"signalement_id": SEED["signalement_id"]}, "body": {"gestionnaireId": SEED["user_manager_id"]}, }, ("POST", "/api/v1/signalements/{signalement_id}/notes"): { "path": {"signalement_id": SEED["signalement_id"]}, "body": {"contenu": "Note de test via api_test_client"}, }, ("GET", "/api/v1/signalements/{signalement_id}/timeline"): { "path": {"signalement_id": SEED["signalement_id"]} }, ("POST", "/api/v1/discussions/"): { "body": {"titre": "Discussion test", "signalementId": SEED["signalement_id"]} }, ("GET", "/api/v1/discussions/"): {"query": {"page": 0, "size": 5}}, ("GET", "/api/v1/discussions/{discussion_id}"): { "path": {"discussion_id": SEED["discussion_id"]} }, ("POST", "/api/v1/messages/{discussion_id}/messages"): { "path": {"discussion_id": SEED["discussion_id"]}, "body": { "discussionId": SEED["discussion_id"], "contenu": "Message test", "type": "USER", "piecesJointes": [], }, }, ("GET", "/api/v1/messages/{discussion_id}/messages"): { "path": {"discussion_id": SEED["discussion_id"]}, "query": {"page": 0, "size": 20}, }, ("PUT", "/api/v1/messages/{discussion_id}/messages/lus"): { "path": {"discussion_id": SEED["discussion_id"]} }, ("POST", "/api/v1/calendriers/"): { "body": { "titre": "Atelier API test", "description": "Creation evenement test", "type": "ATELIER", "dateDebut": "2026-04-01T09:00:00", "dateFin": "2026-04-01T11:00:00", "lieu": "Salle A", "estEnLigne": False, "tags": ["test"], "publicCible": ["tous"], } }, ("GET", "/api/v1/calendriers/"): {"query": {"page": 0, "size": 3}}, ("GET", "/api/v1/calendriers/date/{target_date}"): { "path": {"target_date": "2025-03-20"} }, ("GET", "/api/v1/calendriers/range"): { "query": {"debut": "2025-03-01", "fin": "2025-04-30"} }, ("GET", "/api/v1/calendriers/{evenement_id}"): {"path": {"evenement_id": SEED["calendrier_id"]}}, ("PUT", "/api/v1/calendriers/{evenement_id}"): { "path": {"evenement_id": SEED["calendrier_id"]}, "body": {"titre": "Atelier MAJ API test"}, }, ("POST", "/api/v1/calendriers/{evenement_id}/inscription"): { "path": {"evenement_id": SEED["calendrier_id"]} }, ("DELETE", "/api/v1/calendriers/{evenement_id}/inscription"): { "path": {"evenement_id": SEED["calendrier_id"]} }, ("GET", "/api/v1/calendriers/{evenement_id}/participants"): { "path": {"evenement_id": SEED["calendrier_id"]}, "query": {"page": 0, "size": 10}, }, ("PUT", "/api/v1/calendriers/participants/{participant_id}/statut"): { "path": {"participant_id": SEED["participant_id"]}, "body": {"statut": "ASSISTE"}, }, ("GET", "/api/v1/postes/"): {"query": {"page": 0, "size": 10}}, ("POST", "/api/v1/postes/"): { "body": {"titre": "Post test", "contenu": "Contenu test", "categorie": "TEST", "tags": ["test"]} }, ("GET", "/api/v1/postes/{poste_id}"): {"path": {"poste_id": SEED["poste_id"]}}, ("PUT", "/api/v1/postes/{poste_id}/epingle"): {"path": {"poste_id": SEED["poste_id"]}}, ("DELETE", "/api/v1/postes/{poste_id}/epingle"): {"path": {"poste_id": SEED["poste_id"]}}, ("POST", "/api/v1/commentaires/{poste_id}/commentaires"): { "path": {"poste_id": SEED["poste_id"]}, "body": {"contenu": "Commentaire test"}, }, ("GET", "/api/v1/commentaires/{poste_id}/commentaires"): { "path": {"poste_id": SEED["poste_id"]}, "query": {"page": 0, "size": 20}, }, ("DELETE", "/api/v1/commentaires/{poste_id}/commentaires/{commentaire_id}"): { "path": {"poste_id": SEED["poste_id"], "commentaire_id": SEED["commentaire_id"]} }, ("POST", "/api/v1/reactions/{poste_id}/reactions"): { "path": {"poste_id": SEED["poste_id"]}, "body": {"type": "LIKE"}, }, ("GET", "/api/v1/reactions/{poste_id}/reactions"): {"path": {"poste_id": SEED["poste_id"]}}, ("DELETE", "/api/v1/reactions/{poste_id}/reactions/{reaction_id}"): { "path": {"poste_id": SEED["poste_id"], "reaction_id": SEED["reaction_id"]} }, ("GET", "/api/v1/preferences/"): {}, ("PUT", "/api/v1/preferences/"): {"body": {"theme": "light"}}, ("GET", "/api/v1/roles/"): {}, ("GET", "/api/v1/roles/{nom}/permissions"): {"path": {"nom": "ADMIN_SYSTEME"}}, ("GET", "/api/v1/stats/dashboard"): {}, ("GET", "/api/v1/stats/signalements"): {}, ("POST", "/api/v1/stats/signalements/export"): {"body": {"format": "csv"}}, ("GET", "/api/v1/aide/faqs"): {}, ("GET", "/api/v1/aide/contacts-urgence"): {}, } def _json_or_none(raw: str | None) -> dict[str, Any] | None: if not raw: return None return json.loads(raw) def _request( ctx: Context, method: str, path: str, *, path_params: dict[str, Any] | None = None, query: dict[str, Any] | None = None, body: dict[str, Any] | None = None, use_auth: bool = False, file_path: str | None = None, ) -> httpx.Response: path_params = path_params or {} full_path = path.format(**path_params) url = f"{ctx.base_url.rstrip('/')}{full_path}" headers: dict[str, str] = {} if use_auth and ctx.token: headers["Authorization"] = f"Bearer {ctx.token}" with httpx.Client(timeout=ctx.timeout) as client: if file_path: file_name = Path(file_path).name with open(file_path, "rb") as f: files = {"file": (file_name, f)} return client.request(method, url, params=query, headers=headers, files=files) return client.request(method, url, params=query, headers=headers, json=body) def _login(base_url: str, email: str, password: str, timeout: float) -> tuple[str, dict[str, Any]]: with httpx.Client(timeout=timeout) as client: r = client.post( f"{base_url.rstrip('/')}/api/v1/auth/login", json={"email": email, "motDePasse": password}, ) r.raise_for_status() payload = r.json() data = payload.get("data", {}) token = data.get("token") if not token: raise RuntimeError("Login succeeded but token not found in response") return token, data def _openapi(base_url: str, timeout: float) -> dict[str, Any]: with httpx.Client(timeout=timeout) as client: for candidate in ("/openapi.json", "/api/v1/openapi.json"): r = client.get(f"{base_url.rstrip('/')}{candidate}") if r.status_code == 200: return r.json() r.raise_for_status() return r.json() def cmd_list(args: argparse.Namespace) -> int: spec = _openapi(args.base_url, args.timeout) paths = spec.get("paths", {}) count = 0 for path, methods in sorted(paths.items()): for method in sorted(methods.keys()): print(f"{method.upper():6} {path}") count += 1 print(f"\nTotal operations: {count}") return 0 def cmd_login(args: argparse.Namespace) -> int: token, data = _login(args.base_url, args.email, args.password, args.timeout) print("Login OK") print("token:", token) print("sessionId:", data.get("sessionId")) print("refreshToken:", data.get("refreshToken")) return 0 def cmd_call(args: argparse.Namespace) -> int: token = args.token if args.auth and not token: if not args.password: raise RuntimeError("Missing password. Use --password or CAMPUS_SAFE_TEST_PASSWORD") token, _ = _login(args.base_url, args.email, args.password, args.timeout) ctx = Context(base_url=args.base_url, timeout=args.timeout, token=token) r = _request( ctx, args.method.upper(), args.path, path_params=_json_or_none(args.path_params), query=_json_or_none(args.query), body=_json_or_none(args.body), use_auth=args.auth, file_path=args.file, ) print(f"{args.method.upper()} {args.path} -> {r.status_code}") text = r.text if len(text) > 2000: text = text[:2000] + "\n... (truncated)" print(text) return 0 def cmd_smoke(args: argparse.Namespace) -> int: token = args.token login_data: dict[str, Any] = {} if args.auth and not token: if not args.password: raise RuntimeError("Missing password. Use --password or CAMPUS_SAFE_TEST_PASSWORD") token, login_data = _login(args.base_url, args.email, args.password, args.timeout) ctx = Context(base_url=args.base_url, timeout=args.timeout, token=token) spec = _openapi(args.base_url, args.timeout) # Inject refresh token/session from login to exercise refresh endpoint. if login_data: refresh_case = PRESET_CASES.get(("POST", "/api/v1/auth/refresh")) if refresh_case and "body" in refresh_case: refresh_case["body"]["refreshToken"] = login_data.get("refreshToken") refresh_case["body"]["sessionId"] = login_data.get("sessionId") ok = 0 ko = 0 skipped = 0 for path, methods in sorted(spec.get("paths", {}).items()): for method in sorted(methods.keys()): m = method.upper() key = (m, path) preset = PRESET_CASES.get(key) if preset is None: print(f"SKIP {m:6} {path} (no preset, use 'call' command)") skipped += 1 continue try: r = _request( ctx, m, path, path_params=preset.get("path"), query=preset.get("query"), body=preset.get("body"), use_auth=args.auth, file_path=preset.get("file"), ) # Consider auth/validation conflicts as a tested operation too. tested_ok = r.status_code < 500 mark = "OK " if tested_ok else "FAIL" print(f"{mark} {m:6} {path} -> {r.status_code}") # Refresh invalidates the previous access token; keep testing with the new one. if key == ("POST", "/api/v1/auth/refresh") and r.status_code < 400: try: payload = r.json() new_token = payload.get("data", {}).get("token") if new_token: ctx.token = new_token except Exception: pass if tested_ok: ok += 1 else: ko += 1 except Exception as exc: print(f"FAIL {m:6} {path} -> EXC: {exc}") ko += 1 print("\nSummary") print(f" ok: {ok}") print(f" fail: {ko}") print(f" skipped: {skipped}") return 1 if ko else 0 def _resolve_roles(selected: list[str] | None) -> list[str]: if not selected: return ["admin", "manager", "user"] unknown = [r for r in selected if r not in ROLE_ACCOUNTS] if unknown: raise RuntimeError(f"Unknown roles: {', '.join(unknown)}") return selected def _print_rbac_row(case_id: str, role_status: dict[str, int], role_pass: dict[str, bool]) -> None: parts = [f"{case_id:28}"] for role in ("admin", "manager", "user"): if role in role_status: status = role_status[role] mark = "PASS" if role_pass[role] else "FAIL" parts.append(f"{role}:{status}({mark})") print(" | ".join(parts)) def cmd_multirole(args: argparse.Namespace) -> int: roles = _resolve_roles(args.roles) # Login each role once and reuse the token. contexts: dict[str, Context] = {} for role in roles: account = ROLE_ACCOUNTS[role] token, _ = _login(args.base_url, account.email, account.password, args.timeout) contexts[role] = Context(base_url=args.base_url, timeout=args.timeout, token=token) print("RBAC matrix (expected access control per role)") print("case | statuses") total_checks = 0 failed_checks = 0 failures: list[tuple[str, str, int, str]] = [] for case in RBAC_CASES: role_status: dict[str, int] = {} role_pass: dict[str, bool] = {} for role in roles: ctx = contexts[role] expected = set(case["expected"].get(role, [])) response = _request( ctx, case["method"], case["path"], path_params=case.get("path_params"), query=case.get("query"), body=case.get("body"), use_auth=True, ) role_status[role] = response.status_code role_pass[role] = response.status_code in expected total_checks += 1 if not role_pass[role]: failed_checks += 1 snippet = response.text.replace("\n", " ").strip() if len(snippet) > 220: snippet = snippet[:220] + "..." failures.append((case["id"], role, response.status_code, snippet)) _print_rbac_row(case["id"], role_status, role_pass) print("\nSummary") print(f" checks: {total_checks}") print(f" failed: {failed_checks}") print(f" passed: {total_checks - failed_checks}") if failures: print("\nFailures detail") for case_id, role, status, snippet in failures: print(f" - {case_id} | {role} | {status} | {snippet}") return 1 if failed_checks else 0 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="Campus Safe API test client") p.add_argument("--base-url", default="http://127.0.0.1:8000") p.add_argument("--timeout", type=float, default=20.0) sub = p.add_subparsers(dest="cmd", required=True) p_list = sub.add_parser("list", help="List endpoints from OpenAPI") p_list.set_defaults(func=cmd_list) p_login = sub.add_parser("login", help="Login and print token") p_login.add_argument("--email", default=SEED["user_admin_email"]) p_login.add_argument("--password", default=SEED["user_admin_password"]) p_login.set_defaults(func=cmd_login) p_call = sub.add_parser("call", help="Call a specific endpoint") p_call.add_argument("--method", required=True, help="GET/POST/PUT/DELETE") p_call.add_argument("--path", required=True, help="OpenAPI path, e.g. /api/v1/postes/{poste_id}") p_call.add_argument("--path-params", help='JSON for path params, e.g. {"poste_id":"..."}') p_call.add_argument("--query", help='JSON query params, e.g. {"page":0,"size":20}') p_call.add_argument("--body", help='JSON body payload') p_call.add_argument("--file", help="Path to upload file (multipart form under key 'file')") p_call.add_argument("--auth", action="store_true", help="Attach Bearer token") p_call.add_argument("--token", help="Bearer token (optional if --auth + login creds)") p_call.add_argument("--email", default=SEED["user_admin_email"]) p_call.add_argument("--password", default=SEED["user_admin_password"]) p_call.set_defaults(func=cmd_call) p_smoke = sub.add_parser("smoke", help="Run preset smoke test over OpenAPI routes") p_smoke.add_argument("--auth", action="store_true", help="Attach Bearer token") p_smoke.add_argument("--token", help="Bearer token (optional if --auth + login creds)") p_smoke.add_argument("--email", default=SEED["user_admin_email"]) p_smoke.add_argument("--password", default=SEED["user_admin_password"]) p_smoke.set_defaults(func=cmd_smoke) p_multirole = sub.add_parser("multirole", help="Run RBAC matrix test across multiple roles") p_multirole.add_argument( "--roles", nargs="*", choices=["admin", "manager", "user"], help="Subset of roles to test (default: admin manager user)", ) p_multirole.set_defaults(func=cmd_multirole) return p def main() -> int: parser = build_parser() args = parser.parse_args() return int(args.func(args)) if __name__ == "__main__": sys.exit(main())