juwan-backend/deploy/dev/test_all_apis.py

#!/usr/bin/env python3
"""Full API integration test for juwan-backend.

Runs against the local dev docker-compose environment.
Gateway: http://127.0.0.1:18080
Reads verification codes from local Redis only for dev registration flow.
"""

import json
import random
import string
import sys
import urllib.request
import urllib.error
import urllib.parse
import http.cookiejar
import os
import subprocess

GATEWAY = "http://127.0.0.1:18080"
ADMIN_USERNAME = os.getenv("ADMIN_USERNAME", "admin")
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin123")
REDIS_CONTAINER = os.getenv("REDIS_CONTAINER", "juwan-redis")

passed = 0
failed = 0
errors_list = []


def rand_str(n=8):
    return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))


def as_int(value, default=0):
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def pick_items(body):
    if isinstance(body.get("items"), list):
        return body["items"]
    if isinstance(body.get("list"), list):
        return body["list"]
    return []


def pick_user_id(body):
    if not isinstance(body, dict):
        return 0
    if isinstance(body.get("user"), dict):
        return as_int(body["user"].get("id"))
    return as_int(body.get("id"))


def read_vcode_from_redis(request_id, scene, account):
    if not request_id:
        return ""

    result = subprocess.run(
        [
            "docker",
            "exec",
            REDIS_CONTAINER,
            "redis-cli",
            "GET",
            f"vcode:{request_id}:{scene}:{account}",
        ],
        capture_output=True,
        text=True,
        timeout=5,
    )
    return result.stdout.strip()


def build_multipart_form(fields, files):
    boundary = f"----juwan{rand_str(12)}"
    chunks = []

    for name, value in fields.items():
        chunks.extend(
            [
                f"--{boundary}\r\n".encode(),
                f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(),
                str(value).encode(),
                b"\r\n",
            ]
        )

    for name, (filename, content, content_type) in files.items():
        if isinstance(content, str):
            content = content.encode()
        chunks.extend(
            [
                f"--{boundary}\r\n".encode(),
                (
                    f'Content-Disposition: form-data; name="{name}"; '
                    f'filename="{filename}"\r\n'
                ).encode(),
                f"Content-Type: {content_type}\r\n\r\n".encode(),
                content,
                b"\r\n",
            ]
        )

    chunks.append(f"--{boundary}--\r\n".encode())
    body = b"".join(chunks)
    return f"multipart/form-data; boundary={boundary}", body


class Session:
    """Minimal cookie-aware HTTP session using stdlib only."""

    def __init__(self):
        self.cookie_jar = http.cookiejar.CookieJar(
            policy=http.cookiejar.DefaultCookiePolicy(
                secure_protocols=("https", "http")
            )
        )
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cookie_jar)
        )

    def get_cookie(self, name):
        for c in self.cookie_jar:
            if c.name == name:
                return c.value
        return None

    def request(
        self,
        method,
        url,
        json_body=None,
        headers=None,
        form_data=None,
        raw_body=None,
    ):
        hdrs = headers or {}
        body = None
        if json_body is not None:
            body = json.dumps(json_body).encode()
            hdrs.setdefault("Content-Type", "application/json")
        elif form_data is not None:
            body = urllib.parse.urlencode(form_data).encode()
            hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded")
        elif raw_body is not None:
            body = raw_body

        req = urllib.request.Request(url, data=body, headers=hdrs, method=method)
        try:
            resp = self.opener.open(req, timeout=15)
            data = resp.read().decode()
            try:
                return resp.status, json.loads(data) if data else {}, dict(resp.headers)
            except json.JSONDecodeError:
                return resp.status, {"_raw": data}, dict(resp.headers)
        except urllib.error.HTTPError as e:
            data = e.read().decode() if e.fp else ""
            try:
                return e.code, json.loads(data) if data else {}, dict(e.headers)
            except json.JSONDecodeError:
                return e.code, {"_raw": data}, dict(e.headers)
        except Exception as e:
            return 0, {"_error": str(e)}, {}

    def get(self, url, **kw):
        return self.request("GET", url, **kw)

    def post(self, url, **kw):
        return self.request("POST", url, **kw)

    def post_multipart(self, url, fields, files, headers=None):
        hdrs = dict(headers or {})
        content_type, body = build_multipart_form(fields, files)
        hdrs["Content-Type"] = content_type
        return self.post(url, headers=hdrs, raw_body=body)

    def put(self, url, **kw):
        return self.request("PUT", url, **kw)

    def delete(self, url, **kw):
        return self.request("DELETE", url, **kw)

    def csrf_headers(self):
        token = self.get_cookie("__Host-XSRF-TOKEN")
        return {"xsrf-token": token} if token else {}


def report(name, status_code, body, expect_status=200):
    global passed, failed
    if isinstance(expect_status, (list, tuple, set)):
        ok = status_code in expect_status
    else:
        ok = status_code == expect_status
    mark = "PASS" if ok else "FAIL"
    if not ok:
        failed += 1
        errors_list.append((name, status_code, body))
    else:
        passed += 1
    body_preview = json.dumps(body, ensure_ascii=False)
    if len(body_preview) > 200:
        body_preview = body_preview[:200] + "..."
    print(f"  [{mark}] {name}: HTTP {status_code} {body_preview}")
    return ok


def report_check(name, ok, body=None):
    global passed, failed
    mark = "PASS" if ok else "FAIL"
    if not ok:
        failed += 1
        errors_list.append((name, "CHECK", body))
    else:
        passed += 1
    body_preview = json.dumps(body, ensure_ascii=False)
    if len(body_preview) > 200:
        body_preview = body_preview[:200] + "..."
    print(f"  [{mark}] {name}: {body_preview}")
    return ok


