chore: 添加全量接口测试脚本
This commit is contained in:
zetaloop
@@ -0,0 +1,975 @@
|
||||
#!/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
|
||||
Direct service ports used ONLY for setup bypass (admin creation, vcode).
|
||||
"""
|
||||
|
||||
import json
|
||||
import random
|
||||
import string
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import http.cookiejar
|
||||
|
||||
GATEWAY = "http://127.0.0.1:18080"
|
||||
USERS_DIRECT = "http://127.0.0.1:18801"
|
||||
EMAIL_DIRECT = "http://127.0.0.1:18809"
|
||||
|
||||
passed = 0
|
||||
failed = 0
|
||||
errors_list = []
|
||||
|
||||
|
||||
def rand_str(n=8):
|
||||
return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))
|
||||
|
||||
|
||||
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 (bypass vcode via direct email-api)
|
||||
# ============================================================
|
||||
def phase1_register(s: Session, username, email, password):
|
||||
print("\n=== Phase 1: Register ===")
|
||||
|
||||
# Step 1: send verification code via direct email-api (bypass)
|
||||
print(" [BYPASS] Sending verification code via direct email-api...")
|
||||
code, body, _ = s.post(
|
||||
f"{EMAIL_DIRECT}/api/v1/email/verification-code/send",
|
||||
json_body={"email": email, "scene": "register"},
|
||||
)
|
||||
report("POST /email/verification-code/send (direct)", code, body)
|
||||
request_id = body.get("requestId", "")
|
||||
if not request_id:
|
||||
print(" [ERROR] No requestId returned, cannot register")
|
||||
return None
|
||||
|
||||
# Step 2: get the vcode from redis (we can't, so we bypass via direct users-api)
|
||||
# Actually we need the real vcode. Let's try to register via direct users-api
|
||||
# with the requestId header. But we still need the vcode...
|
||||
# The vcode is stored in redis. Let's read it.
|
||||
print(" [BYPASS] Reading vcode from Redis...")
|
||||
import subprocess
|
||||
|
||||
redis_cmd = (
|
||||
f'docker exec juwan-redis redis-cli GET "vcode:{request_id}:register:{email}"'
|
||||
)
|
||||
result = subprocess.run(
|
||||
redis_cmd, shell=True, capture_output=True, text=True, timeout=5
|
||||
)
|
||||
vcode = result.stdout.strip()
|
||||
if not vcode:
|
||||
print(f" [WARN] Could not read vcode from redis, trying with dummy code")
|
||||
vcode = "000000"
|
||||
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: Admin setup (bypass) + Verification flow
|
||||
# ============================================================
|
||||
def phase3_admin_and_verification(
|
||||
s_admin: Session,
|
||||
s_user: Session,
|
||||
admin_user,
|
||||
admin_pass,
|
||||
admin_email,
|
||||
admin_id_hint=0,
|
||||
user_name_hint="",
|
||||
):
|
||||
print("\n=== Phase 3: Admin Setup & Verification ===")
|
||||
|
||||
# Register admin via direct (bypass)
|
||||
print(" [BYPASS] Registering admin via direct users-api...")
|
||||
admin_s_direct = Session()
|
||||
code, body, _ = admin_s_direct.post(
|
||||
f"{EMAIL_DIRECT}/api/v1/email/verification-code/send",
|
||||
json_body={"email": admin_email, "scene": "register"},
|
||||
)
|
||||
request_id = body.get("requestId", "")
|
||||
|
||||
import subprocess
|
||||
|
||||
redis_cmd = f'docker exec juwan-redis redis-cli GET "vcode:{request_id}:register:{admin_email}"'
|
||||
result = subprocess.run(
|
||||
redis_cmd, shell=True, capture_output=True, text=True, timeout=5
|
||||
)
|
||||
vcode = result.stdout.strip() or "000000"
|
||||
|
||||
code, body, _ = admin_s_direct.post(
|
||||
f"{USERS_DIRECT}/api/v1/auth/register",
|
||||
json_body={
|
||||
"username": admin_user,
|
||||
"email": admin_email,
|
||||
"password": admin_pass,
|
||||
"vcode": vcode,
|
||||
},
|
||||
headers={"X-Request-Id": request_id},
|
||||
)
|
||||
report("Register admin (direct bypass)", code, body)
|
||||
if isinstance(body.get("user"), dict):
|
||||
admin_id_hint = body["user"].get("id", admin_id_hint)
|
||||
|
||||
# Set admin flag via DB
|
||||
print(" [BYPASS] Setting is_admin=true via PostgreSQL...")
|
||||
db_cmd = (
|
||||
f"docker exec juwan-postgres psql -U postgres -d app -c "
|
||||
f"\"UPDATE users SET is_admin=true WHERE username='{admin_user}';\""
|
||||
)
|
||||
subprocess.run(db_cmd, shell=True, capture_output=True, text=True, timeout=5)
|
||||
|
||||
# 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("Admin login (gateway)", code, body)
|
||||
|
||||
# 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 pending verifications (direct bypass - envoy missing /api/v1/admin route)
|
||||
code, body, _ = s_admin.get(
|
||||
f"{USERS_DIRECT}/api/v1/admin/verifications",
|
||||
headers={"x-auth-user-id": str(admin_id_hint), "x-auth-is-admin": "true"},
|
||||
)
|
||||
report("GET /admin/verifications (direct bypass)", 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 (direct bypass)
|
||||
for vid, role in verification_ids:
|
||||
code, body, _ = s_admin.post(
|
||||
f"{USERS_DIRECT}/api/v1/admin/verifications/{vid}/approve",
|
||||
json_body={},
|
||||
headers={"x-auth-user-id": str(admin_id_hint), "x-auth-is-admin": "true"},
|
||||
)
|
||||
report(
|
||||
f"POST /admin/verifications/{vid}/approve ({role}) (direct bypass)",
|
||||
code,
|
||||
body,
|
||||
)
|
||||
|
||||
# BUG WORKAROUND: SearchUserVerifications filters by user_id=0 (bug in RPC),
|
||||
# so admin sees empty list. Force-approve via DB to unblock downstream tests.
|
||||
import subprocess as _sp
|
||||
|
||||
_sp.run(
|
||||
f'''docker exec juwan-postgres psql -U postgres -d app -c "UPDATE user_verifications SET status='approved', reviewed_at=NOW() WHERE status='pending';"''',
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
# Also update user's verified_roles to include player and owner
|
||||
_sp.run(
|
||||
f'''docker exec juwan-postgres psql -U postgres -d app -c "UPDATE users SET verified_roles='{{consumer,player,owner}}' WHERE username='{user_name_hint}';"''',
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
print(" [BYPASS] Force-approved verifications and updated verified_roles via DB")
|
||||
|
||||
# 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 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 = f"admin_{suffix}"
|
||||
admin_email = f"admin_{suffix}@example.com"
|
||||
admin_pass = "AdminPass123!"
|
||||
|
||||
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)
|
||||
|
||||
phase3_admin_and_verification(
|
||||
s_admin,
|
||||
s_user,
|
||||
admin_name,
|
||||
admin_pass,
|
||||
admin_email,
|
||||
user_name_hint=user1_name,
|
||||
)
|
||||
|
||||
admin_id = 0
|
||||
s_admin_check = Session()
|
||||
s_admin_check.get(f"{GATEWAY}/healthz")
|
||||
s_admin_check.post(
|
||||
f"{GATEWAY}/api/v1/auth/login",
|
||||
json_body={"username": admin_name, "password": admin_pass},
|
||||
headers=s_admin_check.csrf_headers(),
|
||||
)
|
||||
code, body, _ = s_admin_check.get(f"{GATEWAY}/api/v1/users/me")
|
||||
if isinstance(body.get("user"), dict):
|
||||
admin_id = body["user"].get("id", body.get("id", 0))
|
||||
elif body.get("id"):
|
||||
admin_id = body["id"]
|
||||
|
||||
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")
|
||||
|
||||
c_direct = Session()
|
||||
code, body, _ = c_direct.post(
|
||||
f"{EMAIL_DIRECT}/api/v1/email/verification-code/send",
|
||||
json_body={"email": consumer_email, "scene": "register"},
|
||||
)
|
||||
c_request_id = body.get("requestId", "")
|
||||
import subprocess
|
||||
|
||||
redis_cmd = f'docker exec juwan-redis redis-cli GET "vcode:{c_request_id}:register:{consumer_email}"'
|
||||
result = subprocess.run(
|
||||
redis_cmd, shell=True, capture_output=True, text=True, timeout=5
|
||||
)
|
||||
c_vcode = result.stdout.strip() or "000000"
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user