242 lines
8.9 KiB
Python
242 lines
8.9 KiB
Python
#!/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()
|