def skip(name, reason):
    print(f"  [SKIP] {name}: {reason}")


# ============================================================
# Phase 0: Health check & CSRF
# ============================================================
def phase0_health(s: Session):
    print("\n=== Phase 0: Health & CSRF ===")
    code, body, _ = s.get(f"{GATEWAY}/healthz")
    report("GET /healthz", code, body)
    xsrf = s.get_cookie("__Host-XSRF-TOKEN")
    xsrf_guard = s.get_cookie("__Host-XSRF-GUARD")
    print(f"  XSRF-TOKEN: {xsrf}")
    print(f"  XSRF-GUARD: {xsrf_guard}")
    if not xsrf:
        print("  [WARN] No XSRF-TOKEN cookie received, POST requests will fail")


# ============================================================
# Phase 1: Registration
# ============================================================
def phase1_register(s: Session, username, email, password, label="user"):
    print(f"\n=== Phase 1: Register ({label}) ===")

    # Step 1: send verification code via gateway
    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/email/verification-code/send",
        json_body={"email": email, "scene": "register"},
        headers=s.csrf_headers(),
    )
    report(f"POST /email/verification-code/send ({label})", code, body)
    request_id = body.get("requestId", "")
    if not request_id:
        print("  [ERROR] No requestId returned, cannot register")
        return None

    # Step 2: read the dev verification code from Redis
    print("  [BYPASS] Reading vcode from Redis...")
    vcode = read_vcode_from_redis(request_id, "register", email)
    if not vcode:
        print("  [WARN] No verification code found in Redis")
    else:
        print(f"  Vcode from Redis: {vcode}")

    # Step 3: register via gateway with X-Request-Id
    csrf = s.csrf_headers()
    csrf["X-Request-Id"] = request_id
    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/auth/register",
        json_body={
            "username": username,
            "email": email,
            "password": password,
            "vcode": vcode,
        },
        headers=csrf,
    )
    report(f"POST /auth/register ({label})", code, body)
    return body


def phase1_login(s: Session, username, password, label="user"):
    print(f"\n=== Phase 1b: Login ({label}) ===")
    csrf = s.csrf_headers()
    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/auth/login",
        json_body={"username": username, "password": password},
        headers=csrf,
    )
    report(f"POST /auth/login ({label})", code, body)
    jtoken = s.get_cookie("JToken")
    print(f"  JToken: {jtoken[:30]}..." if jtoken else "  JToken: None")
    return body


# ============================================================
# Phase 2: User endpoints (authenticated)
# ============================================================
def phase2_user(s: Session, user_id):
    print("\n=== Phase 2: User Endpoints ===")
    csrf = s.csrf_headers()

    code, body, _ = s.get(f"{GATEWAY}/api/v1/users/me")
    report("GET /users/me", code, body)

    code, body, _ = s.put(
        f"{GATEWAY}/api/v1/users/me",
        json_body={"nickname": "TestNick", "bio": "testbio"},
        headers=csrf,
    )
    report("PUT /users/me", code, body)

    code, body, _ = s.put(
        f"{GATEWAY}/api/v1/users/me/preferences/notifications",
        json_body={"order": True, "community": True, "system": True},
        headers=csrf,
    )
    report("PUT /users/me/preferences/notifications", code, body)

    code, body, _ = s.put(
        f"{GATEWAY}/api/v1/users/me/preferences/theme",
        json_body={"theme": "dark"},
        headers=csrf,
    )
    report("PUT /users/me/preferences/theme", code, body)

    code, body, _ = s.get(f"{GATEWAY}/api/v1/users/{user_id}")
    report(f"GET /users/{user_id} (public)", code, body)


