#!/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", 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, ) 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()