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

fix: 更新 dev 接口测试脚本

Browse Source
This commit is contained in:
zetaloop
2026-04-23 13:52:53 +08:00
parent 9bc86dd904
commit 3989491b8c
1 changed files with 193 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/dev/test_all_apis.py
+191 -8
View File
@@ -74,6 +74,41 @@ def read_vcode_from_redis(request_id, scene, account):
return result.stdout.strip()
def build_multipart_form(fields, files):
boundary = f"----juwan{rand_str(12)}"
chunks = []
for name, value in fields.items():
chunks.extend(
[
f"--{boundary}\r\n".encode(),
f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(),
str(value).encode(),
b"\r\n",
]
)
for name, (filename, content, content_type) in files.items():
if isinstance(content, str):
content = content.encode()
chunks.extend(
[
f"--{boundary}\r\n".encode(),
(
f'Content-Disposition: form-data; name="{name}"; '
f'filename="{filename}"\r\n'
).encode(),
f"Content-Type: {content_type}\r\n\r\n".encode(),
content,
b"\r\n",
]
)
chunks.append(f"--{boundary}--\r\n".encode())
body = b"".join(chunks)
return f"multipart/form-data; boundary={boundary}", body
class Session:
"""Minimal cookie-aware HTTP session using stdlib only."""
@@ -93,7 +128,15 @@ class Session:
return c.value
return None
def request(self, method, url, json_body=None, headers=None, form_data=None):
def request(
self,
method,
url,
json_body=None,
headers=None,
form_data=None,
raw_body=None,
):
hdrs = headers or {}
body = None
if json_body is not None:
@@ -102,6 +145,8 @@ class Session:
elif form_data is not None:
body = urllib.parse.urlencode(form_data).encode()
hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded")
elif raw_body is not None:
body = raw_body
req = urllib.request.Request(url, data=body, headers=hdrs, method=method)
try:
@@ -126,6 +171,12 @@ class Session:
def post(self, url, **kw):
return self.request("POST", url, **kw)
def post_multipart(self, url, fields, files, headers=None):
hdrs = dict(headers or {})
content_type, body = build_multipart_form(fields, files)
hdrs["Content-Type"] = content_type
return self.post(url, headers=hdrs, raw_body=body)
def put(self, url, **kw):
return self.request("PUT", url, **kw)
@@ -156,6 +207,25 @@ def report(name, status_code, body, expect_status=200):
return ok
def report_check(name, ok, body=None):
global passed, failed
mark = "PASS" if ok else "FAIL"
if not ok:
failed += 1
errors_list.append((name, "CHECK", 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}: {body_preview}")
return ok
def skip(name, reason):
print(f" [SKIP] {name}: {reason}")
# ============================================================
# Phase 0: Health check & CSRF
# ============================================================
@@ -399,6 +469,66 @@ def phase3_admin_and_verification(
return admin_id
def phase3b_secondary_player(s_admin: Session, s_player: Session):
print("\n=== Phase 3b: Secondary Player Setup ===")
csrf_player = s_player.csrf_headers()
code, body, _ = s_player.post(
f"{GATEWAY}/api/v1/users/me/verification",
json_body={
"role": "player",
"materials": {
"idCardFront": "http://example.com/player-front.jpg",
"idCardBack": "http://example.com/player-back.jpg",
"gameScreenshots": ["http://example.com/player-ss1.jpg"],
"voiceDemo": "http://example.com/player-voice.mp3",
},
},
headers=csrf_player,
)
report("POST /users/me/verification (apply player, invited user)", code, body)
code, body, _ = s_player.get(f"{GATEWAY}/api/v1/users/me/verification")
report("GET /users/me/verification (invited user)", code, body)
verification_id = 0
for item in pick_items(body):
if item.get("role") == "player":
verification_id = as_int(item.get("id"))
break
report_check(
"lookup player verification id (invited user)",
bool(verification_id),
{"id": verification_id},
)
if verification_id:
code, body, _ = s_admin.post(
f"{GATEWAY}/api/v1/admin/verifications/{verification_id}/approve",
json_body={},
headers=s_admin.csrf_headers(),
)
report(
f"POST /admin/verifications/{verification_id}/approve (invited user player)",
code,
body,
)
code, body, _ = s_player.post(
f"{GATEWAY}/api/v1/users/me/switch-role",
json_body={"role": "player"},
headers=s_player.csrf_headers(),
)
report("POST /users/me/switch-role (invited user player)", code, body)
code, body, _ = s_player.post(
f"{GATEWAY}/api/v1/players/me",
json_body={},
headers=s_player.csrf_headers(),
)
report("POST /players/me (invited user init)", code, body)
return as_int(body.get("id"))
def phase4_follow(s: Session, target_user_id):
print("\n=== Phase 4: Follow/Unfollow ===")
csrf = s.csrf_headers()
@@ -517,7 +647,7 @@ def phase6_player(s: Session, game_id):
return player_id, service_id
def phase7_shop(s_owner: Session, owner_user_id, player_id):
def phase7_shop(s_owner: Session, owner_user_id, invited_player_id):
print("\n=== Phase 7: Shop ===")
s_owner.post(
@@ -569,18 +699,38 @@ def phase7_shop(s_owner: Session, owner_user_id, player_id):
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/users/{owner_user_id}/shop")
report(f"GET /users/{owner_user_id}/shop", code, body)
announcement = f"Grand opening {rand_str(4)}!"
code, body, _ = s_owner.post(
f"{GATEWAY}/api/v1/shops/{shop_id}/announcements",
json_body={"content": "Grand opening!"},
json_body={"content": announcement},
headers=csrf,
)
report(f"POST /shops/{shop_id}/announcements", code, body)
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
report(f"GET /shops/{shop_id} (after announcement)", code, body)
announcement_index = -1
announcements = body.get("announcements") if isinstance(body, dict) else None
if isinstance(announcements, list):
for idx in range(len(announcements) - 1, -1, -1):
if announcement in str(announcements[idx]):
announcement_index = idx
break
report_check(
f"locate announcement index ({shop_id})",
announcement_index >= 0,
{"index": announcement_index},
)
if announcement_index >= 0:
code, body, _ = s_owner.delete(
f"{GATEWAY}/api/v1/shops/{shop_id}/announcements/0",
f"{GATEWAY}/api/v1/shops/{shop_id}/announcements/{announcement_index}",
headers=csrf,
)
report(f"DELETE /shops/{shop_id}/announcements/0", code, body)
report(
f"DELETE /shops/{shop_id}/announcements/{announcement_index}",
code,
body,
)
code, body, _ = s_owner.put(
f"{GATEWAY}/api/v1/shops/{shop_id}/template",
@@ -594,13 +744,25 @@ def phase7_shop(s_owner: Session, owner_user_id, player_id):
)
report(f"GET /shops/{shop_id}/income-stats", code, body)
if player_id:
if invited_player_id:
code, body, _ = s_owner.post(
f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
json_body={"playerId": player_id},
json_body={"playerId": invited_player_id},
headers=csrf,
)
report(f"POST /shops/{shop_id}/invitations", code, body)
skip(
"POST /shops/invitations/:id/accept",
"create invitation does not return invitation id",
)
skip(
"DELETE /shops/invitations/:id",
"create invitation does not return invitation id",
)
skip(
f"DELETE /shops/{shop_id}/players/{invited_player_id}",
"player removal depends on accepted invitation flow",
)
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/mine")
report("GET /shops/mine", code, body)
@@ -825,6 +987,26 @@ def phase10_community(s: Session, user_id):
def phase11_objectstory(s: Session):
print("\n=== Phase 11: Objectstory (File) ===")
code, body, _ = s.post_multipart(
f"{GATEWAY}/api/v1/upload",
fields={"type": "post"},
files={
"file": (
f"test-{rand_str(4)}.txt",
f"juwan-objectstory-{rand_str(8)}",
"text/plain",
)
},
headers=s.csrf_headers(),
)
report("POST /upload", code, body)
if code == 200:
report_check(
"POST /upload returned url",
bool(body.get("url")),
{"url": body.get("url", "")},
)
code, body, _ = s.get(f"{GATEWAY}/api/v1/files?key=nonexistent")
report(
"GET /files?key=nonexistent (expect error)",
@@ -974,6 +1156,7 @@ def main():
admin_name,
admin_pass,
)
invited_player_id = phase3b_secondary_player(s_admin, s_consumer)
if admin_id and user_id:
phase4_follow(s_user, admin_id)
@@ -986,7 +1169,7 @@ 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, user_id, player_id)
shop_id = phase7_shop(s_user, user_id, invited_player_id)
phase8_order(s_consumer, s_user, player_id, service_id, shop_id)
phase9_wallet(s_user)