# ============================================================
# Phase 3: Built-in Admin + Verification flow
# ============================================================
def phase3_admin_and_verification(
    s_admin: Session,
    s_user: Session,
    s_reject: Session,
    admin_user,
    admin_pass,
):
    print("\n=== Phase 3: Built-in Admin & Verification ===")

    # Login admin via gateway
    s_admin.get(f"{GATEWAY}/healthz")  # get CSRF
    csrf = s_admin.csrf_headers()
    code, body, _ = s_admin.post(
        f"{GATEWAY}/api/v1/auth/login",
        json_body={"username": admin_user, "password": admin_pass},
        headers=csrf,
    )
    report("POST /auth/login (admin)", code, body)

    code, body, _ = s_admin.get(f"{GATEWAY}/api/v1/users/me")
    report("GET /users/me (admin)", code, body)
    admin_id = pick_user_id(body)

    # User applies for player verification
    csrf_user = s_user.csrf_headers()
    code, body, _ = s_user.post(
        f"{GATEWAY}/api/v1/users/me/verification",
        json_body={
            "role": "player",
            "materials": {
                "idCardFront": "http://example.com/front.jpg",
                "idCardBack": "http://example.com/back.jpg",
                "gameScreenshots": ["http://example.com/ss1.jpg"],
                "voiceDemo": "http://example.com/voice.mp3",
            },
        },
        headers=csrf_user,
    )
    report("POST /users/me/verification (apply player)", code, body)

    # User applies for owner verification
    code, body, _ = s_user.post(
        f"{GATEWAY}/api/v1/users/me/verification",
        json_body={
            "role": "owner",
            "materials": {
                "idCardFront": "http://example.com/front.jpg",
                "idCardBack": "http://example.com/back.jpg",
                "gameScreenshots": [],
                "voiceDemo": "",
            },
        },
        headers=csrf_user,
    )
    report("POST /users/me/verification (apply owner)", code, body)

    # Another user applies so admin reject flow is covered too
    csrf_reject = s_reject.csrf_headers()
    code, body, _ = s_reject.post(
        f"{GATEWAY}/api/v1/users/me/verification",
        json_body={
            "role": "owner",
            "materials": {
                "idCardFront": "http://example.com/reject-front.jpg",
                "idCardBack": "http://example.com/reject-back.jpg",
                "gameScreenshots": [],
                "voiceDemo": "",
            },
        },
        headers=csrf_reject,
    )
    report("POST /users/me/verification (apply owner, reject user)", code, body)

    # Get my verifications
    code, body, _ = s_user.get(f"{GATEWAY}/api/v1/users/me/verification")
    report("GET /users/me/verification", code, body)
    user_verification_ids = {}
    for v in pick_items(body):
        user_verification_ids[v.get("role")] = as_int(v.get("id"))

    code, body, _ = s_reject.get(f"{GATEWAY}/api/v1/users/me/verification")
    report("GET /users/me/verification (reject user)", code, body)
    reject_verification_ids = {}
    for v in pick_items(body):
        reject_verification_ids[v.get("role")] = as_int(v.get("id"))

    # Admin: list and approve via gateway
    code, body, _ = s_admin.get(f"{GATEWAY}/api/v1/admin/verifications")
    report("GET /admin/verifications", code, body)

    # Admin: approve only the current test user's records
    for role in ("player", "owner"):
        vid = user_verification_ids.get(role)
        if not vid:
            continue
        code, body, _ = s_admin.post(
            f"{GATEWAY}/api/v1/admin/verifications/{vid}/approve",
            json_body={},
            headers=s_admin.csrf_headers(),
        )
        report(
            f"POST /admin/verifications/{vid}/approve ({role})",
            code,
            body,
        )

    reject_vid = reject_verification_ids.get("owner")
    if reject_vid:
        code, body, _ = s_admin.post(
            f"{GATEWAY}/api/v1/admin/verifications/{reject_vid}/reject",
            json_body={"reason": "test reject flow"},
            headers=s_admin.csrf_headers(),
        )
        report(
            f"POST /admin/verifications/{reject_vid}/reject (owner)",
            code,
            body,
        )

    code, body, _ = s_reject.get(f"{GATEWAY}/api/v1/users/me/verification")
    report("GET /users/me/verification (after reject)", code, body)

    # User: switch role to player
    code, body, _ = s_user.post(
        f"{GATEWAY}/api/v1/users/me/switch-role",
        json_body={"role": "player"},
        headers=csrf_user,
    )
    report("POST /users/me/switch-role (player)", code, body)

    return admin_id


def phase3b_secondary_player(s_admin: Session, s_player: Session):
    print("\n=== Phase 3b: Secondary Player Setup ===")

    csrf_player = s_player.csrf_headers()
    code, body, _ = s_player.post(
        f"{GATEWAY}/api/v1/users/me/verification",
        json_body={
            "role": "player",
            "materials": {
                "idCardFront": "http://example.com/player-front.jpg",
                "idCardBack": "http://example.com/player-back.jpg",
                "gameScreenshots": ["http://example.com/player-ss1.jpg"],
                "voiceDemo": "http://example.com/player-voice.mp3",
            },
        },
        headers=csrf_player,
    )
    report("POST /users/me/verification (apply player, invited user)", code, body)

    code, body, _ = s_player.get(f"{GATEWAY}/api/v1/users/me/verification")
    report("GET /users/me/verification (invited user)", code, body)
    verification_id = 0
    for item in pick_items(body):
        if item.get("role") == "player":
            verification_id = as_int(item.get("id"))
            break
    report_check(
        "lookup player verification id (invited user)",
        bool(verification_id),
        {"id": verification_id},
    )

    if verification_id:
        code, body, _ = s_admin.post(
            f"{GATEWAY}/api/v1/admin/verifications/{verification_id}/approve",
            json_body={},
            headers=s_admin.csrf_headers(),
        )
        report(
            f"POST /admin/verifications/{verification_id}/approve (invited user player)",
            code,
            body,
        )

    code, body, _ = s_player.post(
        f"{GATEWAY}/api/v1/users/me/switch-role",
        json_body={"role": "player"},
        headers=s_player.csrf_headers(),
    )
    report("POST /users/me/switch-role (invited user player)", code, body)

    code, body, _ = s_player.post(
        f"{GATEWAY}/api/v1/players/me",
        json_body={},
        headers=s_player.csrf_headers(),
    )
    report("POST /players/me (invited user init)", code, body)
    return as_int(body.get("id"))


def phase4_follow(s: Session, target_user_id):
    print("\n=== Phase 4: Follow/Unfollow ===")
    csrf = s.csrf_headers()

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/users/{target_user_id}/follow",
        json_body={},
        headers=csrf,
    )
    report(f"POST /users/{target_user_id}/follow", code, body)

    code, body, _ = s.delete(
        f"{GATEWAY}/api/v1/users/{target_user_id}/follow",
        headers=csrf,
    )
    report(f"DELETE /users/{target_user_id}/follow", code, body)


def phase5_games(s: Session, s_admin: Session):
    print("\n=== Phase 5: Games ===")

    code, body, _ = s.get(f"{GATEWAY}/api/v1/games")
    report("GET /games", code, body)

    csrf_admin = s_admin.csrf_headers()
    code, body, _ = s_admin.post(
        f"{GATEWAY}/api/v1/games",
        json_body={
            "name": f"TestGame_{rand_str(4)}",
            "icon": "icon.png",
            "category": "MOBA",
        },
        headers=csrf_admin,
    )
    report("POST /games (create)", code, body)
    game_id = body.get("id", 0)

    code, body2, _ = s.get(f"{GATEWAY}/api/v1/games")
    report("GET /games (after create)", code, body2)
    items = pick_items(body2)
    if not game_id and items:
        game_id = as_int(items[-1].get("id"))

    if game_id:
        code, body, _ = s.get(f"{GATEWAY}/api/v1/games/{game_id}")
        report(f"GET /games/{game_id}", code, body)

    return game_id


