zetaloop
910 lines
27 KiB
Python
910 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Full API integration test for juwan-backend.
|
|
|
|
Runs against the local dev docker-compose environment.
|
|
Gateway: http://127.0.0.1:18080
|
|
Reads verification codes from local Redis only for dev registration flow.
|
|
"""
|
|
|
|
import json
|
|
import random
|
|
import string
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import urllib.parse
|
|
import http.cookiejar
|
|
import os
|
|
import subprocess
|
|
|
|
GATEWAY = "http://127.0.0.1:18080"
|
|
ADMIN_USERNAME = os.getenv("ADMIN_USERNAME", "admin")
|
|
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin123")
|
|
REDIS_CONTAINER = os.getenv("REDIS_CONTAINER", "juwan-redis")
|
|
|
|
passed = 0
|
|
failed = 0
|
|
errors_list = []
|
|
|
|
|
|
def rand_str(n=8):
|
|
return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))
|
|
|
|
|
|
def read_vcode_from_redis(request_id, scene, account):
|
|
if not request_id:
|
|
return ""
|
|
|
|
result = subprocess.run(
|
|
[
|
|
"docker",
|
|
"exec",
|
|
REDIS_CONTAINER,
|
|
"redis-cli",
|
|
"GET",
|
|
f"vcode:{request_id}:{scene}:{account}",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
class Session:
|
|
"""Minimal cookie-aware HTTP session using stdlib only."""
|
|
|
|
def __init__(self):
|
|
self.cookie_jar = http.cookiejar.CookieJar()
|
|
self.opener = urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(self.cookie_jar)
|
|
)
|
|
|
|
def get_cookie(self, name):
|
|
for c in self.cookie_jar:
|
|
if c.name == name:
|
|
return c.value
|
|
return None
|
|
|
|
def request(self, method, url, json_body=None, headers=None, form_data=None):
|
|
hdrs = headers or {}
|
|
body = None
|
|
if json_body is not None:
|
|
body = json.dumps(json_body).encode()
|
|
hdrs.setdefault("Content-Type", "application/json")
|
|
elif form_data is not None:
|
|
body = urllib.parse.urlencode(form_data).encode()
|
|
hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded")
|
|
|
|
req = urllib.request.Request(url, data=body, headers=hdrs, method=method)
|
|
try:
|
|
resp = self.opener.open(req, timeout=15)
|
|
data = resp.read().decode()
|
|
try:
|
|
return resp.status, json.loads(data) if data else {}, dict(resp.headers)
|
|
except json.JSONDecodeError:
|
|
return resp.status, {"_raw": data}, dict(resp.headers)
|
|
except urllib.error.HTTPError as e:
|
|
data = e.read().decode() if e.fp else ""
|
|
try:
|
|
return e.code, json.loads(data) if data else {}, dict(e.headers)
|
|
except json.JSONDecodeError:
|
|
return e.code, {"_raw": data}, dict(e.headers)
|
|
except Exception as e:
|
|
return 0, {"_error": str(e)}, {}
|
|
|
|
def get(self, url, **kw):
|
|
return self.request("GET", url, **kw)
|
|
|
|
def post(self, url, **kw):
|
|
return self.request("POST", url, **kw)
|
|
|
|
def put(self, url, **kw):
|
|
return self.request("PUT", url, **kw)
|
|
|
|
def delete(self, url, **kw):
|
|
return self.request("DELETE", url, **kw)
|
|
|
|
def csrf_headers(self):
|
|
token = self.get_cookie("XSRF-TOKEN")
|
|
return {"xsrf-token": token} if token else {}
|
|
|
|
|
|
def report(name, status_code, body, expect_status=200):
|
|
global passed, failed
|
|
ok = status_code == expect_status
|
|
mark = "PASS" if ok else "FAIL"
|
|
if not ok:
|
|
failed += 1
|
|
errors_list.append((name, status_code, body))
|
|
else:
|
|
passed += 1
|
|
body_preview = json.dumps(body, ensure_ascii=False)
|
|
if len(body_preview) > 200:
|
|
body_preview = body_preview[:200] + "..."
|
|
print(f" [{mark}] {name}: HTTP {status_code} {body_preview}")
|
|
return ok
|
|
|
|
|
|
# ============================================================
|
|
# Phase 0: Health check & CSRF
|
|
# ============================================================
|
|
def phase0_health(s: Session):
|
|
print("\n=== Phase 0: Health & CSRF ===")
|
|
code, body, hdrs = s.get(f"{GATEWAY}/healthz")
|
|
report("GET /healthz", code, body)
|
|
xsrf = s.get_cookie("XSRF-TOKEN")
|
|
xsrf_guard = s.get_cookie("XSRF-GUARD")
|
|
print(f" XSRF-TOKEN: {xsrf}")
|
|
print(f" XSRF-GUARD: {xsrf_guard}")
|
|
if not xsrf:
|
|
print(" [WARN] No XSRF-TOKEN cookie received, POST requests will fail")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 1: Registration
|
|
# ============================================================
|
|
def phase1_register(s: Session, username, email, password):
|
|
print("\n=== Phase 1: Register ===")
|
|
|
|
# Step 1: send verification code via gateway
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/email/verification-code/send",
|
|
json_body={"email": email, "scene": "register"},
|
|
headers=s.csrf_headers(),
|
|
)
|
|
report("POST /email/verification-code/send (gateway)", code, body)
|
|
request_id = body.get("requestId", "")
|
|
if not request_id:
|
|
print(" [ERROR] No requestId returned, cannot register")
|
|
return None
|
|
|
|
# Step 2: read the dev verification code from Redis
|
|
print(" [BYPASS] Reading vcode from Redis...")
|
|
vcode = read_vcode_from_redis(request_id, "register", email)
|
|
if not vcode:
|
|
print(" [WARN] No verification code found in Redis")
|
|
else:
|
|
print(f" Vcode from Redis: {vcode}")
|
|
|
|
# Step 3: register via gateway with X-Request-Id
|
|
csrf = s.csrf_headers()
|
|
csrf["X-Request-Id"] = request_id
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/auth/register",
|
|
json_body={
|
|
"username": username,
|
|
"email": email,
|
|
"password": password,
|
|
"vcode": vcode,
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report("POST /auth/register (gateway)", code, body)
|
|
return body
|
|
|
|
|
|
def phase1_login(s: Session, username, password):
|
|
print("\n=== Phase 1b: Login ===")
|
|
csrf = s.csrf_headers()
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/auth/login",
|
|
json_body={"username": username, "password": password},
|
|
headers=csrf,
|
|
)
|
|
report("POST /auth/login", code, body)
|
|
jtoken = s.get_cookie("JToken")
|
|
print(f" JToken: {jtoken[:30]}..." if jtoken else " JToken: None")
|
|
return body
|
|
|
|
|
|
# ============================================================
|
|
# Phase 2: User endpoints (authenticated)
|
|
# ============================================================
|
|
def phase2_user(s: Session, user_id):
|
|
print("\n=== Phase 2: User Endpoints ===")
|
|
csrf = s.csrf_headers()
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me", code, body)
|
|
|
|
code, body, _ = s.put(
|
|
f"{GATEWAY}/api/v1/users/me",
|
|
json_body={"nickname": "TestNick", "bio": "testbio"},
|
|
headers=csrf,
|
|
)
|
|
report("PUT /users/me", code, body)
|
|
|
|
code, body, _ = s.put(
|
|
f"{GATEWAY}/api/v1/users/me/preferences/notifications",
|
|
json_body={"order": True, "community": True, "system": True},
|
|
headers=csrf,
|
|
)
|
|
report("PUT /users/me/preferences/notifications", code, body)
|
|
|
|
code, body, _ = s.put(
|
|
f"{GATEWAY}/api/v1/users/me/preferences/theme",
|
|
json_body={"theme": "dark"},
|
|
headers=csrf,
|
|
)
|
|
report("PUT /users/me/preferences/theme", code, body)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/users/{user_id}")
|
|
report(f"GET /users/{user_id} (public)", code, body)
|
|
|
|
|
|
# ============================================================
|
|
# Phase 3: Built-in Admin + Verification flow
|
|
# ============================================================
|
|
def phase3_admin_and_verification(
|
|
s_admin: Session,
|
|
s_user: Session,
|
|
admin_user,
|
|
admin_pass,
|
|
):
|
|
print("\n=== Phase 3: Built-in Admin & Verification ===")
|
|
|
|
# Login admin via gateway
|
|
s_admin.get(f"{GATEWAY}/healthz") # get CSRF
|
|
csrf = s_admin.csrf_headers()
|
|
code, body, _ = s_admin.post(
|
|
f"{GATEWAY}/api/v1/auth/login",
|
|
json_body={"username": admin_user, "password": admin_pass},
|
|
headers=csrf,
|
|
)
|
|
report("POST /auth/login (admin)", code, body)
|
|
|
|
admin_id = 0
|
|
code, body, _ = s_admin.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (admin)", code, body)
|
|
if isinstance(body.get("user"), dict):
|
|
admin_id = body["user"].get("id", admin_id)
|
|
elif body.get("id"):
|
|
admin_id = body.get("id", admin_id)
|
|
|
|
# User applies for player verification
|
|
csrf_user = s_user.csrf_headers()
|
|
code, body, _ = s_user.post(
|
|
f"{GATEWAY}/api/v1/users/me/verification",
|
|
json_body={
|
|
"role": "player",
|
|
"materials": {
|
|
"idCardFront": "http://example.com/front.jpg",
|
|
"idCardBack": "http://example.com/back.jpg",
|
|
"gameScreenshots": ["http://example.com/ss1.jpg"],
|
|
"voiceDemo": "http://example.com/voice.mp3",
|
|
},
|
|
},
|
|
headers=csrf_user,
|
|
)
|
|
report("POST /users/me/verification (apply player)", code, body)
|
|
|
|
# User applies for owner verification
|
|
code, body, _ = s_user.post(
|
|
f"{GATEWAY}/api/v1/users/me/verification",
|
|
json_body={
|
|
"role": "owner",
|
|
"materials": {
|
|
"idCardFront": "http://example.com/front.jpg",
|
|
"idCardBack": "http://example.com/back.jpg",
|
|
"gameScreenshots": [],
|
|
"voiceDemo": "",
|
|
},
|
|
},
|
|
headers=csrf_user,
|
|
)
|
|
report("POST /users/me/verification (apply owner)", code, body)
|
|
|
|
# Get my verifications
|
|
code, body, _ = s_user.get(f"{GATEWAY}/api/v1/users/me/verification")
|
|
report("GET /users/me/verification", code, body)
|
|
|
|
# Admin: list and approve via gateway
|
|
code, body, _ = s_admin.get(f"{GATEWAY}/api/v1/admin/verifications")
|
|
report("GET /admin/verifications", code, body)
|
|
|
|
verification_ids = []
|
|
if isinstance(body.get("list"), list):
|
|
for v in body["list"]:
|
|
verification_ids.append((v.get("id"), v.get("role")))
|
|
|
|
# Admin: approve all
|
|
for vid, role in verification_ids:
|
|
code, body, _ = s_admin.post(
|
|
f"{GATEWAY}/api/v1/admin/verifications/{vid}/approve",
|
|
json_body={},
|
|
headers=s_admin.csrf_headers(),
|
|
)
|
|
report(
|
|
f"POST /admin/verifications/{vid}/approve ({role})",
|
|
code,
|
|
body,
|
|
)
|
|
|
|
# User: switch role to player
|
|
code, body, _ = s_user.post(
|
|
f"{GATEWAY}/api/v1/users/me/switch-role",
|
|
json_body={"role": "player"},
|
|
headers=csrf_user,
|
|
)
|
|
report("POST /users/me/switch-role (player)", code, body)
|
|
|
|
return admin_id, verification_ids
|
|
|
|
|
|
def phase4_follow(s: Session, target_user_id):
|
|
print("\n=== Phase 4: Follow/Unfollow ===")
|
|
csrf = s.csrf_headers()
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/users/{target_user_id}/follow",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /users/{target_user_id}/follow", code, body)
|
|
|
|
code, body, _ = s.delete(
|
|
f"{GATEWAY}/api/v1/users/{target_user_id}/follow",
|
|
headers=csrf,
|
|
)
|
|
report(f"DELETE /users/{target_user_id}/follow", code, body)
|
|
|
|
|
|
def phase5_games(s: Session, s_admin: Session):
|
|
print("\n=== Phase 5: Games ===")
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/games")
|
|
report("GET /games", code, body)
|
|
|
|
csrf_admin = s_admin.csrf_headers()
|
|
code, body, _ = s_admin.post(
|
|
f"{GATEWAY}/api/v1/games",
|
|
json_body={
|
|
"name": f"TestGame_{rand_str(4)}",
|
|
"icon": "icon.png",
|
|
"category": "MOBA",
|
|
},
|
|
headers=csrf_admin,
|
|
)
|
|
report("POST /games (create)", code, body)
|
|
game_id = body.get("id", 0)
|
|
|
|
# createGameLogic returns empty Game{} (bug), so get game_id from list
|
|
code, body2, _ = s.get(f"{GATEWAY}/api/v1/games")
|
|
report("GET /games (after create)", code, body2)
|
|
if not game_id and isinstance(body2.get("list"), list) and body2["list"]:
|
|
game_id = body2["list"][-1].get("id", 0)
|
|
|
|
if game_id:
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/games/{game_id}")
|
|
report(f"GET /games/{game_id}", code, body)
|
|
|
|
return game_id
|
|
|
|
|
|
def phase6_player(s: Session, game_id):
|
|
print("\n=== Phase 6: Player ===")
|
|
csrf = s.csrf_headers()
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/players/me",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report("POST /players/me (init)", code, body)
|
|
player_id = body.get("id", 0)
|
|
|
|
code, body, _ = s.put(
|
|
f"{GATEWAY}/api/v1/players/me/status",
|
|
json_body={"status": "online"},
|
|
headers=csrf,
|
|
)
|
|
report("PUT /players/me/status", code, body)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/players")
|
|
report("GET /players", code, body)
|
|
|
|
if player_id:
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/players/{player_id}")
|
|
report(f"GET /players/{player_id}", code, body)
|
|
|
|
svc_body = None
|
|
if game_id:
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/services",
|
|
json_body={
|
|
"gameId": game_id,
|
|
"title": "Boosting Service",
|
|
"description": "Rank boost",
|
|
"price": 50.0,
|
|
"unit": "game",
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report("POST /services (create)", code, body)
|
|
svc_body = body
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/services")
|
|
report("GET /services", code, body)
|
|
|
|
service_id = svc_body.get("id", 0) if svc_body else 0
|
|
if service_id:
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/services/{service_id}")
|
|
report(f"GET /services/{service_id}", code, body)
|
|
|
|
code, body, _ = s.put(
|
|
f"{GATEWAY}/api/v1/services/{service_id}",
|
|
json_body={
|
|
"title": "Updated Service",
|
|
"price": 60.0,
|
|
"availability": ["weekday"],
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report(f"PUT /services/{service_id}", code, body)
|
|
|
|
if player_id:
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/players/{player_id}/services")
|
|
report(f"GET /players/{player_id}/services", code, body)
|
|
|
|
return player_id, service_id
|
|
|
|
|
|
def phase7_shop(s_owner: Session, player_id):
|
|
print("\n=== Phase 7: Shop ===")
|
|
|
|
s_owner.post(
|
|
f"{GATEWAY}/api/v1/users/me/switch-role",
|
|
json_body={"role": "owner"},
|
|
headers=s_owner.csrf_headers(),
|
|
)
|
|
|
|
csrf = s_owner.csrf_headers()
|
|
code, body, _ = s_owner.post(
|
|
f"{GATEWAY}/api/v1/shops",
|
|
json_body={
|
|
"name": f"TestShop_{rand_str(4)}",
|
|
"description": "A test shop",
|
|
"commissionType": "percentage",
|
|
"commissionValue": "10",
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report("POST /shops (create)", code, body)
|
|
shop_id_str = body.get("id", "0")
|
|
try:
|
|
shop_id = int(shop_id_str)
|
|
except (ValueError, TypeError):
|
|
shop_id = 0
|
|
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops")
|
|
report("GET /shops", code, body)
|
|
|
|
if shop_id:
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
|
|
report(f"GET /shops/{shop_id}", code, body)
|
|
|
|
code, body, _ = s_owner.put(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/template",
|
|
json_body={"sections": json.dumps({"layout": "grid", "theme": "dark"})},
|
|
headers=csrf,
|
|
)
|
|
report(f"PUT /shops/{shop_id}", code, body)
|
|
|
|
code, body, _ = s_owner.post(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/announcements",
|
|
json_body={"content": "Grand opening!"},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /shops/{shop_id}/announcements", code, body)
|
|
|
|
code, body, _ = s_owner.delete(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/announcements/0",
|
|
headers=csrf,
|
|
)
|
|
report(f"DELETE /shops/{shop_id}/announcements/0", code, body)
|
|
|
|
code, body, _ = s_owner.put(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/template",
|
|
json_body={"sections": json.dumps({"layout": "grid", "theme": "dark"})},
|
|
headers=csrf,
|
|
)
|
|
report(f"PUT /shops/{shop_id}/template", code, body)
|
|
|
|
code, body, _ = s_owner.get(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/income-stats",
|
|
)
|
|
report(f"GET /shops/{shop_id}/income-stats", code, body)
|
|
|
|
if player_id:
|
|
code, body, _ = s_owner.post(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
|
|
json_body={"playerId": player_id},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /shops/{shop_id}/invitations", code, body)
|
|
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/mine")
|
|
report("GET /shops/mine", code, body)
|
|
|
|
return shop_id
|
|
|
|
|
|
def phase8_order(s_consumer: Session, player_id, service_id, shop_id):
|
|
print("\n=== Phase 8: Orders ===")
|
|
|
|
s_consumer.post(
|
|
f"{GATEWAY}/api/v1/users/me/switch-role",
|
|
json_body={"role": "consumer"},
|
|
headers=s_consumer.csrf_headers(),
|
|
)
|
|
|
|
csrf = s_consumer.csrf_headers()
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders",
|
|
json_body={
|
|
"playerId": player_id,
|
|
"serviceId": service_id,
|
|
"shopId": shop_id,
|
|
"quantity": 1,
|
|
"note": "test order",
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report("POST /orders (create)", code, body)
|
|
order_obj = body.get("order", {})
|
|
order_id = order_obj.get("id", 0) if isinstance(order_obj, dict) else 0
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders?role=consumer")
|
|
report("GET /orders?role=consumer", code, body)
|
|
|
|
if order_id:
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders/{order_id}")
|
|
report(f"GET /orders/{order_id}", code, body)
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order_id}/pay",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order_id}/pay", code, body)
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order_id}/cancel",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order_id}/cancel", code, body)
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders",
|
|
json_body={
|
|
"playerId": player_id,
|
|
"serviceId": service_id,
|
|
"quantity": 1,
|
|
},
|
|
headers=csrf,
|
|
)
|
|
order2 = body.get("order", {})
|
|
order2_id = order2.get("id", 0) if isinstance(order2, dict) else 0
|
|
|
|
if order2_id:
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order2_id}/reorder",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order2_id}/reorder", code, body)
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/paid",
|
|
json_body={
|
|
"playerId": player_id,
|
|
"serviceId": service_id,
|
|
"quantity": 1,
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report("POST /orders/paid (create+pay)", code, body)
|
|
|
|
return order_id
|
|
|
|
|
|
def phase9_wallet(s: Session):
|
|
print("\n=== Phase 9: Wallet ===")
|
|
csrf = s.csrf_headers()
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/wallet/balance")
|
|
report("GET /wallet/balance", code, body)
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/wallet/topup",
|
|
json_body={"amount": "100.00", "method": "alipay"},
|
|
headers=csrf,
|
|
)
|
|
report("POST /wallet/topup", code, body)
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/wallet/withdraw",
|
|
json_body={"amount": "10.00", "method": "alipay"},
|
|
headers=csrf,
|
|
)
|
|
report("POST /wallet/withdraw", code, body)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/wallet/transactions")
|
|
report("GET /wallet/transactions", code, body)
|
|
|
|
|
|
def phase10_community(s: Session, user_id):
|
|
print("\n=== Phase 10: Community ===")
|
|
csrf = s.csrf_headers()
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/posts",
|
|
json_body={
|
|
"title": "Test Post",
|
|
"content": "Hello world",
|
|
"images": [],
|
|
"tags": ["test"],
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report("POST /posts (create)", code, body)
|
|
post_id = body.get("id", 0)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/posts")
|
|
report("GET /posts", code, body)
|
|
|
|
if post_id:
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/posts/{post_id}")
|
|
report(f"GET /posts/{post_id}", code, body)
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/posts/{post_id}/like",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /posts/{post_id}/like", code, body)
|
|
|
|
code, body, _ = s.delete(
|
|
f"{GATEWAY}/api/v1/posts/{post_id}/like",
|
|
headers=csrf,
|
|
)
|
|
report(f"DELETE /posts/{post_id}/like", code, body)
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/posts/{post_id}/pin",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /posts/{post_id}/pin", code, body)
|
|
|
|
code, body, _ = s.delete(
|
|
f"{GATEWAY}/api/v1/posts/{post_id}/pin",
|
|
headers=csrf,
|
|
)
|
|
report(f"DELETE /posts/{post_id}/pin", code, body)
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/posts/{post_id}/comments",
|
|
json_body={"content": "Nice post!"},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /posts/{post_id}/comments", code, body)
|
|
comment_id = body.get("id", 0)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/posts/{post_id}/comments")
|
|
report(f"GET /posts/{post_id}/comments", code, body)
|
|
|
|
if comment_id:
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/comments/{comment_id}/like",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /comments/{comment_id}/like", code, body)
|
|
|
|
code, body, _ = s.delete(
|
|
f"{GATEWAY}/api/v1/comments/{comment_id}/like",
|
|
headers=csrf,
|
|
)
|
|
report(f"DELETE /comments/{comment_id}/like", code, body)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/users/{user_id}/posts")
|
|
report(f"GET /users/{user_id}/posts", code, body)
|
|
|
|
return post_id
|
|
|
|
|
|
def phase11_objectstory(s: Session):
|
|
print("\n=== Phase 11: Objectstory (File) ===")
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/files?key=nonexistent")
|
|
report("GET /files?key=nonexistent (expect error)", code, body, expect_status=400)
|
|
|
|
return
|
|
|
|
|
|
def phase12_email(s: Session):
|
|
print("\n=== Phase 12: Email ===")
|
|
csrf = s.csrf_headers()
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/email/verification-code/send",
|
|
json_body={"email": f"test_{rand_str(4)}@example.com", "scene": "register"},
|
|
headers=csrf,
|
|
)
|
|
report("POST /email/verification-code/send (gateway)", code, body)
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/auth/forgot-password/send",
|
|
json_body={"email": f"test_{rand_str(4)}@example.com"},
|
|
headers=csrf,
|
|
)
|
|
report("POST /auth/forgot-password/send (gateway)", code, body)
|
|
|
|
|
|
def phase13_logout(s: Session):
|
|
print("\n=== Phase 13: Logout ===")
|
|
csrf = s.csrf_headers()
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/auth/logout",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report("POST /auth/logout", code, body)
|
|
|
|
|
|
def phase14_misc_auth(s: Session):
|
|
print("\n=== Phase 14: Forgot/Reset Password ===")
|
|
s.get(f"{GATEWAY}/healthz")
|
|
csrf = s.csrf_headers()
|
|
|
|
test_email = f"reset_{rand_str(4)}@example.com"
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/auth/forgot-password/send",
|
|
json_body={"email": test_email},
|
|
headers=csrf,
|
|
)
|
|
report("POST /auth/forgot-password/send", code, body)
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/auth/reset-password",
|
|
json_body={
|
|
"email": test_email,
|
|
"vcode": "000000",
|
|
"newPassword": "newpass123",
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report(
|
|
"POST /auth/reset-password (expect fail, wrong vcode)",
|
|
code,
|
|
body,
|
|
expect_status=400,
|
|
)
|
|
|
|
|
|
def phase15_player_service_delete(s: Session, service_id):
|
|
print("\n=== Phase 15: Delete Service ===")
|
|
if not service_id:
|
|
print(" [SKIP] No service to delete")
|
|
return
|
|
csrf = s.csrf_headers()
|
|
code, body, _ = s.delete(
|
|
f"{GATEWAY}/api/v1/services/{service_id}",
|
|
headers=csrf,
|
|
)
|
|
report(f"DELETE /services/{service_id}", code, body)
|
|
|
|
|
|
def main():
|
|
global passed, failed
|
|
|
|
suffix = rand_str(6)
|
|
user1_name = f"testuser_{suffix}"
|
|
user1_email = f"testuser_{suffix}@example.com"
|
|
user1_pass = "TestPass123!"
|
|
admin_name = ADMIN_USERNAME
|
|
admin_pass = ADMIN_PASSWORD
|
|
|
|
print(f"Test run: user={user1_name}, admin={admin_name}")
|
|
|
|
s_user = Session()
|
|
s_admin = Session()
|
|
|
|
phase0_health(s_user)
|
|
phase1_register(s_user, user1_name, user1_email, user1_pass)
|
|
login_resp = phase1_login(s_user, user1_name, user1_pass)
|
|
|
|
user_id = 0
|
|
if isinstance(login_resp.get("user"), dict):
|
|
user_id = login_resp["user"].get("id", 0)
|
|
print(f" User ID: {user_id}")
|
|
|
|
phase2_user(s_user, user_id)
|
|
|
|
admin_id, _ = phase3_admin_and_verification(
|
|
s_admin,
|
|
s_user,
|
|
admin_name,
|
|
admin_pass,
|
|
)
|
|
|
|
if admin_id and user_id:
|
|
phase4_follow(s_user, admin_id)
|
|
|
|
s_user.post(
|
|
f"{GATEWAY}/api/v1/users/me/switch-role",
|
|
json_body={"role": "player"},
|
|
headers=s_user.csrf_headers(),
|
|
)
|
|
|
|
game_id = phase5_games(s_user, s_admin)
|
|
player_id, service_id = phase6_player(s_user, game_id)
|
|
shop_id = phase7_shop(s_user, player_id)
|
|
|
|
s_consumer = Session()
|
|
consumer_name = f"consumer_{suffix}"
|
|
consumer_email = f"consumer_{suffix}@example.com"
|
|
consumer_pass = "ConsumerPass123!"
|
|
s_consumer.get(f"{GATEWAY}/healthz")
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/email/verification-code/send",
|
|
json_body={"email": consumer_email, "scene": "register"},
|
|
headers=s_consumer.csrf_headers(),
|
|
)
|
|
report("POST /email/verification-code/send (consumer)", code, body)
|
|
c_request_id = body.get("requestId", "")
|
|
|
|
c_vcode = read_vcode_from_redis(c_request_id, "register", consumer_email)
|
|
if not c_vcode:
|
|
print(" [WARN] No consumer verification code found in Redis")
|
|
|
|
csrf_c = s_consumer.csrf_headers()
|
|
csrf_c["X-Request-Id"] = c_request_id
|
|
s_consumer.post(
|
|
f"{GATEWAY}/api/v1/auth/register",
|
|
json_body={
|
|
"username": consumer_name,
|
|
"email": consumer_email,
|
|
"password": consumer_pass,
|
|
"vcode": c_vcode,
|
|
},
|
|
headers=csrf_c,
|
|
)
|
|
s_consumer.post(
|
|
f"{GATEWAY}/api/v1/auth/login",
|
|
json_body={"username": consumer_name, "password": consumer_pass},
|
|
headers=s_consumer.csrf_headers(),
|
|
)
|
|
|
|
phase8_order(s_consumer, player_id, service_id, shop_id)
|
|
phase9_wallet(s_user)
|
|
phase10_community(s_user, user_id)
|
|
phase11_objectstory(s_user)
|
|
phase12_email(s_user)
|
|
phase14_misc_auth(Session())
|
|
phase15_player_service_delete(s_user, service_id)
|
|
phase13_logout(s_user)
|
|
|
|
print(f"\n{'=' * 60}")
|
|
print(f"RESULTS: {passed} passed, {failed} failed, {passed + failed} total")
|
|
print(f"{'=' * 60}")
|
|
if errors_list:
|
|
print("\nFailed tests:")
|
|
for name, status, body in errors_list:
|
|
body_s = json.dumps(body, ensure_ascii=False)
|
|
if len(body_s) > 300:
|
|
body_s = body_s[:300] + "..."
|
|
print(f" - {name}: HTTP {status} {body_s}")
|
|
|
|
sys.exit(1 if failed else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|