zetaloop
2253 lines
76 KiB
Python
2253 lines
76 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
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
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 as_decimal(value):
|
|
try:
|
|
return Decimal(str(value))
|
|
except (InvalidOperation, TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def same_id(left, right):
|
|
return str(left) == str(right)
|
|
|
|
|
|
def pick_items(body):
|
|
if isinstance(body.get("items"), list):
|
|
return body["items"]
|
|
if isinstance(body.get("list"), list):
|
|
return body["list"]
|
|
return []
|
|
|
|
|
|
def find_item(items, predicate):
|
|
for item in items:
|
|
if predicate(item):
|
|
return item
|
|
return None
|
|
|
|
|
|
def find_item_by_id(items, item_id):
|
|
return find_item(items, lambda item: same_id(item.get("id"), item_id))
|
|
|
|
|
|
def user_from(body):
|
|
user = body.get("user") if isinstance(body, dict) else None
|
|
return user if isinstance(user, dict) else {}
|
|
|
|
|
|
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}")
|
|
|
|
|
|
def check_order_status(session, order_id, expected_status, label):
|
|
code, body, _ = session.get(f"{GATEWAY}/api/v1/orders/{order_id}")
|
|
report(f"GET /orders/{order_id} ({label})", code, body)
|
|
report_check(
|
|
f"order {order_id} status is {expected_status} ({label})",
|
|
code == 200 and body.get("status") == expected_status,
|
|
body,
|
|
)
|
|
return body
|
|
|
|
|
|
# ============================================================
|
|
# 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)
|
|
user = user_from(body)
|
|
report_check(
|
|
f"registered user fields ({label})",
|
|
code == 200
|
|
and as_int(user.get("id")) > 0
|
|
and user.get("username") == username
|
|
and user.get("role") == "consumer"
|
|
and "consumer" in (user.get("verifiedRoles") or []),
|
|
user,
|
|
)
|
|
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")
|
|
user = user_from(body)
|
|
report_check(
|
|
f"login session fields ({label})",
|
|
code == 200
|
|
and bool(jtoken)
|
|
and as_int(user.get("id")) > 0
|
|
and user.get("username") == username,
|
|
{"user": user, "hasJToken": bool(jtoken)},
|
|
)
|
|
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)
|
|
report_check(
|
|
"GET /users/me returns current user",
|
|
code == 200 and same_id(body.get("id"), user_id),
|
|
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)
|
|
report_check(
|
|
"PUT /users/me persists profile fields",
|
|
code == 200
|
|
and same_id(body.get("id"), user_id)
|
|
and body.get("nickname") == "TestNick"
|
|
and body.get("bio") == "testbio",
|
|
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)
|
|
report_check(
|
|
f"GET /users/{user_id} exposes updated public profile",
|
|
code == 200
|
|
and same_id(body.get("id"), user_id)
|
|
and body.get("nickname") == "TestNick"
|
|
and body.get("bio") == "testbio",
|
|
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"))
|
|
report_check(
|
|
"verification applications are pending",
|
|
all(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item, role=role: item.get("role") == role
|
|
and item.get("status") == "pending",
|
|
)
|
|
for role in ("player", "owner")
|
|
),
|
|
body,
|
|
)
|
|
|
|
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"))
|
|
report_check(
|
|
"reject user verification is pending",
|
|
bool(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item: item.get("role") == "owner"
|
|
and item.get("status") == "pending",
|
|
)
|
|
),
|
|
body,
|
|
)
|
|
|
|
# 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)
|
|
report_check(
|
|
"rejected verification keeps reason",
|
|
bool(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item: item.get("role") == "owner"
|
|
and item.get("status") == "rejected"
|
|
and item.get("rejectReason") == "test reject flow",
|
|
)
|
|
),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_user.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (after approval)", code, body)
|
|
verified_roles = body.get("verifiedRoles") or []
|
|
verification_status = body.get("verificationStatus") or {}
|
|
report_check(
|
|
"approved roles are visible on current user",
|
|
"player" in verified_roles
|
|
and "owner" in verified_roles
|
|
and verification_status.get("player") == "approved"
|
|
and verification_status.get("owner") == "approved",
|
|
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)
|
|
code, body, _ = s_user.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (after switch player)", code, body)
|
|
report_check("current role is player", body.get("role") == "player", 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},
|
|
)
|
|
report_check(
|
|
"invited user player verification is pending",
|
|
bool(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item: item.get("role") == "player"
|
|
and item.get("status") == "pending",
|
|
)
|
|
),
|
|
body,
|
|
)
|
|
|
|
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.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (invited user player role)", code, body)
|
|
report_check("invited user current role is player", body.get("role") == "player", 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)
|
|
report_check(
|
|
"invited player profile initialized",
|
|
as_int(body.get("id")) > 0
|
|
and as_int((body.get("user") or {}).get("id")) > 0
|
|
and body.get("status") == "offline",
|
|
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)
|
|
report_check("GET /games returns list shape", isinstance(pick_items(body), list), body)
|
|
|
|
csrf_admin = s_admin.csrf_headers()
|
|
game_name = f"TestGame_{rand_str(4)}"
|
|
game_icon = "icon.png"
|
|
game_category = "MOBA"
|
|
code, body, _ = s_admin.post(
|
|
f"{GATEWAY}/api/v1/games",
|
|
json_body={
|
|
"name": game_name,
|
|
"icon": game_icon,
|
|
"category": game_category,
|
|
},
|
|
headers=csrf_admin,
|
|
)
|
|
report("POST /games (create)", code, body)
|
|
game_id = body.get("id", 0)
|
|
report_check(
|
|
"created game fields",
|
|
as_int(game_id) > 0
|
|
and body.get("name") == game_name
|
|
and body.get("icon") == game_icon
|
|
and body.get("category") == game_category,
|
|
body,
|
|
)
|
|
|
|
code, body2, _ = s.get(f"{GATEWAY}/api/v1/games")
|
|
report("GET /games (after create)", code, body2)
|
|
items = pick_items(body2)
|
|
created_game = find_item_by_id(items, game_id)
|
|
report_check(
|
|
"created game appears in list",
|
|
bool(created_game) and created_game.get("name") == game_name,
|
|
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)
|
|
report_check(
|
|
f"GET /games/{game_id} returns created game",
|
|
code == 200
|
|
and same_id(body.get("id"), game_id)
|
|
and body.get("name") == game_name
|
|
and body.get("category") == game_category,
|
|
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)
|
|
report_check(
|
|
"player profile initialized",
|
|
as_int(player_id) > 0
|
|
and body.get("status") == "offline"
|
|
and as_int((body.get("user") or {}).get("id")) > 0,
|
|
body,
|
|
)
|
|
|
|
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)
|
|
report_check("GET /players returns list shape", isinstance(pick_items(body), list), body)
|
|
|
|
if player_id:
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/players/{player_id}")
|
|
report(f"GET /players/{player_id}", code, body)
|
|
report_check(
|
|
f"GET /players/{player_id} returns initialized online player",
|
|
code == 200
|
|
and same_id(body.get("id"), player_id)
|
|
and body.get("status") == "online"
|
|
and body.get("gender") is True
|
|
and as_int((body.get("user") or {}).get("id")) > 0
|
|
and isinstance(body.get("services"), list),
|
|
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
|
|
report_check(
|
|
"created service fields",
|
|
code == 200
|
|
and as_int(body.get("id")) > 0
|
|
and same_id(body.get("playerId"), player_id)
|
|
and same_id(body.get("gameId"), game_id)
|
|
and body.get("title") == "Boosting Service"
|
|
and body.get("description") == "Rank boost"
|
|
and as_decimal(body.get("price")) == Decimal("50")
|
|
and body.get("unit") == "game",
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/services")
|
|
report("GET /services", code, body)
|
|
report_check("GET /services returns list shape", isinstance(pick_items(body), list), 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)
|
|
report_check(
|
|
f"GET /services/{service_id} returns created service",
|
|
code == 200
|
|
and same_id(body.get("id"), service_id)
|
|
and body.get("title") == "Boosting Service",
|
|
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)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/services/{service_id}")
|
|
report(f"GET /services/{service_id} (after update)", code, body)
|
|
report_check(
|
|
f"GET /services/{service_id} reflects update",
|
|
code == 200
|
|
and body.get("title") == "Updated Service"
|
|
and as_decimal(body.get("price")) == Decimal("60"),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/players/{player_id}")
|
|
report(f"GET /players/{player_id} (after service update)", code, body)
|
|
service = find_item_by_id(body.get("services") or [], service_id)
|
|
report_check(
|
|
f"GET /players/{player_id} includes updated service",
|
|
code == 200
|
|
and body.get("gender") is True
|
|
and bool(service)
|
|
and service.get("title") == "Updated Service"
|
|
and as_decimal(service.get("price")) == Decimal("60"),
|
|
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)
|
|
report_check(
|
|
f"GET /players/{player_id}/services contains service",
|
|
bool(find_item_by_id(pick_items(body), service_id)),
|
|
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(),
|
|
)
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (owner role)", code, body)
|
|
report_check("current role is owner", body.get("role") == "owner", body)
|
|
|
|
csrf = s_owner.csrf_headers()
|
|
shop_name = f"TestShop_{rand_str(4)}"
|
|
code, body, _ = s_owner.post(
|
|
f"{GATEWAY}/api/v1/shops",
|
|
json_body={
|
|
"name": shop_name,
|
|
"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
|
|
report_check(
|
|
"created shop fields",
|
|
as_int(shop_id) > 0
|
|
and body.get("name") == shop_name
|
|
and body.get("description") == "A test shop"
|
|
and body.get("commissionType") == "percentage"
|
|
and body.get("commissionValue") == "10"
|
|
and same_id((body.get("owner") or {}).get("id"), owner_user_id),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops")
|
|
report("GET /shops", code, body)
|
|
report_check(
|
|
"created shop appears in list",
|
|
bool(find_item_by_id(pick_items(body), shop_id)),
|
|
body,
|
|
)
|
|
|
|
if shop_id:
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
|
|
report(f"GET /shops/{shop_id}", code, body)
|
|
report_check(
|
|
f"GET /shops/{shop_id} returns created shop",
|
|
same_id(body.get("id"), shop_id) and body.get("name") == shop_name,
|
|
body,
|
|
)
|
|
|
|
updated_shop_name = f"UpdatedShop_{rand_str(4)}"
|
|
code, body, _ = s_owner.put(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}",
|
|
json_body={
|
|
"name": updated_shop_name,
|
|
"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)
|
|
report_check(
|
|
f"PUT /shops/{shop_id} returns updated fields",
|
|
same_id(body.get("id"), shop_id)
|
|
and body.get("name") == updated_shop_name
|
|
and body.get("description") == "An updated test shop"
|
|
and body.get("commissionValue") == "12"
|
|
and body.get("allowMultiShop") is True
|
|
and body.get("allowIndependentOrders") is False
|
|
and body.get("dispatchMode") == "manual",
|
|
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)
|
|
report_check(
|
|
f"GET /users/{owner_user_id}/shop returns owned shop",
|
|
same_id(body.get("id"), shop_id)
|
|
and same_id((body.get("owner") or {}).get("id"), owner_user_id),
|
|
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.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
|
|
report(f"GET /shops/{shop_id} (after announcement delete)", code, body)
|
|
report_check(
|
|
"deleted announcement is absent",
|
|
announcement not in [str(item) for item in body.get("announcements") or []],
|
|
body,
|
|
)
|
|
|
|
template_sections = json.dumps({"layout": "grid", "theme": "dark"})
|
|
code, body, _ = s_owner.put(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/template",
|
|
json_body={"sections": template_sections},
|
|
headers=csrf,
|
|
)
|
|
report(f"PUT /shops/{shop_id}/template", code, body)
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
|
|
report(f"GET /shops/{shop_id} (after template)", code, body)
|
|
report_check(
|
|
"shop template is persisted",
|
|
bool(body.get("templateConfig")),
|
|
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)
|
|
report_check(
|
|
"income stats response has business fields",
|
|
all(
|
|
key in body
|
|
for key in (
|
|
"monthlyIncome",
|
|
"pendingSettlement",
|
|
"totalWithdrawn",
|
|
"totalOrders",
|
|
"completedOrders",
|
|
)
|
|
),
|
|
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},
|
|
)
|
|
invitation = find_item_by_id(pick_items(body), invitation_id)
|
|
report_check(
|
|
"created invitation is pending",
|
|
bool(invitation)
|
|
and same_id(invitation.get("shopId"), shop_id)
|
|
and same_id(invitation.get("playerId"), invited_player_id)
|
|
and invitation.get("status") == "pending"
|
|
and same_id(invitation.get("invitedBy"), owner_user_id),
|
|
body,
|
|
)
|
|
|
|
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.get(
|
|
f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
|
|
)
|
|
report(f"GET /shops/{shop_id}/invitations (after accept)", code, body)
|
|
accepted = find_item_by_id(pick_items(body), invitation_id)
|
|
report_check(
|
|
"accepted invitation has accepted status",
|
|
bool(accepted)
|
|
and accepted.get("status") == "accepted"
|
|
and as_int(accepted.get("respondedAt")) > 0,
|
|
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,
|
|
)
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
|
|
report(f"GET /shops/{shop_id} (after player remove)", code, body)
|
|
report_check(
|
|
"removed player is reflected in shop profile",
|
|
as_int(body.get("playerCount")) >= 0,
|
|
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/{shop_id}/invitations",
|
|
)
|
|
report(f"GET /shops/{shop_id}/invitations (after reject)", code, body)
|
|
rejected = find_item_by_id(pick_items(body), reinvite_id)
|
|
report_check(
|
|
"rejected invitation has rejected status",
|
|
bool(rejected)
|
|
and rejected.get("status") == "rejected"
|
|
and as_int(rejected.get("respondedAt")) > 0,
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/mine")
|
|
report("GET /shops/mine", code, body)
|
|
report_check("GET /shops/mine returns owned shop", same_id(body.get("id"), shop_id), 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(),
|
|
)
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (consumer role)", code, body)
|
|
consumer_user_id = pick_user_id(body)
|
|
report_check("current role is consumer", body.get("role") == "consumer", body)
|
|
|
|
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
|
|
report_check(
|
|
"created order starts pending payment",
|
|
body.get("ok") is True
|
|
and as_int(order_id) > 0
|
|
and same_id(order_obj.get("consumerId"), consumer_user_id)
|
|
and same_id(order_obj.get("playerId"), player_id)
|
|
and same_id(order_obj.get("shopId"), shop_id)
|
|
and order_obj.get("status") == "pending_payment"
|
|
and as_decimal(order_obj.get("totalPrice")) == Decimal("50")
|
|
and order_obj.get("note") == "test order",
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders?role=consumer")
|
|
report("GET /orders?role=consumer", code, body)
|
|
report_check(
|
|
"consumer order list contains created order",
|
|
bool(find_item_by_id(pick_items(body), order_id)),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_actor.get(f"{GATEWAY}/api/v1/orders?role=player")
|
|
report("GET /orders?role=player", code, body)
|
|
report_check(
|
|
"player order list contains created order",
|
|
bool(find_item_by_id(pick_items(body), order_id)),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_actor.get(f"{GATEWAY}/api/v1/orders?role=owner")
|
|
report("GET /orders?role=owner", code, body)
|
|
report_check(
|
|
"owner order list contains created order",
|
|
bool(find_item_by_id(pick_items(body), order_id)),
|
|
body,
|
|
)
|
|
|
|
if order_id:
|
|
body = check_order_status(s_consumer, order_id, "pending_payment", "after create")
|
|
report_check(
|
|
"created order detail matches participants",
|
|
same_id(body.get("consumerId"), consumer_user_id)
|
|
and same_id(body.get("playerId"), player_id)
|
|
and same_id(body.get("shopId"), shop_id),
|
|
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)
|
|
check_order_status(s_consumer, order_id, "pending_accept", "after pay")
|
|
|
|
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)
|
|
body = check_order_status(s_consumer, order_id, "in_progress", "after accept")
|
|
report_check("accepted order has acceptedAt", bool(body.get("acceptedAt")), 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)
|
|
check_order_status(s_consumer, order_id, "pending_close", "after request-close")
|
|
|
|
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)
|
|
check_order_status(s_consumer, order_id, "pending_review", "after confirm-close")
|
|
|
|
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
|
|
report_check(
|
|
"second order starts pending payment",
|
|
as_int(order2_id) > 0 and order2.get("status") == "pending_payment",
|
|
body,
|
|
)
|
|
|
|
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)
|
|
reordered = body.get("order", {}) if isinstance(body, dict) else {}
|
|
report_check(
|
|
"reorder creates a new pending order",
|
|
body.get("ok") is True
|
|
and as_int(reordered.get("id")) > 0
|
|
and not same_id(reordered.get("id"), order_id)
|
|
and reordered.get("status") == "pending_payment",
|
|
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)
|
|
check_order_status(s_consumer, order2_id, "pending_accept", "second order after pay")
|
|
|
|
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)
|
|
check_order_status(s_consumer, order2_id, "cancelled", "second order after cancel")
|
|
|
|
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)
|
|
paid_order = body.get("order", {}) if isinstance(body, dict) else {}
|
|
report_check(
|
|
"create+pay order starts pending accept",
|
|
body.get("ok") is True
|
|
and as_int(paid_order.get("id")) > 0
|
|
and paid_order.get("status") == "pending_accept",
|
|
body,
|
|
)
|
|
|
|
return order_id
|
|
|
|
|
|
def phase8b_review(
|
|
s_consumer: Session,
|
|
s_actor: Session,
|
|
order_id,
|
|
consumer_user_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_actor.post(
|
|
f"{GATEWAY}/api/v1/orders/{order_id}/review",
|
|
json_body={"rating": 4, "content": "smooth buyer"},
|
|
headers=s_actor.csrf_headers(),
|
|
)
|
|
report(f"POST /orders/{order_id}/review (counterparty)", code, body)
|
|
|
|
order_body = check_order_status(s_consumer, order_id, "completed", "after two reviews")
|
|
report_check("completed order has completedAt", bool(order_body.get("completedAt")), order_body)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders/{order_id}/reviews")
|
|
report(f"GET /orders/{order_id}/reviews (after two reviews)", code, body)
|
|
reviews = pick_items(body)
|
|
report_check(
|
|
"two-sided reviews are visible and unsealed",
|
|
len(reviews) == 2
|
|
and all(same_id(item.get("orderId"), order_id) for item in reviews)
|
|
and all(item.get("sealed") is False for item in reviews)
|
|
and bool(find_item(reviews, lambda item: item.get("content") == "great service"))
|
|
and bool(find_item(reviews, lambda item: item.get("content") == "smooth buyer")),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/reviews?limit=20")
|
|
report("GET /reviews?limit=20", code, body)
|
|
report_check(
|
|
"public review list contains unsealed review",
|
|
bool(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item: same_id(item.get("orderId"), order_id)
|
|
and item.get("sealed") is False,
|
|
)
|
|
),
|
|
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)
|
|
report_check(
|
|
"player received review is listed",
|
|
bool(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item: same_id(item.get("orderId"), order_id)
|
|
and same_id(item.get("fromUserId"), consumer_user_id)
|
|
and item.get("content") == "great service",
|
|
)
|
|
),
|
|
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
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (dispute consumer)", code, body)
|
|
consumer_user_id = pick_user_id(body)
|
|
|
|
code, body, _ = s_actor.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (dispute actor)", code, body)
|
|
actor_user_id = pick_user_id(body)
|
|
|
|
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
|
|
report_check(
|
|
"dispute order starts pending payment",
|
|
as_int(order_id) > 0 and order_obj.get("status") == "pending_payment",
|
|
body,
|
|
)
|
|
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)
|
|
check_order_status(s_consumer, order_id, "pending_accept", "dispute order after pay")
|
|
|
|
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)
|
|
check_order_status(s_consumer, order_id, "in_progress", "dispute order after accept")
|
|
|
|
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)
|
|
check_order_status(s_consumer, order_id, "disputed", "after dispute create")
|
|
|
|
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
|
|
report_check(
|
|
"created dispute fields",
|
|
dispute_id > 0
|
|
and same_id(body.get("orderId"), order_id)
|
|
and same_id(body.get("initiatorId"), consumer_user_id)
|
|
and same_id(body.get("respondentId"), actor_user_id)
|
|
and body.get("reason") == "test dispute reason"
|
|
and body.get("evidence") == ["http://example.com/evidence.jpg"]
|
|
and body.get("status") == "open",
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/disputes")
|
|
report("GET /disputes", code, body)
|
|
report_check(
|
|
"dispute list contains created dispute",
|
|
bool(find_item_by_id(pick_items(body), dispute_id)),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/disputes?status=open")
|
|
report("GET /disputes?status=open", code, body)
|
|
report_check(
|
|
"open dispute list contains created open dispute",
|
|
bool(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item: same_id(item.get("id"), dispute_id)
|
|
and item.get("status") == "open",
|
|
)
|
|
),
|
|
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.get(f"{GATEWAY}/api/v1/orders/{order_id}/dispute")
|
|
report(f"GET /orders/{order_id}/dispute (after response)", code, body)
|
|
report_check(
|
|
"responded dispute fields",
|
|
body.get("status") == "reviewing"
|
|
and body.get("respondentReason") == "test respondent response"
|
|
and body.get("respondentEvidence") == ["http://example.com/response.jpg"],
|
|
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)
|
|
report_check(
|
|
"GET /notifications returns list shape",
|
|
code == 200 and isinstance(pick_items(body), list) and isinstance(body.get("meta"), dict),
|
|
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)
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/notifications")
|
|
report("GET /notifications (after single read)", code, body)
|
|
notification = find_item_by_id(pick_items(body), notification_id)
|
|
report_check(
|
|
"single notification is marked read",
|
|
bool(notification) and notification.get("read") is True,
|
|
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)
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/notifications")
|
|
report("GET /notifications (after read-all)", code, body)
|
|
items = pick_items(body)
|
|
report_check(
|
|
"all returned notifications are read",
|
|
all(item.get("read") is True for item in items),
|
|
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)
|
|
report_check(
|
|
"GET /favorites returns list shape",
|
|
code == 200 and isinstance(pick_items(body), list) and isinstance(body.get("meta"), dict),
|
|
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_items = pick_items(body)
|
|
report_check(
|
|
"player favorite is listed",
|
|
bool(
|
|
find_item(
|
|
favorite_items,
|
|
lambda item: item.get("targetType") == "player"
|
|
and same_id(item.get("targetId"), player_id)
|
|
and same_id(item.get("userId"), user_id),
|
|
)
|
|
),
|
|
body,
|
|
)
|
|
if shop_id:
|
|
report_check(
|
|
"shop favorite is listed",
|
|
bool(
|
|
find_item(
|
|
favorite_items,
|
|
lambda item: item.get("targetType") == "shop"
|
|
and same_id(item.get("targetId"), shop_id)
|
|
and same_id(item.get("userId"), user_id),
|
|
)
|
|
),
|
|
body,
|
|
)
|
|
favorite_id = 0
|
|
for item in favorite_items:
|
|
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)
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/favorites")
|
|
report("GET /favorites (after delete)", code, body)
|
|
report_check(
|
|
"deleted player favorite is absent",
|
|
not bool(
|
|
find_item(
|
|
pick_items(body),
|
|
lambda item: item.get("targetType") == "player"
|
|
and same_id(item.get("targetId"), player_id),
|
|
)
|
|
),
|
|
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)
|
|
initial_balance = as_decimal(body.get("balance")) or Decimal("0")
|
|
report_check(
|
|
"wallet balance response has numeric fields",
|
|
as_decimal(body.get("balance")) is not None
|
|
and as_decimal(body.get("frozenBalance")) is not None,
|
|
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.get(f"{GATEWAY}/api/v1/wallet/balance")
|
|
report("GET /wallet/balance (after topup)", code, body)
|
|
topup_balance = as_decimal(body.get("balance"))
|
|
report_check(
|
|
"wallet balance increases after topup",
|
|
topup_balance == initial_balance + Decimal("100"),
|
|
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/balance")
|
|
report("GET /wallet/balance (after withdraw)", code, body)
|
|
withdraw_balance = as_decimal(body.get("balance"))
|
|
report_check(
|
|
"wallet balance decreases after withdraw",
|
|
topup_balance is not None and withdraw_balance == topup_balance - Decimal("10"),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/wallet/transactions")
|
|
report("GET /wallet/transactions", code, body)
|
|
transactions = pick_items(body)
|
|
report_check(
|
|
"wallet transactions include topup and withdrawal",
|
|
bool(
|
|
find_item(
|
|
transactions,
|
|
lambda item: item.get("type") == "topup"
|
|
and as_decimal(item.get("amount")) == Decimal("100")
|
|
and item.get("description") == "topup via alipay",
|
|
)
|
|
)
|
|
and bool(
|
|
find_item(
|
|
transactions,
|
|
lambda item: item.get("type") == "withdrawal"
|
|
and as_decimal(item.get("amount")) == Decimal("10")
|
|
and item.get("description") == "withdraw via alipay",
|
|
)
|
|
),
|
|
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)
|
|
report_check(
|
|
"created post fields",
|
|
as_int(post_id) > 0
|
|
and body.get("title") == "Test Post"
|
|
and body.get("content") == "Hello world"
|
|
and body.get("tags") == ["test"]
|
|
and body.get("pinned") is False
|
|
and as_int(body.get("likeCount")) == 0
|
|
and as_int(body.get("commentCount")) == 0
|
|
and body.get("liked") is False
|
|
and same_id((body.get("author") or {}).get("id"), user_id),
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/posts")
|
|
report("GET /posts", code, body)
|
|
report_check("created post appears in list", bool(find_item_by_id(pick_items(body), post_id)), body)
|
|
|
|
if post_id:
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/posts/{post_id}")
|
|
report(f"GET /posts/{post_id}", code, body)
|
|
report_check(
|
|
f"GET /posts/{post_id} returns created post",
|
|
same_id(body.get("id"), post_id) and body.get("title") == "Test Post",
|
|
body,
|
|
)
|
|
original_like_count = as_int(body.get("likeCount"))
|
|
|
|
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.get(f"{GATEWAY}/api/v1/posts/{post_id}")
|
|
report(f"GET /posts/{post_id} (after like)", code, body)
|
|
report_check(
|
|
"post like state is visible",
|
|
body.get("liked") is True
|
|
and as_int(body.get("likeCount")) >= original_like_count + 1,
|
|
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.get(f"{GATEWAY}/api/v1/posts/{post_id}")
|
|
report(f"GET /posts/{post_id} (after unlike)", code, body)
|
|
report_check(
|
|
"post unlike state is visible",
|
|
body.get("liked") is False
|
|
and as_int(body.get("likeCount")) == original_like_count,
|
|
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.get(f"{GATEWAY}/api/v1/posts/{post_id}")
|
|
report(f"GET /posts/{post_id} (after pin)", code, body)
|
|
report_check("post pin state is visible", body.get("pinned") is True, 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.get(f"{GATEWAY}/api/v1/posts/{post_id}")
|
|
report(f"GET /posts/{post_id} (after unpin)", code, body)
|
|
report_check("post unpin state is visible", body.get("pinned") is False, 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)
|
|
report_check(
|
|
"created comment fields",
|
|
as_int(comment_id) > 0
|
|
and body.get("content") == "Nice post!"
|
|
and same_id((body.get("author") or {}).get("id"), user_id)
|
|
and as_int(body.get("likeCount")) == 0
|
|
and body.get("liked") is False,
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/posts/{post_id}/comments")
|
|
report(f"GET /posts/{post_id}/comments", code, body)
|
|
report_check(
|
|
"created comment appears in list",
|
|
bool(find_item_by_id(pick_items(body), comment_id)),
|
|
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.get(f"{GATEWAY}/api/v1/posts/{post_id}/comments")
|
|
report(f"GET /posts/{post_id}/comments (after comment like)", code, body)
|
|
comment = find_item_by_id(pick_items(body), comment_id)
|
|
report_check(
|
|
"comment like state is visible",
|
|
bool(comment)
|
|
and comment.get("liked") is True
|
|
and as_int(comment.get("likeCount")) >= 1,
|
|
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/posts/{post_id}/comments")
|
|
report(f"GET /posts/{post_id}/comments (after comment unlike)", code, body)
|
|
comment = find_item_by_id(pick_items(body), comment_id)
|
|
report_check(
|
|
"comment unlike state is visible",
|
|
bool(comment)
|
|
and comment.get("liked") is False
|
|
and as_int(comment.get("likeCount")) == 0,
|
|
body,
|
|
)
|
|
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/users/{user_id}/posts")
|
|
report(f"GET /users/{user_id}/posts", code, body)
|
|
report_check("user posts contain created post", bool(find_item_by_id(pick_items(body), post_id)), 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",
|
|
isinstance(body.get("url"), str) and body.get("url", "").startswith("http"),
|
|
{"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)
|
|
report_check(
|
|
"email verification send returns request id",
|
|
code == 200 and bool(body.get("requestId")) and as_int(body.get("expireInSec")) > 0,
|
|
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)
|
|
report_check(
|
|
"forgot password send returns request id",
|
|
code == 200 and bool(body.get("requestId")) and as_int(body.get("expireInSec")) > 0,
|
|
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)
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/users/me")
|
|
report("GET /users/me (after logout)", code, body, expect_status=(401, 403))
|
|
|
|
|
|
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)
|
|
report_check(
|
|
"reset password allows login with new password",
|
|
code == 200 and bool(s_login.get_cookie("JToken")) and pick_user_id(body) > 0,
|
|
{"userId": pick_user_id(body), "hasJToken": bool(s_login.get_cookie("JToken"))},
|
|
)
|
|
|
|
|
|
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)
|
|
code, body, _ = s.get(f"{GATEWAY}/api/v1/services")
|
|
report("GET /services (after delete)", code, body)
|
|
report_check(
|
|
"deleted service is absent from service list",
|
|
not bool(find_item_by_id(pick_items(body), service_id)),
|
|
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, s_user, order_id, consumer_user_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()
|