def phase6_player(s: Session, game_id):
    print("\n=== Phase 6: Player ===")
    csrf = s.csrf_headers()

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/players/me",
        json_body={},
        headers=csrf,
    )
    report("POST /players/me (init)", code, body)
    player_id = body.get("id", 0)

    code, body, _ = s.put(
        f"{GATEWAY}/api/v1/players/me/status",
        json_body={"status": "online"},
        headers=csrf,
    )
    report("PUT /players/me/status", code, body)

    code, body, _ = s.get(f"{GATEWAY}/api/v1/players")
    report("GET /players", code, body)

    if player_id:
        code, body, _ = s.get(f"{GATEWAY}/api/v1/players/{player_id}")
        report(f"GET /players/{player_id}", code, body)

    svc_body = None
    if game_id:
        code, body, _ = s.post(
            f"{GATEWAY}/api/v1/services",
            json_body={
                "gameId": game_id,
                "title": "Boosting Service",
                "description": "Rank boost",
                "price": 50.0,
                "unit": "game",
            },
            headers=csrf,
        )
        report("POST /services (create)", code, body)
        svc_body = body

    code, body, _ = s.get(f"{GATEWAY}/api/v1/services")
    report("GET /services", code, body)

    service_id = svc_body.get("id", 0) if svc_body else 0
    if service_id:
        code, body, _ = s.get(f"{GATEWAY}/api/v1/services/{service_id}")
        report(f"GET /services/{service_id}", code, body)

        code, body, _ = s.put(
            f"{GATEWAY}/api/v1/services/{service_id}",
            json_body={
                "title": "Updated Service",
                "price": 60.0,
                "availability": ["weekday"],
            },
            headers=csrf,
        )
        report(f"PUT /services/{service_id}", code, body)

    if player_id:
        code, body, _ = s.get(f"{GATEWAY}/api/v1/players/{player_id}/services")
        report(f"GET /players/{player_id}/services", code, body)

    return player_id, service_id


def phase7_shop(
    s_owner: Session, s_invited_player: Session, owner_user_id, invited_player_id
):
    print("\n=== Phase 7: Shop ===")

    s_owner.post(
        f"{GATEWAY}/api/v1/users/me/switch-role",
        json_body={"role": "owner"},
        headers=s_owner.csrf_headers(),
    )

    csrf = s_owner.csrf_headers()
    code, body, _ = s_owner.post(
        f"{GATEWAY}/api/v1/shops",
        json_body={
            "name": f"TestShop_{rand_str(4)}",
            "description": "A test shop",
            "commissionType": "percentage",
            "commissionValue": "10",
        },
        headers=csrf,
    )
    report("POST /shops (create)", code, body)
    shop_id_str = body.get("id", "0")
    try:
        shop_id = int(shop_id_str)
    except (ValueError, TypeError):
        shop_id = 0

    code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops")
    report("GET /shops", code, body)

    if shop_id:
        code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
        report(f"GET /shops/{shop_id}", code, body)

        code, body, _ = s_owner.put(
            f"{GATEWAY}/api/v1/shops/{shop_id}",
            json_body={
                "name": f"UpdatedShop_{rand_str(4)}",
                "description": "An updated test shop",
                "commissionType": "percentage",
                "commissionValue": "12",
                "allowMultiShop": True,
                "allowIndependentOrders": False,
                "dispatchMode": "manual",
            },
            headers=csrf,
        )
        report(f"PUT /shops/{shop_id}", code, body)

        code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/users/{owner_user_id}/shop")
        report(f"GET /users/{owner_user_id}/shop", code, body)

        announcement = f"Grand opening {rand_str(4)}!"
        code, body, _ = s_owner.post(
            f"{GATEWAY}/api/v1/shops/{shop_id}/announcements",
            json_body={"content": announcement},
            headers=csrf,
        )
        report(f"POST /shops/{shop_id}/announcements", code, body)

        code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/{shop_id}")
        report(f"GET /shops/{shop_id} (after announcement)", code, body)
        announcement_index = -1
        announcements = body.get("announcements") if isinstance(body, dict) else None
        if isinstance(announcements, list):
            for idx in range(len(announcements) - 1, -1, -1):
                if announcement in str(announcements[idx]):
                    announcement_index = idx
                    break
        report_check(
            f"locate announcement index ({shop_id})",
            announcement_index >= 0,
            {"index": announcement_index},
        )
        if announcement_index >= 0:
            code, body, _ = s_owner.delete(
                f"{GATEWAY}/api/v1/shops/{shop_id}/announcements/{announcement_index}",
                headers=csrf,
            )
            report(
                f"DELETE /shops/{shop_id}/announcements/{announcement_index}",
                code,
                body,
            )

        code, body, _ = s_owner.put(
            f"{GATEWAY}/api/v1/shops/{shop_id}/template",
            json_body={"sections": json.dumps({"layout": "grid", "theme": "dark"})},
            headers=csrf,
        )
        report(f"PUT /shops/{shop_id}/template", code, body)

        code, body, _ = s_owner.get(
            f"{GATEWAY}/api/v1/shops/{shop_id}/income-stats",
        )
        report(f"GET /shops/{shop_id}/income-stats", code, body)

        if invited_player_id:
            code, body, _ = s_owner.post(
                f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
                json_body={"playerId": invited_player_id},
                headers=csrf,
            )
            report(f"POST /shops/{shop_id}/invitations", code, body)

            code, body, _ = s_owner.get(
                f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
            )
            report(f"GET /shops/{shop_id}/invitations", code, body)

            invitation_id = 0
            for item in pick_items(body):
                if as_int(item.get("playerId")) == invited_player_id:
                    invitation_id = as_int(item.get("id"))
                    break
            report_check(
                "locate invitation id",
                bool(invitation_id),
                {"id": invitation_id},
            )

            s_invited_player.post(
                f"{GATEWAY}/api/v1/users/me/switch-role",
                json_body={"role": "player"},
                headers=s_invited_player.csrf_headers(),
            )
            code, body, _ = s_invited_player.get(
                f"{GATEWAY}/api/v1/shops/invitations/mine",
            )
            report("GET /shops/invitations/mine", code, body)

            if invitation_id:
                inv_csrf = s_invited_player.csrf_headers()
                code, body, _ = s_invited_player.post(
                    f"{GATEWAY}/api/v1/shops/invitations/{invitation_id}/accept",
                    json_body={},
                    headers=inv_csrf,
                )
                report(
                    f"POST /shops/invitations/{invitation_id}/accept",
                    code,
                    body,
                )

                code, body, _ = s_owner.delete(
                    f"{GATEWAY}/api/v1/shops/{shop_id}/players/{invited_player_id}",
                    headers=csrf,
                )
                report(
                    f"DELETE /shops/{shop_id}/players/{invited_player_id}",
                    code,
                    body,
                )

                # Reject invitation flow: re-invite the removed player, then reject
                code, body, _ = s_owner.post(
                    f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
                    json_body={"playerId": invited_player_id},
                    headers=csrf,
                )
                report(f"POST /shops/{shop_id}/invitations (re-invite)", code, body)

                code, body, _ = s_owner.get(
                    f"{GATEWAY}/api/v1/shops/{shop_id}/invitations",
                )
                reinvite_id = 0
                for item in pick_items(body):
                    if (
                        as_int(item.get("playerId")) == invited_player_id
                        and item.get("status") == "pending"
                    ):
                        reinvite_id = as_int(item.get("id"))
                        break
                if reinvite_id:
                    code, body, _ = s_invited_player.delete(
                        f"{GATEWAY}/api/v1/shops/invitations/{reinvite_id}",
                        headers=s_invited_player.csrf_headers(),
                    )
                    report(
                        f"DELETE /shops/invitations/{reinvite_id} (reject)",
                        code,
                        body,
                    )

    code, body, _ = s_owner.get(f"{GATEWAY}/api/v1/shops/mine")
    report("GET /shops/mine", code, body)

    return shop_id


