zetaloop
1517 lines
48 KiB
Python
1517 lines
48 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 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 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 ""
|
|
|
|
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()
|
|
|
|
|
|
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."""
|
|
|
|
def __init__(self):
|
|
self.cookie_jar = http.cookiejar.CookieJar(
|
|
policy=http.cookiejar.DefaultCookiePolicy(
|
|
secure_protocols=("https", "http")
|
|
)
|
|
)
|
|
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,
|
|
raw_body=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")
|
|
elif raw_body is not None:
|
|
body = raw_body
|
|
|
|
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 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)
|
|
|
|
def delete(self, url, **kw):
|
|
return self.request("DELETE", url, **kw)
|
|
|
|
def csrf_headers(self):
|
|
token = self.get_cookie("__Host-XSRF-TOKEN")
|
|
return {"xsrf-token": token} if token else {}
|
|
|
|
|
|
def report(name, status_code, body, expect_status=200):
|
|
global passed, failed
|
|
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
|
|
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
|
|
|
|
|
|
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
|
|
# ============================================================
|
|
def phase0_health(s: Session):
|
|
print("\n=== Phase 0: Health & CSRF ===")
|
|
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")
|
|
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, label="user"):
|
|
print(f"\n=== Phase 1: Register ({label}) ===")
|
|
|
|
# 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(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")
|
|
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(f"POST /auth/register ({label})", code, body)
|
|
return body
|
|
|
|
|
|
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(f"POST /auth/login ({label})", 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,
|
|
s_reject: 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)
|
|
|
|
code, body, _ = s_admin.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (admin)", code, body)
|
|
admin_id = pick_user_id(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)
|
|
|
|
# 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)
|
|
|
|
# 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={},
|
|
headers=s_admin.csrf_headers(),
|
|
)
|
|
report(
|
|
f"POST /admin/verifications/{vid}/approve ({role})",
|
|
code,
|
|
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",
|
|
json_body={"role": "player"},
|
|
headers=csrf_user,
|
|
)
|
|
report("POST /users/me/switch-role (player)", code, body)
|
|
|
|
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()
|
|
|
|
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)
|
|
|
|
code, body2, _ = s.get(f"{GATEWAY}/api/v1/games")
|
|
report("GET /games (after create)", code, body2)
|
|
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}")
|
|
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, s_invited_player: Session, owner_user_id, invited_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}",
|
|
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)
|
|
|
|
announcement = f"Grand opening {rand_str(4)}!"
|
|
code, body, _ = s_owner.post(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/announcements",
|
|
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/{announcement_index}",
|
|
headers=csrf,
|
|
)
|
|
report(
|
|
f"DELETE /shops/{shop_id}/announcements/{announcement_index}",
|
|
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 invited_player_id:
|
|
code, body, _ = s_owner.post(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
|
|
json_body={"playerId": invited_player_id},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /shops/{shop_id}/invitations", code, body)
|
|
|
|
code, body, _ = s_owner.get(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
|
|
)
|
|
report(f"GET /shops/{shop_id}/invitations", code, body)
|
|
|
|
invitation_id = 0
|
|
for item in pick_items(body):
|
|
if as_int(item.get("playerId")) == invited_player_id:
|
|
invitation_id = as_int(item.get("id"))
|
|
break
|
|
report_check(
|
|
"locate invitation id",
|
|
bool(invitation_id),
|
|
{"id": invitation_id},
|
|
)
|
|
|
|
s_invited_player.post(
|
|
f"{GATEWAY}/api/v1/users/me/switch-role",
|
|
json_body={"role": "player"},
|
|
headers=s_invited_player.csrf_headers(),
|
|
)
|
|
code, body, _ = s_invited_player.get(
|
|
f"{GATEWAY}/api/v1/shops/invitations/mine",
|
|
)
|
|
report("GET /shops/invitations/mine", code, body)
|
|
|
|
if invitation_id:
|
|
inv_csrf = s_invited_player.csrf_headers()
|
|
code, body, _ = s_invited_player.post(
|
|
f"{GATEWAY}/api/v1/shops/invitations/{invitation_id}/accept",
|
|
json_body={},
|
|
headers=inv_csrf,
|
|
)
|
|
report(
|
|
f"POST /shops/invitations/{invitation_id}/accept",
|
|
code,
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_owner.delete(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/players/{invited_player_id}",
|
|
headers=csrf,
|
|
)
|
|
report(
|
|
f"DELETE /shops/{shop_id}/players/{invited_player_id}",
|
|
code,
|
|
body,
|
|
)
|
|
|
|
# Reject invitation flow: re-invite the removed player, then reject
|
|
code, body, _ = s_owner.post(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
|
|
json_body={"playerId": invited_player_id},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /shops/{shop_id}/invitations (re-invite)", code, body)
|
|
|
|
code, body, _ = s_owner.get(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
|
|
)
|
|
reinvite_id = 0
|
|
for item in pick_items(body):
|
|
if (
|
|
as_int(item.get("playerId")) == invited_player_id
|
|
and item.get("status") == "pending"
|
|
):
|
|
reinvite_id = as_int(item.get("id"))
|
|
break
|
|
if reinvite_id:
|
|
code, body, _ = s_invited_player.delete(
|
|
f"{GATEWAY}/api/v1/shops/invitations/{reinvite_id}",
|
|
headers=s_invited_player.csrf_headers(),
|
|
)
|
|
report(
|
|
f"DELETE /shops/invitations/{reinvite_id} (reject)",
|
|
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, s_actor: 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)
|
|
|
|
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)
|
|
|
|
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_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}/confirm-close",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order_id}/confirm-close", 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 order_id:
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order_id}/reorder",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order_id}/reorder", code, body)
|
|
|
|
if order2_id:
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order2_id}/pay",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order2_id}/pay (before cancel)", code, body)
|
|
|
|
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",
|
|
json_body={
|
|
"playerId": player_id,
|
|
"serviceId": service_id,
|
|
"quantity": 1,
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report("POST /orders/paid (create+pay)", code, body)
|
|
|
|
return order_id
|
|
|
|
|
|
def phase8b_review(s_consumer: Session, order_id, player_user_id):
|
|
print("\n=== Phase 8b: Reviews ===")
|
|
if not order_id:
|
|
skip("Review flow", "No pending_review order id")
|
|
return
|
|
|
|
csrf = s_consumer.csrf_headers()
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order_id}/review",
|
|
json_body={"rating": 5, "content": "great service"},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order_id}/review", code, body)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders/{order_id}/reviews")
|
|
report(f"GET /orders/{order_id}/reviews", code, body)
|
|
if code == 200:
|
|
items = pick_items(body)
|
|
report_check(
|
|
f"GET /orders/{order_id}/reviews shape",
|
|
isinstance(items, list) and isinstance(body.get("meta"), dict),
|
|
body,
|
|
)
|
|
report_check(
|
|
f"GET /orders/{order_id}/reviews hides sealed review",
|
|
len(items) == 0,
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/reviews?limit=20")
|
|
report("GET /reviews?limit=20", code, body)
|
|
|
|
if player_user_id:
|
|
code, body, _ = s_consumer.get(
|
|
f"{GATEWAY}/api/v1/users/{player_user_id}/reviews?limit=20",
|
|
)
|
|
report(f"GET /users/{player_user_id}/reviews?limit=20", code, body)
|
|
|
|
|
|
def phase8c_dispute(s_consumer: Session, s_actor: Session, player_id, service_id, shop_id):
|
|
print("\n=== Phase 8c: Disputes ===")
|
|
if not player_id or not service_id:
|
|
skip("Dispute flow", "Missing player or service id")
|
|
return 0
|
|
|
|
csrf = s_consumer.csrf_headers()
|
|
payload = {
|
|
"playerId": player_id,
|
|
"serviceId": service_id,
|
|
"quantity": 1,
|
|
"note": "test dispute order",
|
|
}
|
|
if shop_id:
|
|
payload["shopId"] = shop_id
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders",
|
|
json_body=payload,
|
|
headers=csrf,
|
|
)
|
|
report("POST /orders (create for dispute)", code, body)
|
|
order_obj = body.get("order", {}) if isinstance(body, dict) else {}
|
|
order_id = order_obj.get("id", 0) if isinstance(order_obj, dict) else 0
|
|
if not order_id:
|
|
skip("Dispute flow", "No order id returned")
|
|
return 0
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order_id}/pay",
|
|
json_body={},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order_id}/pay (dispute flow)", 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 (dispute flow)", code, body)
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/orders/{order_id}/dispute",
|
|
json_body={
|
|
"reason": "test dispute reason",
|
|
"evidence": ["http://example.com/evidence.jpg"],
|
|
},
|
|
headers=csrf,
|
|
)
|
|
report(f"POST /orders/{order_id}/dispute", code, body)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders/{order_id}/dispute")
|
|
report(f"GET /orders/{order_id}/dispute", code, body)
|
|
dispute_id = as_int(body.get("id")) if isinstance(body, dict) else 0
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/disputes")
|
|
report("GET /disputes", code, body)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/disputes?status=open")
|
|
report("GET /disputes?status=open", code, body)
|
|
|
|
if dispute_id:
|
|
code, body, _ = s_actor.post(
|
|
f"{GATEWAY}/api/v1/disputes/{dispute_id}/response",
|
|
json_body={
|
|
"reason": "test respondent response",
|
|
"evidence": ["http://example.com/response.jpg"],
|
|
},
|
|
headers=s_actor.csrf_headers(),
|
|
)
|
|
report(
|
|
f"POST /disputes/{dispute_id}/response",
|
|
code,
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_consumer.post(
|
|
f"{GATEWAY}/api/v1/disputes/{dispute_id}/appeal",
|
|
json_body={"reason": "test appeal guard"},
|
|
headers=csrf,
|
|
)
|
|
report(
|
|
f"POST /disputes/{dispute_id}/appeal (expect status check)",
|
|
code,
|
|
body,
|
|
expect_status=(400, 403, 500),
|
|
)
|
|
|
|
return dispute_id
|
|
|
|
|
|
def phase8d_notifications(s: Session):
|
|
print("\n=== Phase 8d: Notifications ===")
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/notifications")
|
|
report("GET /notifications", code, body)
|
|
|
|
items = pick_items(body) if code == 200 and isinstance(body, dict) else []
|
|
if items:
|
|
notification_id = as_int(items[0].get("id"))
|
|
if notification_id:
|
|
code, body, _ = s.put(
|
|
f"{GATEWAY}/api/v1/notifications/{notification_id}/read",
|
|
json_body={},
|
|
headers=s.csrf_headers(),
|
|
)
|
|
report(f"PUT /notifications/{notification_id}/read", code, body)
|
|
else:
|
|
skip("PUT /notifications/:id/read", "No notification item returned")
|
|
|
|
code, body, _ = s.put(
|
|
f"{GATEWAY}/api/v1/notifications/read-all",
|
|
json_body={},
|
|
headers=s.csrf_headers(),
|
|
)
|
|
report("PUT /notifications/read-all", code, body)
|
|
|
|
|
|
def phase8e_search_and_favorites(s: Session, user_id, player_id, shop_id):
|
|
print("\n=== Phase 8e: Search & Favorites ===")
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/search?q=LOL&limit=10")
|
|
report("GET /search?q=LOL&limit=10", code, body)
|
|
if code == 200:
|
|
report_check(
|
|
"GET /search response shape",
|
|
isinstance(pick_items(body), list) and isinstance(body.get("meta"), dict),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/recommendations/home?limit=10")
|
|
report("GET /recommendations/home?limit=10", code, body)
|
|
if code == 200:
|
|
report_check(
|
|
"GET /recommendations/home response shape",
|
|
isinstance(pick_items(body), list) and isinstance(body.get("meta"), dict),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/favorites")
|
|
report("GET /favorites", code, body)
|
|
|
|
if not player_id or not user_id:
|
|
skip("Favorites mutation flow", "Missing user or player id")
|
|
return
|
|
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/favorites",
|
|
json_body={"targetType": "player", "targetId": str(player_id)},
|
|
headers=s.csrf_headers(),
|
|
)
|
|
report("POST /favorites (player)", code, body)
|
|
|
|
code, body, _ = s.get(
|
|
f"{GATEWAY}/api/v1/users/{user_id}/favorites/check"
|
|
f"?targetType=player&targetId={player_id}",
|
|
)
|
|
report(f"GET /users/{user_id}/favorites/check (player)", code, body)
|
|
if code == 200:
|
|
report_check("favorite check after add", body.get("favorited") is True, body)
|
|
|
|
if shop_id:
|
|
code, body, _ = s.post(
|
|
f"{GATEWAY}/api/v1/favorites",
|
|
json_body={"targetType": "shop", "targetId": str(shop_id)},
|
|
headers=s.csrf_headers(),
|
|
)
|
|
report("POST /favorites (shop)", code, body)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/favorites")
|
|
report("GET /favorites (after add)", code, body)
|
|
favorite_id = 0
|
|
for item in pick_items(body):
|
|
if (
|
|
item.get("targetType") == "player"
|
|
and str(item.get("targetId")) == str(player_id)
|
|
):
|
|
favorite_id = as_int(item.get("id"))
|
|
break
|
|
report_check("locate favorite id", bool(favorite_id), {"id": favorite_id})
|
|
|
|
if favorite_id:
|
|
code, body, _ = s.delete(
|
|
f"{GATEWAY}/api/v1/favorites/{favorite_id}",
|
|
headers=s.csrf_headers(),
|
|
)
|
|
report(f"DELETE /favorites/{favorite_id}", code, body)
|
|
|
|
code, body, _ = s.get(
|
|
f"{GATEWAY}/api/v1/users/{user_id}/favorites/check"
|
|
f"?targetType=player&targetId={player_id}",
|
|
)
|
|
report(f"GET /users/{user_id}/favorites/check (after delete)", code, body)
|
|
if code == 200:
|
|
report_check("favorite check after delete", body.get("favorited") is False, body)
|
|
|
|
|
|
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.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)",
|
|
code,
|
|
body,
|
|
expect_status=(400, 500),
|
|
)
|
|
|
|
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_reset_password(username, email, new_password):
|
|
print("\n=== Phase 14: Forgot/Reset Password ===")
|
|
|
|
s_reset = Session()
|
|
s_reset.get(f"{GATEWAY}/healthz")
|
|
csrf = s_reset.csrf_headers()
|
|
|
|
code, body, _ = s_reset.post(
|
|
f"{GATEWAY}/api/v1/auth/forgot-password/send",
|
|
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
|
|
|
|
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": email,
|
|
"vcode": vcode,
|
|
"newPassword": new_password,
|
|
},
|
|
headers=csrf,
|
|
)
|
|
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):
|
|
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!"
|
|
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
|
|
|
|
print(f"Test run: user={user1_name}, admin={admin_name}")
|
|
|
|
s_user = Session()
|
|
s_admin = Session()
|
|
s_consumer = Session()
|
|
|
|
phase0_health(s_user)
|
|
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 = pick_user_id(login_resp)
|
|
print(f" User ID: {user_id}")
|
|
|
|
phase2_user(s_user, user_id)
|
|
|
|
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,
|
|
)
|
|
invited_player_id = phase3b_secondary_player(s_admin, s_consumer)
|
|
|
|
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, s_consumer, user_id, invited_player_id)
|
|
|
|
order_id = phase8_order(s_consumer, s_user, player_id, service_id, shop_id)
|
|
phase8b_review(s_consumer, order_id, user_id)
|
|
phase8c_dispute(s_consumer, s_user, player_id, service_id, shop_id)
|
|
phase8d_notifications(s_consumer)
|
|
phase8e_search_and_favorites(s_consumer, consumer_user_id, player_id, shop_id)
|
|
phase9_wallet(s_user)
|
|
phase10_community(s_user, user_id)
|
|
phase11_objectstory(s_user)
|
|
phase12_email(s_user)
|
|
phase14_reset_password(consumer_name, consumer_email, consumer_new_pass)
|
|
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()
|