fix: автодетект --cli в Bootstrap, отображение локальных сборок в JFX, обработка ошибок установки и друзей
This commit is contained in:
SashegDev
+191
@@ -0,0 +1,191 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user