def phase8_order(s_consumer: Session, s_actor: Session, player_id, service_id, shop_id):
    print("\n=== Phase 8: Orders ===")

    s_consumer.post(
        f"{GATEWAY}/api/v1/users/me/switch-role",
        json_body={"role": "consumer"},
        headers=s_consumer.csrf_headers(),
    )

    csrf = s_consumer.csrf_headers()
    code, body, _ = s_consumer.post(
        f"{GATEWAY}/api/v1/orders",
        json_body={
            "playerId": player_id,
            "serviceId": service_id,
            "shopId": shop_id,
            "quantity": 1,
            "note": "test order",
        },
        headers=csrf,
    )
    report("POST /orders (create)", code, body)
    order_obj = body.get("order", {})
    order_id = order_obj.get("id", 0) if isinstance(order_obj, dict) else 0

    code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders?role=consumer")
    report("GET /orders?role=consumer", code, body)

    code, body, _ = s_actor.get(f"{GATEWAY}/api/v1/orders?role=player")
    report("GET /orders?role=player", code, body)

    code, body, _ = s_actor.get(f"{GATEWAY}/api/v1/orders?role=owner")
    report("GET /orders?role=owner", code, body)

    if order_id:
        code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders/{order_id}")
        report(f"GET /orders/{order_id}", code, body)

        code, body, _ = s_consumer.post(
            f"{GATEWAY}/api/v1/orders/{order_id}/pay",
            json_body={},
            headers=csrf,
        )
        report(f"POST /orders/{order_id}/pay", code, body)

        code, body, _ = s_actor.post(
            f"{GATEWAY}/api/v1/orders/{order_id}/accept",
            json_body={},
            headers=s_actor.csrf_headers(),
        )
        report(f"POST /orders/{order_id}/accept", code, body)

        code, body, _ = s_actor.post(
            f"{GATEWAY}/api/v1/orders/{order_id}/request-close",
            json_body={},
            headers=s_actor.csrf_headers(),
        )
        report(f"POST /orders/{order_id}/request-close", code, body)

        code, body, _ = s_consumer.post(
            f"{GATEWAY}/api/v1/orders/{order_id}/confirm-close",
            json_body={},
            headers=csrf,
        )
        report(f"POST /orders/{order_id}/confirm-close", code, body)

    code, body, _ = s_consumer.post(
        f"{GATEWAY}/api/v1/orders",
        json_body={
            "playerId": player_id,
            "serviceId": service_id,
            "quantity": 1,
        },
        headers=csrf,
    )
    order2 = body.get("order", {})
    order2_id = order2.get("id", 0) if isinstance(order2, dict) else 0

    if order_id:
        code, body, _ = s_consumer.post(
            f"{GATEWAY}/api/v1/orders/{order_id}/reorder",
            json_body={},
            headers=csrf,
        )
        report(f"POST /orders/{order_id}/reorder", code, body)

    if order2_id:
        code, body, _ = s_consumer.post(
            f"{GATEWAY}/api/v1/orders/{order2_id}/pay",
            json_body={},
            headers=csrf,
        )
        report(f"POST /orders/{order2_id}/pay (before cancel)", code, body)

        code, body, _ = s_consumer.post(
            f"{GATEWAY}/api/v1/orders/{order2_id}/cancel",
            json_body={},
            headers=csrf,
        )
        report(f"POST /orders/{order2_id}/cancel", code, body)

    code, body, _ = s_consumer.post(
        f"{GATEWAY}/api/v1/orders/paid",
        json_body={
            "playerId": player_id,
            "serviceId": service_id,
            "quantity": 1,
        },
        headers=csrf,
    )
    report("POST /orders/paid (create+pay)", code, body)

    return order_id


