171 lines
5.9 KiB
Python
171 lines
5.9 KiB
Python
"""无头联机测试：2 个程序化真人客户端 + 10 个 AI 打一整局。
|
|
|
|
验证：REST 建房、WS 加入、座位分配、房主控制（补AI/开局）、
|
|
个性化快照、人类行动提交（发言/投票/夜间技能）、断线重连、胜负结算。
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import random
|
|
import sys
|
|
import urllib.request
|
|
|
|
import websockets
|
|
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
BASE = "http://127.0.0.1:8000"
|
|
random.seed(7)
|
|
|
|
|
|
class Client:
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
self.snap: dict | None = None
|
|
self.token: str | None = None
|
|
self.ws = None
|
|
self.task = None
|
|
|
|
async def connect(self, code: str, token: str | None = None) -> None:
|
|
url = f"ws://127.0.0.1:8000/ws/{code}?name={self.name}"
|
|
if token:
|
|
url += f"&token={token}"
|
|
self.ws = await websockets.connect(url)
|
|
self.task = asyncio.create_task(self._recv())
|
|
|
|
async def _recv(self) -> None:
|
|
try:
|
|
async for raw in self.ws:
|
|
data = json.loads(raw)
|
|
if data["type"] == "snapshot":
|
|
self.snap = data["data"]
|
|
if self.snap.get("token"):
|
|
self.token = self.snap["token"]
|
|
except Exception:
|
|
pass
|
|
|
|
async def send(self, obj: dict) -> None:
|
|
await self.ws.send(json.dumps(obj, ensure_ascii=False))
|
|
|
|
def seat(self):
|
|
me = (self.snap or {}).get("me")
|
|
return me["seat"] if me else None
|
|
|
|
def pending(self):
|
|
return (self.snap or {}).get("pending")
|
|
|
|
def phase(self):
|
|
return (self.snap or {}).get("phase")
|
|
|
|
|
|
async def act(c: Client) -> bool:
|
|
"""若该客户端有待办则提交一个动作，返回是否提交过。"""
|
|
p = c.pending()
|
|
if not p:
|
|
return False
|
|
key = p["key"]
|
|
if p.get("input_text"):
|
|
text = "大家听我分析，我目前是好人视角，先跟票看看。" if p.get("purpose") != "election" \
|
|
else "我想拿警徽带队，请大家支持。"
|
|
await c.send({"type": "speech", "text": text})
|
|
return True
|
|
if key == "elect":
|
|
await c.send({"type": "action", "action": "elect", "target": random.choice([1, None])})
|
|
return True
|
|
options = p.get("options") or []
|
|
if p.get("allow_skip") and random.random() < 0.15:
|
|
await c.send({"type": "action", "action": key, "target": None})
|
|
return True
|
|
if not options:
|
|
await c.send({"type": "action", "action": key, "target": None})
|
|
return True
|
|
me = c.seat()
|
|
pool = [o for o in options if o != me] or options
|
|
await c.send({"type": "action", "action": key, "target": random.choice(pool)})
|
|
return True
|
|
|
|
|
|
async def main() -> None:
|
|
req = urllib.request.Request(
|
|
f"{BASE}/api/rooms",
|
|
data=json.dumps({"board": "classic12", "host_name": "甲"}).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
code = json.loads(urllib.request.urlopen(req).read())["code"]
|
|
print("房间:", code)
|
|
|
|
a, b = Client("甲"), Client("乙")
|
|
await a.connect(code)
|
|
await b.connect(code)
|
|
await asyncio.sleep(0.6)
|
|
assert a.seat() == 1, f"甲座位错误: {a.seat()}"
|
|
assert b.seat() == 2, f"乙座位错误: {b.seat()}"
|
|
print("座位分配 OK: 甲=1号(房主), 乙=2号")
|
|
|
|
await a.send({"type": "control", "op": "fill_ai"})
|
|
await asyncio.sleep(0.5)
|
|
n = len(a.snap["players"])
|
|
assert n == 12, f"补AI后玩家数 {n}"
|
|
print("一键补满 AI OK: 12 名玩家")
|
|
|
|
# 乙断线重连（大厅阶段）
|
|
old_token = b.token
|
|
await b.ws.close()
|
|
await asyncio.sleep(0.4)
|
|
await b.connect(code, token=old_token)
|
|
await asyncio.sleep(0.6)
|
|
assert b.seat() == 2, f"重连后座位错误: {b.seat()}"
|
|
print("断线重连 OK: 乙重新回到 2 号")
|
|
|
|
await a.send({"type": "control", "op": "start"})
|
|
await asyncio.sleep(1.0)
|
|
assert a.phase() != "lobby", "开局失败"
|
|
print("开局 OK, 阶段:", a.snap["phase_text"])
|
|
print("甲身份:", a.snap["me"].get("role_name"), "| 乙身份:", b.snap["me"].get("role_name"))
|
|
|
|
# 自动打完整局
|
|
deadline = asyncio.get_event_loop().time() + 300
|
|
acted_log = []
|
|
while asyncio.get_event_loop().time() < deadline:
|
|
for c in (a, b):
|
|
if await act(c):
|
|
acted_log.append((c.name, c.phase()))
|
|
if a.phase() == "game_over" and b.phase() == "game_over":
|
|
break
|
|
await asyncio.sleep(0.4)
|
|
|
|
print("\n=== 对局结束 ===")
|
|
sa, sb = a.snap, b.snap
|
|
assert sa["phase"] == "game_over", f"超时未结束: {sa['phase']} {sa['phase_text']}"
|
|
print("胜者:", sa["winner"], "|", sa["winner_text"])
|
|
print("甲(1号)身份:", next(p["role_name"] for p in sa["players"] if p["seat"] == 1),
|
|
"存活:", next(p["alive"] for p in sa["players"] if p["seat"] == 1))
|
|
print("乙(2号)身份:", next(p["role_name"] for p in sb["players"] if p["seat"] == 2),
|
|
"存活:", next(p["alive"] for p in sb["players"] if p["seat"] == 2))
|
|
print("结算页显示全部身份:", all(p["role_name"] for p in sa["players"]))
|
|
|
|
# 对局中乙断线重连测试
|
|
token_b = b.token
|
|
await b.ws.close()
|
|
await asyncio.sleep(0.3)
|
|
await b.connect(code, token=token_b)
|
|
await asyncio.sleep(0.8)
|
|
assert b.seat() == 2 and b.phase() == "game_over"
|
|
print("对局结束后重连 OK")
|
|
|
|
# 房主 reset 回大厅再来一局
|
|
await a.send({"type": "control", "op": "reset"})
|
|
await asyncio.sleep(0.6)
|
|
assert a.phase() == "lobby", f"reset 失败: {a.phase()}"
|
|
print("再来一局(reset) OK")
|
|
await a.send({"type": "control", "op": "start"})
|
|
await asyncio.sleep(1.0)
|
|
assert a.phase() != "lobby", "第二局开局失败"
|
|
print("第二局开局 OK:", a.snap["phase_text"])
|
|
|
|
print("\n✅ 全部联机测试通过")
|
|
await a.ws.close()
|
|
await b.ws.close()
|
|
|
|
|
|
asyncio.run(main())
|