SashegDev
192 lines
7.0 KiB
Python
192 lines
7.0 KiB
Python
import sys, os, json, time, subprocess, signal
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'server'))
|
|
|
|
import httpx
|
|
from pathlib import Path
|
|
|
|
BASE = "http://127.0.0.1:1589"
|
|
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
|
|
|
|
def start_server():
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = os.path.join(os.path.dirname(__file__), "server")
|
|
proc = subprocess.Popen(
|
|
[sys.executable, "-m", "uvicorn", "main:app", "--host", "127.0.0.1", "--port", "1589"],
|
|
cwd=os.path.join(os.path.dirname(__file__), "server"),
|
|
env=env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
for i in range(30):
|
|
try:
|
|
r = httpx.get(f"{BASE}/news", timeout=2)
|
|
if r.status_code == 200:
|
|
return proc
|
|
except:
|
|
time.sleep(1)
|
|
return proc
|
|
|
|
proc = None
|
|
try:
|
|
print("Starting server...")
|
|
proc = start_server()
|
|
print("Server started\n")
|
|
client = httpx.Client(base_url=BASE, timeout=10)
|
|
|
|
# ---------- NEWS ----------
|
|
print("=== NEWS ===")
|
|
r = client.get("/news")
|
|
check("GET /news", r.status_code == 200 and "news" in r.json())
|
|
data = r.json()
|
|
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["news"]:
|
|
first = data["news"][0]
|
|
check("news item has title", "title" in first)
|
|
check("news item has body", "body" in first)
|
|
check("news item has type", "type" in first)
|
|
|
|
# ---------- AUTH ----------
|
|
print("\n=== AUTH ===")
|
|
test_user = f"test_{int(time.time())}"
|
|
test_pass = "test123"
|
|
r = client.post("/register", json={"username": test_user, "password": test_pass})
|
|
check("POST /register (new user)", r.status_code == 200)
|
|
reg_data = r.json()
|
|
check("register returns token", "access_token" in reg_data)
|
|
|
|
r2 = client.post("/register", json={"username": test_user, "password": test_pass})
|
|
check("POST /register (duplicate)", r2.status_code == 200)
|
|
|
|
r3 = client.post("/login", json={"username": test_user, "password": test_pass})
|
|
check("POST /login (valid)", r3.status_code == 200)
|
|
login_data = r3.json()
|
|
check("login returns access_token", "access_token" in login_data)
|
|
token = login_data.get("access_token", "")
|
|
|
|
r4 = client.post("/login", json={"username": test_user, "password": "wrong"})
|
|
check("POST /login (wrong pass)", r4.status_code == 401)
|
|
|
|
r5 = client.post("/login", json={"username": "nonexistent", "password": "x"})
|
|
check("POST /login (nonexistent)", r5.status_code == 401)
|
|
|
|
authed = httpx.Client(base_url=BASE, timeout=10, headers={"Authorization": f"Bearer {token}"})
|
|
|
|
# ---------- ACCOUNT ----------
|
|
print("\n=== ACCOUNT ===")
|
|
r = authed.get("/account")
|
|
check("GET /account (authed)", r.status_code == 200)
|
|
acct = r.json()
|
|
check("account has username", acct.get("username") == test_user,
|
|
f"got {acct.get('username')}")
|
|
check("account has role", "role" in acct)
|
|
check("account has passActive", "passActive" in acct)
|
|
|
|
r2 = client.get("/account")
|
|
check("GET /account (unauthed)", r2.status_code == 401)
|
|
|
|
# ---------- PACKS ----------
|
|
print("\n=== PACKS ===")
|
|
r = client.get("/packs")
|
|
check("GET /packs (unauthed)", r.status_code == 401 or r.status_code == 403,
|
|
f"got {r.status_code}")
|
|
|
|
r2 = authed.get("/packs")
|
|
check("GET /packs (authed)", r2.status_code in (200, 403),
|
|
f"got {r2.status_code}")
|
|
if r2.status_code == 200:
|
|
packs = r2.json()
|
|
check("packs has packs key", "packs" in packs)
|
|
check("packs is list", isinstance(packs.get("packs"), list))
|
|
|
|
# ---------- FRIENDS (no auth) ----------
|
|
print("\n=== FRIENDS ===")
|
|
r = client.get("/api/friends/list")
|
|
check("GET /friends/list (unauthed)", r.status_code == 401)
|
|
|
|
r2 = authed.get("/api/friends/list")
|
|
check("GET /friends/list (authed)", r2.status_code == 200)
|
|
flist = r2.json()
|
|
check("friends list has friends key", "friends" in flist)
|
|
|
|
r3 = authed.get("/api/friends/requests")
|
|
check("GET /friends/requests (authed)", r3.status_code == 200)
|
|
freq = r3.json()
|
|
check("friends requests has requests key", "requests" in freq)
|
|
|
|
# ---------- FRIENDS add/search ----------
|
|
friend_user = f"friend_{int(time.time())}"
|
|
r = client.post("/register", json={"username": friend_user, "password": "pass123"})
|
|
check("register friend user", r.status_code == 200)
|
|
friend_token = r.json().get("access_token", "")
|
|
friend_authed = httpx.Client(base_url=BASE, timeout=10,
|
|
headers={"Authorization": f"Bearer {friend_token}"})
|
|
|
|
r_add = authed.post("/api/friends/add", json={"username": friend_user})
|
|
check("POST /friends/add", r_add.status_code == 200,
|
|
f"got {r_add.status_code}: {r_add.text[:100]}")
|
|
|
|
r_dup = authed.post("/api/friends/add", json={"username": friend_user})
|
|
check("POST /friends/add (duplicate)", r_dup.status_code == 400)
|
|
|
|
r_self = authed.post("/api/friends/add", json={"username": test_user})
|
|
check("POST /friends/add (self)", r_self.status_code == 400)
|
|
|
|
r_nonexist = authed.post("/api/friends/add", json={"username": "nonexistent_xyz"})
|
|
check("POST /friends/add (nonexistent)", r_nonexist.status_code == 404)
|
|
|
|
# Friend accepts
|
|
r_accept = friend_authed.post("/api/friends/accept", json={"user_id": acct.get("id", 0)})
|
|
check("POST /friends/accept", r_accept.status_code in (200, 404),
|
|
f"got {r_accept.status_code}: {r_accept.text[:100]}")
|
|
|
|
# Check friends list now shows each other
|
|
r_list = authed.get("/api/friends/list")
|
|
check("GET /friends/list (after accept)", r_list.status_code == 200)
|
|
|
|
# Remove friend
|
|
friend_acct = r.json()
|
|
r_remove = authed.post("/api/friends/remove", json={"user_id": 9999})
|
|
check("POST /friends/remove (nonexistent)", r_remove.status_code == 404)
|
|
|
|
# ---------- PLAYTIME ----------
|
|
print("\n=== PLAYTIME ===")
|
|
r = authed.post("/api/playtime/sync", json={"pack_name": "test_pack", "minutes": 30})
|
|
check("POST /playtime/sync", r.status_code == 200, r.text[:100])
|
|
|
|
r2 = authed.get("/api/playtime/stats")
|
|
check("GET /playtime/stats", r2.status_code == 200)
|
|
stats = r2.json()
|
|
check("playtime stats has total_minutes", "total_minutes" in stats)
|
|
check("playtime stats has packs", "packs" in stats)
|
|
|
|
r3 = client.get("/api/playtime/stats")
|
|
check("GET /playtime/stats (unauthed)", r3.status_code == 401)
|
|
|
|
# ---------- SUMMARY ----------
|
|
total = passed + failed
|
|
print(f"\n{'='*40}")
|
|
print(f"RESULTS: {passed}/{total} passed, {failed}/{total} failed")
|
|
|
|
except Exception as e:
|
|
print(f"\nTEST ERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
if proc:
|
|
proc.terminate()
|
|
proc.wait(timeout=5)
|
|
|
|
sys.exit(1 if failed > 0 else 0)
|