def phase8b_review(s_consumer: Session, order_id, player_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:
        report_check(
            f"GET /orders/{order_id}/reviews shape",
            isinstance(pick_items(body), list) and isinstance(body.get("meta"), dict),
            body,
        )

    code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/reviews?limit=20")
    report("GET /reviews?limit=20", code, body)

    if player_id:
        code, body, _ = s_consumer.get(
            f"{GATEWAY}/api/v1/users/{player_id}/reviews?limit=20",
        )
        report(f"GET /users/{player_id}/reviews?limit=20", code, body)


def phase8c_dispute(s_consumer: Session, s_actor: Session, player_id, service_id, shop_id):
    print("\n=== Phase 8c: Disputes ===")
    if not player_id or not service_id:
        skip("Dispute flow", "Missing player or service id")
        return 0

    csrf = s_consumer.csrf_headers()
    payload = {
        "playerId": player_id,
        "serviceId": service_id,
        "quantity": 1,
        "note": "test dispute order",
    }
    if shop_id:
        payload["shopId"] = shop_id

    code, body, _ = s_consumer.post(
        f"{GATEWAY}/api/v1/orders",
        json_body=payload,
        headers=csrf,
    )
    report("POST /orders (create for dispute)", code, body)
    order_obj = body.get("order", {}) if isinstance(body, dict) else {}
    order_id = order_obj.get("id", 0) if isinstance(order_obj, dict) else 0
    if not order_id:
        skip("Dispute flow", "No order id returned")
        return 0

    code, body, _ = s_consumer.post(
        f"{GATEWAY}/api/v1/orders/{order_id}/pay",
        json_body={},
        headers=csrf,
    )
    report(f"POST /orders/{order_id}/pay (dispute flow)", code, body)

    code, body, _ = s_actor.post(
        f"{GATEWAY}/api/v1/orders/{order_id}/accept",
        json_body={},
        headers=s_actor.csrf_headers(),
    )
    report(f"POST /orders/{order_id}/accept (dispute flow)", code, body)

    code, body, _ = s_consumer.post(
        f"{GATEWAY}/api/v1/orders/{order_id}/dispute",
        json_body={
            "reason": "test dispute reason",
            "evidence": ["http://example.com/evidence.jpg"],
        },
        headers=csrf,
    )
    report(f"POST /orders/{order_id}/dispute", code, body)

    code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/orders/{order_id}/dispute")
    report(f"GET /orders/{order_id}/dispute", code, body)
    dispute_id = as_int(body.get("id")) if isinstance(body, dict) else 0

    code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/disputes")
    report("GET /disputes", code, body)

    code, body, _ = s_consumer.get(f"{GATEWAY}/api/v1/disputes?status=open")
    report("GET /disputes?status=open", code, body)

    if dispute_id:
        code, body, _ = s_actor.post(
            f"{GATEWAY}/api/v1/disputes/{dispute_id}/response",
            json_body={
                "reason": "test respondent guard",
                "evidence": ["http://example.com/response.jpg"],
            },
            headers=s_actor.csrf_headers(),
        )
        report(
            f"POST /disputes/{dispute_id}/response (expect participant check)",
            code,
            body,
            expect_status=(400, 403, 500),
        )

        code, body, _ = s_consumer.post(
            f"{GATEWAY}/api/v1/disputes/{dispute_id}/appeal",
            json_body={"reason": "test appeal guard"},
            headers=csrf,
        )
        report(
            f"POST /disputes/{dispute_id}/appeal (expect status check)",
            code,
            body,
            expect_status=(400, 403, 500),
        )

    return dispute_id


def phase8d_notifications(s: Session):
    print("\n=== Phase 8d: Notifications ===")
    code, body, _ = s.get(f"{GATEWAY}/api/v1/notifications")
    report("GET /notifications", code, body)

    items = pick_items(body) if code == 200 and isinstance(body, dict) else []
    if items:
        notification_id = as_int(items[0].get("id"))
        if notification_id:
            code, body, _ = s.put(
                f"{GATEWAY}/api/v1/notifications/{notification_id}/read",
                json_body={},
                headers=s.csrf_headers(),
            )
            report(f"PUT /notifications/{notification_id}/read", code, body)
    else:
        skip("PUT /notifications/:id/read", "No notification item returned")

    code, body, _ = s.put(
        f"{GATEWAY}/api/v1/notifications/read-all",
        json_body={},
        headers=s.csrf_headers(),
    )
    report("PUT /notifications/read-all", code, body)


def phase8e_search_and_favorites(s: Session, user_id, player_id, shop_id):
    print("\n=== Phase 8e: Search & Favorites ===")

    code, body, _ = s.get(f"{GATEWAY}/api/v1/search?q=LOL&limit=10")
    report("GET /search?q=LOL&limit=10", code, body)
    if code == 200:
        report_check(
            "GET /search response shape",
            isinstance(pick_items(body), list) and isinstance(body.get("meta"), dict),
            body,
        )

    code, body, _ = s.get(f"{GATEWAY}/api/v1/recommendations/home?limit=10")
    report("GET /recommendations/home?limit=10", code, body)
    if code == 200:
        report_check(
            "GET /recommendations/home response shape",
            isinstance(pick_items(body), list) and isinstance(body.get("meta"), dict),
            body,
        )

    code, body, _ = s.get(f"{GATEWAY}/api/v1/favorites")
    report("GET /favorites", code, body)

    if not player_id or not user_id:
        skip("Favorites mutation flow", "Missing user or player id")
        return

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/favorites",
        json_body={"targetType": "player", "targetId": str(player_id)},
        headers=s.csrf_headers(),
    )
    report("POST /favorites (player)", code, body)

    code, body, _ = s.get(
        f"{GATEWAY}/api/v1/users/{user_id}/favorites/check"
        f"?targetType=player&targetId={player_id}",
    )
    report(f"GET /users/{user_id}/favorites/check (player)", code, body)
    if code == 200:
        report_check("favorite check after add", body.get("favorited") is True, body)

    if shop_id:
        code, body, _ = s.post(
            f"{GATEWAY}/api/v1/favorites",
            json_body={"targetType": "shop", "targetId": str(shop_id)},
            headers=s.csrf_headers(),
        )
        report("POST /favorites (shop)", code, body)

    code, body, _ = s.get(f"{GATEWAY}/api/v1/favorites")
    report("GET /favorites (after add)", code, body)
    favorite_id = 0
    for item in pick_items(body):
        if (
            item.get("targetType") == "player"
            and str(item.get("targetId")) == str(player_id)
        ):
            favorite_id = as_int(item.get("id"))
            break
    report_check("locate favorite id", bool(favorite_id), {"id": favorite_id})

    if favorite_id:
        code, body, _ = s.delete(
            f"{GATEWAY}/api/v1/favorites/{favorite_id}",
            headers=s.csrf_headers(),
        )
        report(f"DELETE /favorites/{favorite_id}", code, body)

        code, body, _ = s.get(
            f"{GATEWAY}/api/v1/users/{user_id}/favorites/check"
            f"?targetType=player&targetId={player_id}",
        )
        report(f"GET /users/{user_id}/favorites/check (after delete)", code, body)
        if code == 200:
            report_check("favorite check after delete", body.get("favorited") is False, body)


