launcher/run_test.py

#!/usr/bin/env python3
import subprocess, sys, os, json, time, signal

os.chdir(os.path.join(os.path.dirname(__file__), "server"))
workdir = os.path.join(os.path.dirname(__file__), "server")

# Start server
print("Starting server...", flush=True)
proc = subprocess.Popen(
    [sys.executable, "-m", "uvicorn", "main:app",
     "--host", "127.0.0.1", "--port", "1590",
     "--log-level", "warning"],
    cwd=workdir,
    stdout=subprocess.PIPE, stderr=subprocess.PIPE,
)

import httpx

for i in range(30):
    try:
        r = httpx.get("http://127.0.0.1:1590/news", timeout=2)
        if r.status_code == 200:
            print("Server ready!", flush=True)
            break
    except Exception as e:
        if i == 0:
            print(f"Waiting for server ({e})...", flush=True)
    time.sleep(1)
else:
    print("Server failed to start", flush=True)
    proc.terminate()
    proc.wait(timeout=5)
    sys.exit(1)

passed = 0
failed = 0
def check(name, ok, detail=""):
    global passed, failed
    if ok:
        print(f"  PASS  {name}")
        passed += 1
    else:
        print(f"  FAIL  {name}  {detail}")
        failed += 1

BASE = "http://127.0.0.1:1590"
client = httpx.Client(base_url=BASE, timeout=10)

# ===== NEWS =====
print("\n=== NEWS ===")
r = client.get("/news")
check("GET /news status", r.status_code == 200, str(r.status_code))
data = r.json()
check("news has 'news' key", "news" in data)
check("news is array", isinstance(data.get("news"), list))
check("news has items", len(data["news"]) > 0, f"got {len(data['news'])} items")
if data.get("news"):
    n = data["news"][0]
    check("news item has title", "title" in n)
    check("news item has body", "body" in n)
    check("news item has type", "type" in n)

# ===== AUTH =====
print("\n=== AUTH ===")
ts = str(int(time.time()))
tu = f"test_{ts}"
r = client.post("/auth/register", json={"username": tu, "password": "pass123"})
check("register new user", r.status_code == 200, str(r.status_code))
reg = r.json()
check("register returns access_token", "access_token" in reg)

r2 = client.post("/auth/register", json={"username": tu, "password": "pass123"})
check("register existing returns 409", r2.status_code == 409, str(r2.status_code))

r3 = client.post("/auth/login", json={"username": tu, "password": "pass123"})
check("login valid", r3.status_code == 200, str(r3.status_code))
tok = r3.json().get("access_token", "")
if not tok: tok = reg.get("access_token", "")
check("login returns token", bool(tok))

r4 = client.post("/auth/login", json={"username": tu, "password": "wrongpass123"})
check("login wrong pass", r4.status_code == 401, str(r4.status_code))

r5 = client.post("/auth/login", json={"username": "nobody_valid", "password": "pass123456"})
check("login nonexistent", r5.status_code == 401, str(r5.status_code))

# ===== TOKEN VALIDATION =====
print("\n=== TOKEN ===")
ah = {"Authorization": f"Bearer {tok}"}
authed = httpx.Client(base_url=BASE, timeout=10, headers=ah)
r = authed.get("/auth/pass/my")
check("pass/my authed", r.status_code in (200, 404), str(r.status_code))

r2 = client.get("/auth/pass/my")
check("pass/my unauthed", r2.status_code == 401, str(r2.status_code))

# ===== PACKS =====
print("\n=== PACKS ===")
r = client.get("/packs")
check("packs unauthed rejects", r.status_code in (401, 403), str(r.status_code))
r2 = authed.get("/packs")
check("packs authed succeeds", r2.status_code in (200, 403), str(r2.status_code))
if r2.status_code == 200:
    pdata = r2.json()
    check("packs has 'packs' key", "packs" in pdata)

# ===== FRIENDS =====
print("\n=== FRIENDS ===")
r = client.get("/api/friends/list")
check("friends/list unauthed", r.status_code == 401, str(r.status_code))
r = authed.get("/api/friends/list")
check("friends/list authed", r.status_code == 200, str(r.status_code))
check("friends/list has friends key", "friends" in r.json())

r = authed.get("/api/friends/requests")
check("friends/requests authed", r.status_code == 200, str(r.status_code))
check("friends/requests has requests key", "requests" in r.json())

# Register friend user
fts = f"friend_{ts}"
r = client.post("/auth/register", json={"username": fts, "password": "pass123"})
check("register friend user", r.status_code == 200, str(r.status_code))
ftok = r.json().get("access_token", "")
fauthed = httpx.Client(base_url=BASE, timeout=10,
                       headers={"Authorization": f"Bearer {ftok}"})
# Parse user id from JWT (no account endpoint on Python server)
import base64
def parse_user_id(token):
    try:
        payload_b64 = token.split('.')[1]
        payload_b64 += '=' * (4 - len(payload_b64) % 4)
        return json.loads(base64.urlsafe_b64decode(payload_b64))["sub"]
    except:
        return 0
friend_id = parse_user_id(ftok)
my_id = parse_user_id(tok)

# Add friend
r = authed.post("/api/friends/add", json={"username": fts})
check("friends add", r.status_code == 200, f"{r.status_code}: {r.text[:80]}")
ad = r.json()
check("friends add returns message", "message" in ad)

# Duplicate add
r = authed.post("/api/friends/add", json={"username": fts})
check("friends add duplicate", r.status_code == 400, str(r.status_code))

# Add self
r = authed.post("/api/friends/add", json={"username": tu})
check("friends add self", r.status_code == 400, str(r.status_code))

# Add nonexistent
r = authed.post("/api/friends/add", json={"username": "nobody_xyz_123"})
check("friends add nonexistent", r.status_code == 404, str(r.status_code))

# Accept friend request (as friend)
r = fauthed.post("/api/friends/accept", json={"user_id": my_id})
check("friends accept", r.status_code == 200, f"{r.status_code}: {r.text[:80]}")

# Remove nonexistent
r = authed.post("/api/friends/remove", json={"user_id": 99999})
check("friends remove nonexistent", r.status_code == 404, str(r.status_code))

# Remove friend
r = authed.post("/api/friends/remove", json={"user_id": friend_id})
check("friends remove", r.status_code == 200, f"{r.status_code}: {r.text[:80]}")

# ===== PLAYTIME =====
print("\n=== PLAYTIME ===")
r = authed.post("/api/playtime/sync", json={"pack_name": "test_pack", "minutes": 30})
check("playtime sync", r.status_code == 200, f"{r.status_code}: {r.text[:80]}")

r = authed.get("/api/playtime/stats")
check("playtime stats", r.status_code == 200, str(r.status_code))
st = r.json()
check("playtime has total_minutes", "total_minutes" in st)
check("playtime has packs", "packs" in st)

r = client.get("/api/playtime/stats")
check("playtime stats unauthed", r.status_code == 401, str(r.status_code))

# ===== SUMMARY =====
total = passed + failed
print(f"\n{'='*40}")
print(f"RESULTS: {passed}/{total} PASSED, {failed}/{total} FAILED")
if failed > 0:
    print("SOME TESTS FAILED!")

proc.terminate()
proc.wait(timeout=5)
sys.exit(1 if failed else 0)