#!/usr/bin/env python3 import argparse import asyncio import http.cookiejar import json import os import sys import time import urllib.error import urllib.request from pathlib import Path from urllib.parse import urlparse from django.db.models import Q PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT_ROOT / "backend")) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings") import django # noqa: E402 django.setup() from django.conf import settings # noqa: E402 from websockets.asyncio.client import connect # noqa: E402 from accounts.models import User # noqa: E402 from contest.models import ( # noqa: E402 Contest, ContestAttempt, RatingHistory, RealtimeMatch, ) class ApiClient: def __init__(self, base_url): self.base_url = base_url.rstrip("/") self.cookies = http.cookiejar.CookieJar() self.opener = urllib.request.build_opener( urllib.request.HTTPCookieProcessor(self.cookies) ) def request(self, method, path, payload=None, extra_headers=None): body = json.dumps(payload).encode() if payload is not None else None headers = {"Accept": "application/json"} if body is not None: headers["Content-Type"] = "application/json" csrf_token = self.cookie("csrftoken") if csrf_token: headers["X-CSRFToken"] = csrf_token headers.update(extra_headers or {}) request = urllib.request.Request( f"{self.base_url}{path}", data=body, headers=headers, method=method, ) try: with self.opener.open(request, timeout=10) as response: content = response.read() if not content: return None if "application/json" in response.headers.get("Content-Type", ""): return json.loads(content) return content.decode() except urllib.error.HTTPError as exc: content = exc.read().decode() raise RuntimeError(f"{method} {path} -> HTTP {exc.code}: {content}") from exc def get(self, path): return self.request("GET", path) def post(self, path, payload, extra_headers=None): return self.request("POST", path, payload, extra_headers) def cookie(self, name): return next((item.value for item in self.cookies if item.name == name), "") @property def cookie_header(self): return "; ".join(f"{item.name}={item.value}" for item in self.cookies) async def receive_until(websocket, event_type, reason=None): for _ in range(10): payload = json.loads(await asyncio.wait_for(websocket.recv(), timeout=5)) if payload.get("type") == event_type and ( reason is None or payload.get("reason") == reason ): return payload raise RuntimeError(f"未收到 WebSocket 事件: type={event_type}, reason={reason}") def websocket_url(base_url, path): parsed = urlparse(base_url) scheme = "wss" if parsed.scheme == "https" else "ws" return f"{scheme}://{parsed.netloc}{path}" async def run_flow(base_url, first_client, second_client, contest): created = first_client.post( f"/api/v1/contests/{contest.slug}/challenges/", {}, ) code = created["challenge_code"] print(f"[1/7] 玩家 A 创建联机码: {code}") async with connect( websocket_url(base_url, created["websocket_path"]), additional_headers={"Cookie": first_client.cookie_header}, open_timeout=10, ) as first_socket: await receive_until(first_socket, "connected") joined = second_client.post( "/api/v1/contests/challenges/join/", {"challenge_code": code}, ) await receive_until(first_socket, "state", "matched") print("[2/7] 玩家 B 加入，玩家 A 收到 matched 事件") first_state = first_client.get( f"/api/v1/contests/matches/{created['match_id']}/" ) if first_state["status"] != "active": raise RuntimeError("匹配后状态不是 active") async with connect( websocket_url(base_url, joined["websocket_path"]), additional_headers={"Cookie": second_client.cookie_header}, open_timeout=10, ) as second_socket: await receive_until(second_socket, "connected") await first_socket.send( json.dumps({"type": "progress", "answered_count": 1}) ) progress = await receive_until(second_socket, "progress") if progress["answered_count"] != 1: raise RuntimeError("答题进度同步失败") print("[3/7] 两个 WebSocket 已连接，答题进度同步成功") first_attempt = first_state["attempt"] second_attempt = joined["attempt"] first_answers = [ {"order": item["order"], "answer": "0"} for item in first_attempt["questions"] ] second_answers = [ {"order": item["order"], "answer": "0"} for item in second_attempt["questions"] ] first_result = first_client.post( f"/api/v1/contests/attempts/{first_attempt['attempt_id']}/submit/", {"answers": first_answers}, {"Idempotency-Key": f"local-first-{created['match_id']}"}, ) if first_result["status"] != "active": raise RuntimeError("首位玩家提交后比赛不应立即完成") if "correct_answer" in first_result["attempt"]["questions"][0]: raise RuntimeError("首位玩家提前看到了正确答案") await receive_until(second_socket, "state", "submitted") print("[4/7] 玩家 A 提交后答案锁定，未提前泄露正确答案") second_result = second_client.post( f"/api/v1/contests/attempts/{second_attempt['attempt_id']}/submit/", {"answers": second_answers}, {"Idempotency-Key": f"local-second-{created['match_id']}"}, ) if second_result["status"] != "completed": raise RuntimeError("双方提交后比赛没有完成") await receive_until(first_socket, "state", "completed") print("[5/7] 玩家 B 提交后双方收到 completed 事件") final_state = first_client.get( f"/api/v1/contests/matches/{created['match_id']}/" ) if "correct_answer" not in final_state["attempt"]["questions"][0]: raise RuntimeError("完成后没有公开题目解析") if final_state["result"] is None: raise RuntimeError("完成后没有胜负与 Rating 结果") print("[6/7] 最终比分、Rating 和题目解析均可读取") print("[7/7] 本地联机码约战端到端测试通过") def cleanup(users): user_ids = [user.id for user in users] matches = RealtimeMatch.objects.filter( Q(player_one_id__in=user_ids) | Q(player_two_id__in=user_ids) ) match_ids = list(matches.values_list("id", flat=True)) ContestAttempt.objects.filter( Q(user_id__in=user_ids) | Q(match_id__in=match_ids) ).delete() RatingHistory.objects.filter(match_id__in=match_ids).delete() matches.delete() User.objects.filter(id__in=user_ids).delete() def main(): parser = argparse.ArgumentParser(description="本地实时联机码约战端到端测试") parser.add_argument("--base-url", default="http://127.0.0.1:8000") args = parser.parse_args() hostname = urlparse(args.base_url).hostname if not settings.DEBUG or hostname not in {"127.0.0.1", "localhost"}: raise RuntimeError("该脚本只允许在 DEBUG=true 的本机地址运行") stamp = str(int(time.time() * 1000)) password = "LocalRealtime2026!" users = [ User.objects.create_user( username=f"local_ws_a_{stamp}", password=password, nickname="本地联机 A", ), User.objects.create_user( username=f"local_ws_b_{stamp}", password=password, nickname="本地联机 B", ), ] try: contest = Contest.objects.filter( kind=Contest.Kind.REALTIME, status=Contest.Status.PUBLISHED, track="standard", ).first() if contest is None: raise RuntimeError("缺少标准赛道实时比赛，请先执行 make seed") clients = [ApiClient(args.base_url), ApiClient(args.base_url)] for client, user in zip(clients, users): client.get("/") client.post( "/api/v1/accounts/login/", {"username": user.username, "password": password}, ) asyncio.run(run_flow(args.base_url, clients[0], clients[1], contest)) finally: cleanup(users) if __name__ == "__main__": main()