def phase9_wallet(s: Session):
    print("\n=== Phase 9: Wallet ===")
    csrf = s.csrf_headers()

    code, body, _ = s.get(f"{GATEWAY}/api/v1/wallet/balance")
    report("GET /wallet/balance", code, body)

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/wallet/topup",
        json_body={"amount": "100.00", "method": "alipay"},
        headers=csrf,
    )
    report("POST /wallet/topup", code, body)

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/wallet/withdraw",
        json_body={"amount": "10.00", "method": "alipay"},
        headers=csrf,
    )
    report("POST /wallet/withdraw", code, body)

    code, body, _ = s.get(f"{GATEWAY}/api/v1/wallet/transactions")
    report("GET /wallet/transactions", code, body)


def phase10_community(s: Session, user_id):
    print("\n=== Phase 10: Community ===")
    csrf = s.csrf_headers()

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/posts",
        json_body={
            "title": "Test Post",
            "content": "Hello world",
            "images": [],
            "tags": ["test"],
        },
        headers=csrf,
    )
    report("POST /posts (create)", code, body)
    post_id = body.get("id", 0)

    code, body, _ = s.get(f"{GATEWAY}/api/v1/posts")
    report("GET /posts", code, body)

    if post_id:
        code, body, _ = s.get(f"{GATEWAY}/api/v1/posts/{post_id}")
        report(f"GET /posts/{post_id}", code, body)

        code, body, _ = s.post(
            f"{GATEWAY}/api/v1/posts/{post_id}/like",
            json_body={},
            headers=csrf,
        )
        report(f"POST /posts/{post_id}/like", code, body)

        code, body, _ = s.delete(
            f"{GATEWAY}/api/v1/posts/{post_id}/like",
            headers=csrf,
        )
        report(f"DELETE /posts/{post_id}/like", code, body)

        code, body, _ = s.post(
            f"{GATEWAY}/api/v1/posts/{post_id}/pin",
            json_body={},
            headers=csrf,
        )
        report(f"POST /posts/{post_id}/pin", code, body)

        code, body, _ = s.delete(
            f"{GATEWAY}/api/v1/posts/{post_id}/pin",
            headers=csrf,
        )
        report(f"DELETE /posts/{post_id}/pin", code, body)

        code, body, _ = s.post(
            f"{GATEWAY}/api/v1/posts/{post_id}/comments",
            json_body={"content": "Nice post!"},
            headers=csrf,
        )
        report(f"POST /posts/{post_id}/comments", code, body)
        comment_id = body.get("id", 0)

        code, body, _ = s.get(f"{GATEWAY}/api/v1/posts/{post_id}/comments")
        report(f"GET /posts/{post_id}/comments", code, body)

        if comment_id:
            code, body, _ = s.post(
                f"{GATEWAY}/api/v1/comments/{comment_id}/like",
                json_body={},
                headers=csrf,
            )
            report(f"POST /comments/{comment_id}/like", code, body)

            code, body, _ = s.delete(
                f"{GATEWAY}/api/v1/comments/{comment_id}/like",
                headers=csrf,
            )
            report(f"DELETE /comments/{comment_id}/like", code, body)

    code, body, _ = s.get(f"{GATEWAY}/api/v1/users/{user_id}/posts")
    report(f"GET /users/{user_id}/posts", code, body)

    return post_id


