juwan/juwan-backend
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fix: 更新接口测试脚本

Browse Source
This commit is contained in:
zetaloop
2026-04-23 03:24:24 +08:00
parent cd2c6d4192
commit 9bc86dd904
1 changed files with 198 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/dev/test_all_apis.py
+198 -96
View File
@@ -10,7 +10,6 @@ import json
import random
import string
import sys
import time
import urllib.request
import urllib.error
import urllib.parse
@@ -32,6 +31,29 @@ def rand_str(n=8):
return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))
def as_int(value, default=0):
try:
return int(value)
except (TypeError, ValueError):
return default
def pick_items(body):
if isinstance(body.get("items"), list):
return body["items"]
if isinstance(body.get("list"), list):
return body["list"]
return []
def pick_user_id(body):
if not isinstance(body, dict):
return 0
if isinstance(body.get("user"), dict):
return as_int(body["user"].get("id"))
return as_int(body.get("id"))
def read_vcode_from_redis(request_id, scene, account):
if not request_id:
return ""
@@ -117,7 +139,10 @@ class Session:
def report(name, status_code, body, expect_status=200):
global passed, failed
ok = status_code == expect_status
if isinstance(expect_status, (list, tuple, set)):
ok = status_code in expect_status
else:
ok = status_code == expect_status
mark = "PASS" if ok else "FAIL"
if not ok:
failed += 1
@@ -136,7 +161,7 @@ def report(name, status_code, body, expect_status=200):
# ============================================================
def phase0_health(s: Session):
print("\n=== Phase 0: Health & CSRF ===")
code, body, hdrs = s.get(f"{GATEWAY}/healthz")
code, body, _ = s.get(f"{GATEWAY}/healthz")
report("GET /healthz", code, body)
xsrf = s.get_cookie("__Host-XSRF-TOKEN")
xsrf_guard = s.get_cookie("__Host-XSRF-GUARD")
@@ -149,8 +174,8 @@ def phase0_health(s: Session):
# ============================================================
# Phase 1: Registration
# ============================================================
def phase1_register(s: Session, username, email, password):
print("\n=== Phase 1: Register ===")
def phase1_register(s: Session, username, email, password, label="user"):
print(f"\n=== Phase 1: Register ({label}) ===")
# Step 1: send verification code via gateway
code, body, _ = s.post(
@@ -158,7 +183,7 @@ def phase1_register(s: Session, username, email, password):
json_body={"email": email, "scene": "register"},
headers=s.csrf_headers(),
)
report("POST /email/verification-code/send (gateway)", code, body)
report(f"POST /email/verification-code/send ({label})", code, body)
request_id = body.get("requestId", "")
if not request_id:
print(" [ERROR] No requestId returned, cannot register")
@@ -185,19 +210,19 @@ def phase1_register(s: Session, username, email, password):
},
headers=csrf,
)
report("POST /auth/register (gateway)", code, body)
report(f"POST /auth/register ({label})", code, body)
return body
def phase1_login(s: Session, username, password):
print("\n=== Phase 1b: Login ===")
def phase1_login(s: Session, username, password, label="user"):
print(f"\n=== Phase 1b: Login ({label}) ===")
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)
report(f"POST /auth/login ({label})", code, body)
jtoken = s.get_cookie("JToken")
print(f" JToken: {jtoken[:30]}..." if jtoken else " JToken: None")
return body
@@ -244,6 +269,7 @@ def phase2_user(s: Session, user_id):
def phase3_admin_and_verification(
s_admin: Session,
s_user: Session,
s_reject: Session,
admin_user,
admin_pass,
):
@@ -259,13 +285,9 @@ def phase3_admin_and_verification(
)
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)
admin_id = pick_user_id(body)
# User applies for player verification
csrf_user = s_user.csrf_headers()
@@ -300,21 +322,45 @@ def phase3_admin_and_verification(
)
report("POST /users/me/verification (apply owner)", code, body)
# Another user applies so admin reject flow is covered too
csrf_reject = s_reject.csrf_headers()
code, body, _ = s_reject.post(
f"{GATEWAY}/api/v1/users/me/verification",
json_body={
"role": "owner",
"materials": {
"idCardFront": "http://example.com/reject-front.jpg",
"idCardBack": "http://example.com/reject-back.jpg",
"gameScreenshots": [],
"voiceDemo": "",
},
},
headers=csrf_reject,
)
report("POST /users/me/verification (apply owner, reject user)", code, body)
# Get my verifications
code, body, _ = s_user.get(f"{GATEWAY}/api/v1/users/me/verification")
report("GET /users/me/verification", code, body)
user_verification_ids = {}
for v in pick_items(body):
user_verification_ids[v.get("role")] = as_int(v.get("id"))
code, body, _ = s_reject.get(f"{GATEWAY}/api/v1/users/me/verification")
report("GET /users/me/verification (reject user)", code, body)
reject_verification_ids = {}
for v in pick_items(body):
reject_verification_ids[v.get("role")] = as_int(v.get("id"))
# 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:
# Admin: approve only the current test user's records
for role in ("player", "owner"):
vid = user_verification_ids.get(role)
if not vid:
continue
code, body, _ = s_admin.post(
f"{GATEWAY}/api/v1/admin/verifications/{vid}/approve",
json_body={},
@@ -326,6 +372,22 @@ def phase3_admin_and_verification(
body,
)
reject_vid = reject_verification_ids.get("owner")
if reject_vid:
code, body, _ = s_admin.post(
f"{GATEWAY}/api/v1/admin/verifications/{reject_vid}/reject",
json_body={"reason": "test reject flow"},
headers=s_admin.csrf_headers(),
)
report(
f"POST /admin/verifications/{reject_vid}/reject (owner)",
code,
body,
)
code, body, _ = s_reject.get(f"{GATEWAY}/api/v1/users/me/verification")
report("GET /users/me/verification (after reject)", code, body)
# User: switch role to player
code, body, _ = s_user.post(
f"{GATEWAY}/api/v1/users/me/switch-role",
@@ -334,7 +396,7 @@ def phase3_admin_and_verification(
)
report("POST /users/me/switch-role (player)", code, body)
return admin_id, verification_ids
return admin_id
def phase4_follow(s: Session, target_user_id):
@@ -374,11 +436,11 @@ def phase5_games(s: Session, s_admin: Session):
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)
items = pick_items(body2)
if not game_id and items:
game_id = as_int(items[-1].get("id"))
if game_id:
code, body, _ = s.get(f"{GATEWAY}/api/v1/games/{game_id}")
@@ -455,7 +517,7 @@ def phase6_player(s: Session, game_id):
return player_id, service_id
def phase7_shop(s_owner: Session, player_id):
def phase7_shop(s_owner: Session, owner_user_id, player_id):
print("\n=== Phase 7: Shop ===")
s_owner.post(
@@ -490,12 +552,23 @@ def phase7_shop(s_owner: Session, player_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"})},
f"{GATEWAY}/api/v1/shops/{shop_id}",
json_body={
"name": f"UpdatedShop_{rand_str(4)}",
"description": "An updated test shop",
"commissionType": "percentage",
"commissionValue": "12",
"allowMultiShop": True,
"allowIndependentOrders": False,
"dispatchMode": "manual",
},
headers=csrf,
)
report(f"PUT /shops/{shop_id}", code, body)
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/users/{owner_user_id}/shop")
report(f"GET /users/{owner_user_id}/shop", code, body)
code, body, _ = s_owner.post(
f"{GATEWAY}/api/v1/shops/{shop_id}/announcements",
json_body={"content": "Grand opening!"},
@@ -535,7 +608,7 @@ def phase7_shop(s_owner: Session, player_id):
return shop_id
def phase8_order(s_consumer: Session, player_id, service_id, shop_id):
def phase8_order(s_consumer: Session, s_actor: Session, player_id, service_id, shop_id):
print("\n=== Phase 8: Orders ===")
s_consumer.post(
@@ -563,6 +636,12 @@ def phase8_order(s_consumer: Session, player_id, service_id, shop_id):
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders?role=consumer")
report("GET /orders?role=consumer", code, body)
code, body, _ = s_actor.get(f"{GATEWAY}/api/v1/orders?role=player")
report("GET /orders?role=player", code, body)
code, body, _ = s_actor.get(f"{GATEWAY}/api/v1/orders?role=owner")
report("GET /orders?role=owner", 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)
@@ -574,12 +653,26 @@ def phase8_order(s_consumer: Session, player_id, service_id, shop_id):
)
report(f"POST /orders/{order_id}/pay", code, body)
code, body, _ = s_actor.post(
f"{GATEWAY}/api/v1/orders/{order_id}/accept",
json_body={},
headers=s_actor.csrf_headers(),
)
report(f"POST /orders/{order_id}/accept", code, body)
code, body, _ = s_actor.post(
f"{GATEWAY}/api/v1/orders/{order_id}/request-close",
json_body={},
headers=s_actor.csrf_headers(),
)
report(f"POST /orders/{order_id}/request-close", code, body)
code, body, _ = s_consumer.post(
f"{GATEWAY}/api/v1/orders/{order_id}/cancel",
f"{GATEWAY}/api/v1/orders/{order_id}/confirm-close",
json_body={},
headers=csrf,
)
report(f"POST /orders/{order_id}/cancel", code, body)
report(f"POST /orders/{order_id}/confirm-close", code, body)
code, body, _ = s_consumer.post(
f"{GATEWAY}/api/v1/orders",
@@ -593,13 +686,21 @@ def phase8_order(s_consumer: Session, player_id, service_id, shop_id):
order2 = body.get("order", {})
order2_id = order2.get("id", 0) if isinstance(order2, dict) else 0
if order2_id:
if order_id:
code, body, _ = s_consumer.post(
f"{GATEWAY}/api/v1/orders/{order2_id}/reorder",
f"{GATEWAY}/api/v1/orders/{order_id}/reorder",
json_body={},
headers=csrf,
)
report(f"POST /orders/{order2_id}/reorder", code, body)
report(f"POST /orders/{order_id}/reorder", code, body)
if order2_id:
code, body, _ = s_consumer.post(
f"{GATEWAY}/api/v1/orders/{order2_id}/cancel",
json_body={},
headers=csrf,
)
report(f"POST /orders/{order2_id}/cancel", code, body)
code, body, _ = s_consumer.post(
f"{GATEWAY}/api/v1/orders/paid",
@@ -725,7 +826,12 @@ 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)
report(
"GET /files?key=nonexistent (expect error)",
code,
body,
expect_status=(400, 500),
)
return
@@ -760,35 +866,52 @@ def phase13_logout(s: Session):
report("POST /auth/logout", code, body)
def phase14_misc_auth(s: Session):
def phase14_reset_password(username, email, new_password):
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"
s_reset = Session()
s_reset.get(f"{GATEWAY}/healthz")
csrf = s_reset.csrf_headers()
code, body, _ = s.post(
code, body, _ = s_reset.post(
f"{GATEWAY}/api/v1/auth/forgot-password/send",
json_body={"email": test_email},
json_body={"email": email},
headers=csrf,
)
report("POST /auth/forgot-password/send", code, body)
request_id = body.get("requestId", "")
if not request_id:
print(" [WARN] No requestId returned, skip reset-password")
return
code, body, _ = s.post(
vcode = read_vcode_from_redis(request_id, "reset_password", email)
if not vcode:
print(" [WARN] No reset password verification code found in Redis")
return
csrf["X-Request-Id"] = request_id
code, body, _ = s_reset.post(
f"{GATEWAY}/api/v1/auth/reset-password",
json_body={
"email": test_email,
"vcode": "000000",
"newPassword": "newpass123",
"email": email,
"vcode": vcode,
"newPassword": new_password,
},
headers=csrf,
)
report(
"POST /auth/reset-password (expect fail, wrong vcode)",
code,
body,
expect_status=400,
report("POST /auth/reset-password", code, body)
if code != 200:
print(" [SKIP] Reset password failed, skip login verification")
return
s_login = Session()
s_login.get(f"{GATEWAY}/healthz")
code, body, _ = s_login.post(
f"{GATEWAY}/api/v1/auth/login",
json_body={"username": username, "password": new_password},
headers=s_login.csrf_headers(),
)
report("POST /auth/login (after reset)", code, body)
def phase15_player_service_delete(s: Session, service_id):
@@ -811,6 +934,10 @@ def main():
user1_name = f"testuser_{suffix}"
user1_email = f"testuser_{suffix}@example.com"
user1_pass = "TestPass123!"
consumer_name = f"consumer_{suffix}"
consumer_email = f"consumer_{suffix}@example.com"
consumer_pass = "ConsumerPass123!"
consumer_new_pass = "ConsumerPass456!"
admin_name = ADMIN_USERNAME
admin_pass = ADMIN_PASSWORD
@@ -818,21 +945,32 @@ def main():
s_user = Session()
s_admin = Session()
s_consumer = 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)
phase1_register(s_user, user1_name, user1_email, user1_pass, label="primary")
login_resp = phase1_login(s_user, user1_name, user1_pass, label="primary")
user_id = 0
if isinstance(login_resp.get("user"), dict):
user_id = login_resp["user"].get("id", 0)
user_id = pick_user_id(login_resp)
print(f" User ID: {user_id}")
phase2_user(s_user, user_id)
admin_id, _ = phase3_admin_and_verification(
s_consumer.get(f"{GATEWAY}/healthz")
phase1_register(s_consumer, consumer_name, consumer_email, consumer_pass, label="consumer")
consumer_login_resp = phase1_login(
s_consumer,
consumer_name,
consumer_pass,
label="consumer",
)
consumer_user_id = pick_user_id(consumer_login_resp)
print(f" Consumer User ID: {consumer_user_id}")
admin_id = phase3_admin_and_verification(
s_admin,
s_user,
s_consumer,
admin_name,
admin_pass,
)
@@ -848,50 +986,14 @@ def main():
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)
shop_id = phase7_shop(s_user, user_id, 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)
phase8_order(s_consumer, s_user, 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())
phase14_reset_password(consumer_name, consumer_email, consumer_new_pass)
phase15_player_service_delete(s_user, service_id)
phase13_logout(s_user)