modbison/scripts/run_checkpoint_tests.py

#!/usr/bin/env python3
"""Run the full automated test inventory for a Package Checkpoint review.

See docs/testing/test_inventory.md for what each suite actually covers (and
doesn't) -- keep that doc in sync as tests are added, this script just runs
whatever's there.

Usage:
    python scripts/run_checkpoint_tests.py [--keep-postgres] [--postgres-port PORT]

Exits 0 if everything that ran passed, 1 otherwise. Suites with nothing to
run (e.g. no scripts/smoke_test_*.gd yet, or no Godot binary found) are
reported as SKIPPED, not counted as a failure.
"""
import argparse
import glob
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
BACKEND = ROOT / "backend"


def run(cmd, cwd=None, env=None, capture=False):
    print(f"$ {' '.join(cmd)}" + (f"  (cwd={cwd})" if cwd else ""))
    return subprocess.run(cmd, cwd=cwd, env=env, capture_output=capture, text=True)


def find_godot():
    for candidate in ("godot", str(Path.home() / "godot" / "godot")):
        if shutil.which(candidate) or Path(candidate).is_file():
            return candidate
    return None


def wait_for_http(url: str, timeout_s: int = 20) -> bool:
    import urllib.request
    import urllib.error

    deadline = time.time() + timeout_s
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(url, timeout=2) as resp:
                if resp.status == 200:
                    return True
        except (urllib.error.URLError, ConnectionError, OSError):
            pass
        time.sleep(1)
    return False


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--keep-postgres", action="store_true",
                         help="Don't tear down the checkpoint Postgres container afterward")
    parser.add_argument("--postgres-port", default=os.environ.get("CHECKPOINT_POSTGRES_PORT", "5432"),
                         help="Host port for the checkpoint's own Postgres container (default 5432 -- "
                              "override if that's already in use on this machine)")
    parser.add_argument("--backend-port", default=os.environ.get("CHECKPOINT_BACKEND_PORT", "18080"),
                         help="Port to run the backend server on for the Godot smoke tests")
    args = parser.parse_args()

    results = {}

    # --- Backend: bring up Postgres ---
    compose_env = os.environ.copy()
    compose_env["COMPOSE_PROJECT_NAME"] = "agp-checkpoint"
    compose_env["POSTGRES_HOST_PORT"] = args.postgres_port
    compose_env["POSTGRES_PORT"] = args.postgres_port  # some branches use this name instead, see tools/definitions.py
    print("=== Backend: bringing up Postgres ===")
    run(["docker", "compose", "up", "-d", "postgres"], cwd=BACKEND, env=compose_env)

    print("Waiting for Postgres to be ready...")
    pg_ready = False
    for _ in range(20):
        r = run(["docker", "compose", "exec", "-T", "postgres", "pg_isready", "-U", "agp_user", "-d", "agp_db"],
                cwd=BACKEND, env=compose_env, capture=True)
        if r.returncode == 0:
            pg_ready = True
            break
        time.sleep(1)
    if not pg_ready:
        print("Postgres never became ready -- aborting.")
        sys.exit(1)

    db_url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{args.postgres_port}/agp_db"
    backend_env = os.environ.copy()
    backend_env["DATABASE_URL"] = db_url

    print("\n=== Backend: alembic upgrade head ===")
    r = run([sys.executable, "-m", "alembic", "upgrade", "head"], cwd=BACKEND, env=backend_env)
    results["migrations (fresh DB)"] = r.returncode == 0

    print("\n=== Backend: pytest ===")
    r = run([sys.executable, "-m", "pytest", "-v"], cwd=BACKEND, env=backend_env)
    results["backend pytest"] = r.returncode == 0

    # --- Godot: headless smoke tests (need a live backend server) ---
    godot_bin = find_godot()
    smoke_tests = sorted(glob.glob(str(ROOT / "scripts" / "smoke_test_*.gd")))

    if not godot_bin:
        print("\n=== Godot: binary not found (checked PATH and ~/godot/godot), skipping smoke tests ===")
        results["godot smoke tests"] = None
    elif not smoke_tests:
        print("\n=== Godot: no scripts/smoke_test_*.gd found yet, nothing to run ===")
        results["godot smoke tests"] = None
    else:
        print(f"\n=== Backend: starting server on :{args.backend_port} for Godot smoke tests ===")
        server_proc = subprocess.Popen(
            [sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", args.backend_port],
            cwd=BACKEND, env=backend_env,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        )
        try:
            base_url = f"http://127.0.0.1:{args.backend_port}"
            if not wait_for_http(f"{base_url}/api/health"):
                print("Backend server never became healthy -- skipping Godot smoke tests.")
                results["godot smoke tests"] = False
            else:
                godot_env = os.environ.copy()
                godot_env["AGP_API_BASE_URL"] = base_url

                print("=== Godot: import pass (builds the class cache) ===")
                run([godot_bin, "--headless", "--path", str(ROOT), "--import"], env=godot_env)
                run([godot_bin, "--headless", "--path", str(ROOT), "--quit"], env=godot_env)

                godot_ok = True
                for test in smoke_tests:
                    rel = os.path.relpath(test, ROOT)
                    print(f"\n--- {rel} ---")
                    # smoke_test_network_client.gd needs a separate dedicated
                    # server process listening on ENet 7777 -- stand one up
                    # just for that test and tear it down after.
                    game_server = None
                    if os.path.basename(rel) == "smoke_test_network_client.gd":
                        print("(starting scenes/server.tscn for the network client test)")
                        game_server = subprocess.Popen(
                            [godot_bin, "--headless", "--path", str(ROOT), "res://scenes/server.tscn"],
                            env=godot_env,
                        )
                        time.sleep(3)
                    try:
                        r = run([godot_bin, "--headless", "--path", str(ROOT), "--script", f"res://{rel}"], env=godot_env)
                        if r.returncode != 0:
                            godot_ok = False
                    finally:
                        if game_server is not None:
                            game_server.terminate()
                            try:
                                game_server.wait(timeout=10)
                            except subprocess.TimeoutExpired:
                                game_server.kill()
                results["godot smoke tests"] = godot_ok

                # 2-player multiplayer test (its own shell harness: server + 2
                # real client processes).
                mp_script = ROOT / "scripts" / "run_multiplayer_2p_test.sh"
                if mp_script.exists():
                    print("\n--- scripts/run_multiplayer_2p_test.sh ---")
                    mp_env = os.environ.copy()
                    mp_env["GODOT_BIN"] = godot_bin
                    r = run(["bash", str(mp_script)], env=mp_env)
                    results["multiplayer 2p test"] = r.returncode == 0
        finally:
            server_proc.terminate()
            try:
                server_proc.wait(timeout=10)
            except subprocess.TimeoutExpired:
                server_proc.kill()

    if not args.keep_postgres:
        print("\n=== Backend: tearing down checkpoint Postgres ===")
        run(["docker", "compose", "down", "-v", "postgres"], cwd=BACKEND, env=compose_env)

    # --- Summary ---
    print("\n" + "=" * 50)
    print("SUMMARY")
    print("=" * 50)
    overall_ok = True
    for name, ok in results.items():
        if ok is None:
            print(f"  {name}: SKIPPED")
        elif ok:
            print(f"  {name}: PASS")
        else:
            print(f"  {name}: FAIL")
            overall_ok = False

    print("\nSee docs/testing/test_inventory.md for what each suite actually covers (and doesn't).")
    sys.exit(0 if overall_ok else 1)


if __name__ == "__main__":
    main()