#!/usr/bin/env python3 """Test dell'API della classifica: avvia il servizio su un database temporaneo (nessun Docker richiesto, gira con python3) e verifica i casi limite. python3 tools/test_scores_api.py """ import json import os import shutil import socket import subprocess import sys import tempfile import time import urllib.error import urllib.request import uuid from pathlib import Path ROOT = Path(__file__).resolve().parents[1] APP = ROOT / "services/scores-api/app.py" PASSED = [] def check(label: str, condition: bool, detail: str = "") -> None: if condition: PASSED.append(label) print("PASS: %s" % label) else: print("FAIL: %s %s" % (label, detail)) raise SystemExit(1) def free_port() -> int: with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) return sock.getsockname()[1] def call(url: str, method: str = "GET", payload=None, headers=None, raw=None, timeout=5.0): data = raw if raw is not None else (json.dumps(payload).encode() if payload is not None else None) request = urllib.request.Request(url, data=data, method=method) for key, value in (headers or {}).items(): request.add_header(key, value) try: with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read().decode() return response.status, (json.loads(body) if body else {}) except urllib.error.HTTPError as error: body = error.read().decode() try: return error.code, json.loads(body) if body else {} except json.JSONDecodeError: return error.code, {"raw": body} def start_service(db_path: Path, port: int, extra_env=None): env = dict(os.environ, DATABASE_PATH=str(db_path), PORT=str(port), BIND="127.0.0.1") env.update(extra_env or {}) process = subprocess.Popen([sys.executable, str(APP)], env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) base = "http://127.0.0.1:%d" % port for _ in range(100): try: if call(base + "/health", timeout=0.5)[0] == 200: return process, base except Exception: time.sleep(0.1) process.kill() raise SystemExit("il servizio non si e' avviato:\n" + process.stdout.read()) def stop(process) -> None: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() def submission(player: str, name: str, score: int, submission_id: str | None = None) -> dict: return {"player_id": player, "name": name, "score": score, "submission_id": submission_id or str(uuid.uuid4())} def main() -> None: workdir = Path(tempfile.mkdtemp(prefix="zombie-scores-")) db = workdir / "scores.db" port = free_port() fast = {"POST_PER_HOUR": "1000", "POST_MIN_INTERVAL": "0.3", "GET_PER_MINUTE": "1000"} process, base = start_service(db, port, fast) json_header = {"Content-Type": "application/json"} try: # --- salute e classifica vuota --- status, health = call(base + "/health") check("health risponde 200 con nome e versione", status == 200 and health["ok"] and health["service"] == "zombie-fps-scores" and health["rows"] == 0) status, board = call(base + "/scores") check("classifica vuota valida", status == 200 and board["top"] == [] and board["total"] == 0 and board["me"] is None) # --- invio valido --- alice, bob = "player-alice-0001", "player-bob-0002" first_payload = submission(alice, "Alice", 1250) status, first = call(base + "/scores", "POST", first_payload, json_header) check("invio valido risponde 201 con id, posizione e totale", status == 201 and first["ok"] and first["rank"] == 1 and first["total"] == 1 and len(first["id"]) == 36 and not first["duplicate"]) identifier = first["id"] # --- idempotenza: il retry dello STESSO invio non duplica e non e' bloccato --- # (caso reale: il client ritenta dopo un timeout, subito dopo il primo invio) status, again = call(base + "/scores", "POST", first_payload, json_header) check("retry con lo stesso submission_id e' idempotente", status == 200 and again["duplicate"] and again["id"] == identifier and again["total"] == 1) # --- un id diverso, stesso giocatore: seconda partita --- time.sleep(0.35) # intervallo minimo fra invii dello stesso giocatore status, second = call(base + "/scores", "POST", submission(alice, "Alice", 800), json_header) check("seconda partita dello stesso giocatore viene salvata", status == 201 and second["total"] == 2 and second["rank"] == 2) # --- classifica: tutte le partite, ordinate --- time.sleep(0.35) call(base + "/scores", "POST", submission(bob, "Bob", 3000), json_header) status, board = call(base + "/scores?limit=10&player=" + alice) names = [row["name"] for row in board["top"]] scores = [row["score"] for row in board["top"]] ranks = [row["rank"] for row in board["top"]] check("classifica mostra tutte le partite in ordine di punteggio", status == 200 and names == ["Bob", "Alice", "Alice"] and scores == [3000, 1250, 800] and ranks == [1, 2, 3]) check("il riassunto personale riporta miglior punteggio, posizione e partite", board["me"]["best"] == 1250 and board["me"]["rank"] == 2 and board["me"]["games"] == 2) status, limited = call(base + "/scores?limit=2") check("limit viene rispettato", len(limited["top"]) == 2 and limited["limit"] == 2) status, huge = call(base + "/scores?limit=9999") check("limit viene limitato al massimo", huge["limit"] == 50) status, bad_limit = call(base + "/scores?limit=abc") check("limit non numerico viene rifiutato", status == 400 and "limit" in bad_limit["error"]) # --- validazione --- cases = [ ("nome vuoto rifiutato", submission(alice, " ", 100), 400), ("nome troppo lungo rifiutato", submission(alice, "x" * 17, 100), 400), ("caratteri strani nel nome rifiutati", submission(alice, "Anna