def phase11_objectstory(s: Session):
    print("\n=== Phase 11: Objectstory (File) ===")

    code, body, _ = s.post_multipart(
        f"{GATEWAY}/api/v1/upload",
        fields={"type": "post"},
        files={
            "file": (
                f"test-{rand_str(4)}.txt",
                f"juwan-objectstory-{rand_str(8)}",
                "text/plain",
            )
        },
        headers=s.csrf_headers(),
    )
    report("POST /upload", code, body)
    if code == 200:
        report_check(
            "POST /upload returned url",
            bool(body.get("url")),
            {"url": body.get("url", "")},
        )

    code, body, _ = s.get(f"{GATEWAY}/api/v1/files?key=nonexistent")
    report(
        "GET /files?key=nonexistent (expect error)",
        code,
        body,
        expect_status=(400, 500),
    )

    return


def phase12_email(s: Session):
    print("\n=== Phase 12: Email ===")
    csrf = s.csrf_headers()

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/email/verification-code/send",
        json_body={"email": f"test_{rand_str(4)}@example.com", "scene": "register"},
        headers=csrf,
    )
    report("POST /email/verification-code/send (gateway)", code, body)

    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/auth/forgot-password/send",
        json_body={"email": f"test_{rand_str(4)}@example.com"},
        headers=csrf,
    )
    report("POST /auth/forgot-password/send (gateway)", code, body)


def phase13_logout(s: Session):
    print("\n=== Phase 13: Logout ===")
    csrf = s.csrf_headers()
    code, body, _ = s.post(
        f"{GATEWAY}/api/v1/auth/logout",
        json_body={},
        headers=csrf,
    )
    report("POST /auth/logout", code, body)


def phase14_reset_password(username, email, new_password):
    print("\n=== Phase 14: Forgot/Reset Password ===")

    s_reset = Session()
    s_reset.get(f"{GATEWAY}/healthz")
    csrf = s_reset.csrf_headers()

    code, body, _ = s_reset.post(
        f"{GATEWAY}/api/v1/auth/forgot-password/send",
        json_body={"email": email},
        headers=csrf,
    )
    report("POST /auth/forgot-password/send", code, body)
    request_id = body.get("requestId", "")
    if not request_id:
        print("  [WARN] No requestId returned, skip reset-password")
        return

    vcode = read_vcode_from_redis(request_id, "reset_password", email)
    if not vcode:
        print("  [WARN] No reset password verification code found in Redis")
        return

    csrf["X-Request-Id"] = request_id
    code, body, _ = s_reset.post(
        f"{GATEWAY}/api/v1/auth/reset-password",
        json_body={
            "email": email,
            "vcode": vcode,
            "newPassword": new_password,
        },
        headers=csrf,
    )
    report("POST /auth/reset-password", code, body)
    if code != 200:
        print("  [SKIP] Reset password failed, skip login verification")
        return

    s_login = Session()
    s_login.get(f"{GATEWAY}/healthz")
    code, body, _ = s_login.post(
        f"{GATEWAY}/api/v1/auth/login",
        json_body={"username": username, "password": new_password},
        headers=s_login.csrf_headers(),
    )
    report("POST /auth/login (after reset)", code, body)


def phase15_player_service_delete(s: Session, service_id):
    print("\n=== Phase 15: Delete Service ===")
    if not service_id:
        print("  [SKIP] No service to delete")
        return
    csrf = s.csrf_headers()
    code, body, _ = s.delete(
        f"{GATEWAY}/api/v1/services/{service_id}",
        headers=csrf,
    )
    report(f"DELETE /services/{service_id}", code, body)


def main():
    global passed, failed

    suffix = rand_str(6)
    user1_name = f"testuser_{suffix}"
    user1_email = f"testuser_{suffix}@example.com"
    user1_pass = "TestPass123!"
    consumer_name = f"consumer_{suffix}"
    consumer_email = f"consumer_{suffix}@example.com"
    consumer_pass = "ConsumerPass123!"
    consumer_new_pass = "ConsumerPass456!"
    admin_name = ADMIN_USERNAME
    admin_pass = ADMIN_PASSWORD

    print(f"Test run: user={user1_name}, admin={admin_name}")

    s_user = Session()
    s_admin = Session()
    s_consumer = Session()

    phase0_health(s_user)
    phase1_register(s_user, user1_name, user1_email, user1_pass, label="primary")
    login_resp = phase1_login(s_user, user1_name, user1_pass, label="primary")

    user_id = pick_user_id(login_resp)
    print(f"  User ID: {user_id}")

    phase2_user(s_user, user_id)

    s_consumer.get(f"{GATEWAY}/healthz")
    phase1_register(
        s_consumer, consumer_name, consumer_email, consumer_pass, label="consumer"
    )
    consumer_login_resp = phase1_login(
        s_consumer,
        consumer_name,
        consumer_pass,
        label="consumer",
    )
    consumer_user_id = pick_user_id(consumer_login_resp)
    print(f"  Consumer User ID: {consumer_user_id}")

    admin_id = phase3_admin_and_verification(
        s_admin,
        s_user,
        s_consumer,
        admin_name,
        admin_pass,
    )
    invited_player_id = phase3b_secondary_player(s_admin, s_consumer)

    if admin_id and user_id:
        phase4_follow(s_user, admin_id)

    s_user.post(
        f"{GATEWAY}/api/v1/users/me/switch-role",
        json_body={"role": "player"},
        headers=s_user.csrf_headers(),
    )

    game_id = phase5_games(s_user, s_admin)
    player_id, service_id = phase6_player(s_user, game_id)
    shop_id = phase7_shop(s_user, s_consumer, user_id, invited_player_id)

    order_id = phase8_order(s_consumer, s_user, player_id, service_id, shop_id)
    phase8b_review(s_consumer, order_id, player_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()