509 lines
18 KiB
Python
509 lines
18 KiB
Python
"""Behavioral coverage for the CS Major autochess extension."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import re
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
from test_jwebot_edge import MODULE, config
|
||
from jwebot_official_extensions import cs_autochess as AUTOCHESS
|
||
|
||
|
||
class Release064AutochessTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.temp = tempfile.TemporaryDirectory()
|
||
self.root = Path(self.temp.name)
|
||
self.engine = MODULE.EdgeApplication(config(self.root))
|
||
self.counter = 0
|
||
|
||
def tearDown(self):
|
||
self.engine.close()
|
||
self.temp.cleanup()
|
||
|
||
def command(
|
||
self,
|
||
question: str,
|
||
*,
|
||
sender: str = "Alice",
|
||
event_id: str | None = None,
|
||
) -> str:
|
||
self.counter += 1
|
||
row = {
|
||
"question": question,
|
||
"sender": sender,
|
||
"group_fingerprint": "group-a",
|
||
"event_id": event_id or f"autochess-{self.counter}",
|
||
}
|
||
return self.engine.local.handle(row, {})
|
||
|
||
def state(self) -> dict:
|
||
row = self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT state_json FROM autochess_runs
|
||
WHERE group_fingerprint = 'group-a'
|
||
ORDER BY season_number DESC
|
||
LIMIT 1
|
||
"""
|
||
).fetchone()
|
||
self.assertIsNotNone(row)
|
||
return json.loads(row[0])
|
||
|
||
def save_state(self, state: dict) -> None:
|
||
self.engine.ledger.connection.execute(
|
||
"""
|
||
UPDATE autochess_runs SET state_json = ?, state = ?
|
||
WHERE group_fingerprint = 'group-a' AND run_id = ?
|
||
""",
|
||
(
|
||
json.dumps(state, ensure_ascii=False, separators=(",", ":")),
|
||
state["status"],
|
||
state["run_id"],
|
||
),
|
||
)
|
||
|
||
def launch(self, names: list[str]) -> dict:
|
||
self.command("/自走棋开始", sender=names[0])
|
||
for name in names[1:]:
|
||
self.command("/自走棋报名", sender=name)
|
||
self.command("/自走棋开赛", sender=names[0])
|
||
return self.state()
|
||
|
||
def settle_to_completion(self) -> dict:
|
||
state = self.state()
|
||
while state["status"] == "active":
|
||
self.command("/自走棋强制结算")
|
||
state = self.state()
|
||
return state
|
||
|
||
def test_static_pool_parser_and_attribute_ranges(self):
|
||
self.assertEqual(len(AUTOCHESS.PLAYER_POOL), 60)
|
||
self.assertEqual(
|
||
len({card["id"] for card in AUTOCHESS.PLAYER_POOL}),
|
||
60,
|
||
)
|
||
ranges = {
|
||
1: (52, 68),
|
||
2: (60, 76),
|
||
3: (68, 84),
|
||
4: (78, 92),
|
||
5: (88, 98),
|
||
}
|
||
for card in AUTOCHESS.PLAYER_POOL:
|
||
with self.subTest(card=card["id"]):
|
||
lower, upper = ranges[card["cost"]]
|
||
for field in ("firepower", "stability", "clutch"):
|
||
self.assertGreaterEqual(card[field], lower)
|
||
self.assertLessEqual(card[field], upper)
|
||
self.assertIn(card["role"], {"步枪手", "狙击手", "指挥"})
|
||
self.assertLessEqual(len(card["teams"]), 3)
|
||
if card["role"] == "指挥":
|
||
self.assertGreater(card["leadership"], 0)
|
||
self.assertEqual(
|
||
AUTOCHESS.parse_autochess_action("/自走棋开始"),
|
||
("create", ""),
|
||
)
|
||
self.assertEqual(
|
||
AUTOCHESS.parse_autochess_action("/选6"),
|
||
("buy", "6"),
|
||
)
|
||
self.assertEqual(
|
||
AUTOCHESS.parse_autochess_action("/战术残局"),
|
||
("tactic", "残局"),
|
||
)
|
||
|
||
def test_round_plan_matches_player_count_without_duplicate_pairs(self):
|
||
for size in range(2, 9):
|
||
keys = [f"p{index}" for index in range(1, size + 1)]
|
||
plan = AUTOCHESS.build_round_plan(keys)
|
||
group_pairs = [
|
||
tuple(sorted(pair))
|
||
for stage in plan
|
||
if stage["kind"] == "group"
|
||
for pair in stage["pairs"]
|
||
]
|
||
with self.subTest(size=size):
|
||
if size == 2:
|
||
self.assertFalse(group_pairs)
|
||
self.assertFalse(
|
||
any(stage["kind"] == "semifinal" for stage in plan)
|
||
)
|
||
elif size == 4:
|
||
semifinal = next(
|
||
stage
|
||
for stage in plan
|
||
if stage["kind"] == "semifinal"
|
||
)
|
||
self.assertEqual(
|
||
semifinal["pairs"],
|
||
[["p1", "p4"], ["p2", "p3"]],
|
||
)
|
||
self.assertFalse(group_pairs)
|
||
else:
|
||
self.assertEqual(
|
||
len(group_pairs),
|
||
size * (size - 1) // 2,
|
||
)
|
||
self.assertEqual(len(group_pairs), len(set(group_pairs)))
|
||
self.assertEqual(plan[-1]["kind"], "final")
|
||
|
||
def test_launch_has_fixed_five_coach_shared_shop_and_complete_data(self):
|
||
response = self.command("/自走棋开始")
|
||
self.assertTrue(response.startswith("alice #1："))
|
||
self.command("/自走棋报名", sender="Bob")
|
||
self.command("/自走棋报名", sender="Carol")
|
||
self.command("/自走棋报名", sender="Dave")
|
||
response = self.command("/自走棋开赛")
|
||
state = self.state()
|
||
|
||
self.assertEqual(len(state["round_plan"]), 5)
|
||
self.assertEqual(
|
||
[stage["kind"] for stage in state["round_plan"]],
|
||
["pve", "pve", "pve", "semifinal", "final"],
|
||
)
|
||
shops = []
|
||
for player in state["players"].values():
|
||
self.assertEqual(len(player["board"]), 5)
|
||
self.assertEqual(player["coach_level"], 1)
|
||
self.assertEqual(player["tactic"], "稳定")
|
||
self.assertEqual(len(player["shop"]), 6)
|
||
shops.append(player["shop"])
|
||
self.assertTrue(all(shop == shops[0] for shop in shops))
|
||
for slot in range(1, 7):
|
||
line = next(
|
||
value
|
||
for value in response.splitlines()
|
||
if value.startswith(f"{slot}. ")
|
||
)
|
||
self.assertRegex(
|
||
line,
|
||
r"\d费 (?:步枪手|狙击手|指挥).*历史 .*火\d+ 稳\d+ 残\d+",
|
||
)
|
||
self.assertLessEqual(len(response), MODULE.MAX_REPLY_CHARS)
|
||
|
||
def test_tactics_coach_and_lineup_changes_preserve_five_starters(self):
|
||
state = self.launch(["Alice", "Bob"])
|
||
alice_key = state["seed_order"][0]
|
||
before_board = copy.deepcopy(state["players"][alice_key]["board"])
|
||
before_power = AUTOCHESS.lineup_power(
|
||
state["players"][alice_key]
|
||
)["power"]
|
||
|
||
tactic = self.command("/战术残局")
|
||
self.assertIn("战术设为 残局", tactic)
|
||
coach = self.command("/升级教练")
|
||
self.assertIn("教练升到 2 级", coach)
|
||
state = self.state()
|
||
alice = state["players"][alice_key]
|
||
self.assertEqual(alice["board"], before_board)
|
||
self.assertEqual(len(alice["board"]), 5)
|
||
self.assertGreater(AUTOCHESS.lineup_power(alice)["power"], before_power)
|
||
|
||
starter_ids = {unit["card_id"] for unit in alice["board"]}
|
||
slot = next(
|
||
index
|
||
for index, card_id in enumerate(alice["shop"], start=1)
|
||
if card_id not in starter_ids
|
||
)
|
||
self.command(f"/选{slot}")
|
||
self.assertIn("首发必须保持 5 人", self.command("/出售1"))
|
||
swapped = self.command("/上阵 6 2")
|
||
self.assertIn("首发仍为完整 5 人", swapped)
|
||
state = self.state()
|
||
self.assertEqual(len(state["players"][alice_key]["board"]), 5)
|
||
|
||
def test_replay_debits_one_purchase_and_invalid_archive_rolls_back(self):
|
||
state = self.launch(["Alice", "Bob"])
|
||
alice_key = state["seed_order"][0]
|
||
card_id = state["players"][alice_key]["shop"][0]
|
||
cost = AUTOCHESS.CARD_BY_ID[card_id]["cost"]
|
||
old_gold = state["players"][alice_key]["gold"]
|
||
old_pool = state["pool"][card_id]
|
||
replies: list[str] = []
|
||
errors: list[BaseException] = []
|
||
|
||
def buy() -> None:
|
||
try:
|
||
replies.append(
|
||
self.command("/选1", event_id="same-autochess-buy")
|
||
)
|
||
except BaseException as error:
|
||
errors.append(error)
|
||
|
||
threads = [threading.Thread(target=buy) for _ in range(8)]
|
||
for thread in threads:
|
||
thread.start()
|
||
for thread in threads:
|
||
thread.join()
|
||
self.assertFalse(errors)
|
||
self.assertEqual(len(set(replies)), 1)
|
||
state = self.state()
|
||
self.assertEqual(state["players"][alice_key]["gold"], old_gold - cost)
|
||
self.assertEqual(state["pool"][card_id], old_pool - 1)
|
||
|
||
before_json = self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT state_json FROM autochess_runs
|
||
WHERE group_fingerprint = 'group-a' AND run_id = ?
|
||
""",
|
||
(state["run_id"],),
|
||
).fetchone()[0]
|
||
before_stories = self.engine.ledger.connection.execute(
|
||
"SELECT COUNT(*) FROM autochess_story_logs"
|
||
).fetchone()[0]
|
||
|
||
def invalid_archive(current, person, next_season):
|
||
del person, next_season
|
||
current["players"][alice_key]["gold"] += 999
|
||
return {
|
||
"state": current,
|
||
"response": "invalid",
|
||
"stories": [{"bad": True}],
|
||
"battles": [],
|
||
}
|
||
|
||
with self.assertRaisesRegex(
|
||
MODULE.LocalCommandError,
|
||
"剧情归档无效",
|
||
):
|
||
self.engine.ledger.apply_autochess_action(
|
||
"group-a",
|
||
"Alice",
|
||
"invalid-autochess-archive",
|
||
"invalid",
|
||
invalid_archive,
|
||
)
|
||
after_json = self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT state_json FROM autochess_runs
|
||
WHERE group_fingerprint = 'group-a' AND run_id = ?
|
||
""",
|
||
(state["run_id"],),
|
||
).fetchone()[0]
|
||
after_stories = self.engine.ledger.connection.execute(
|
||
"SELECT COUNT(*) FROM autochess_story_logs"
|
||
).fetchone()[0]
|
||
self.assertEqual(after_json, before_json)
|
||
self.assertEqual(after_stories, before_stories)
|
||
|
||
def test_four_team_seed_bracket_is_bo5_with_mr15_scores(self):
|
||
state = self.launch(["Alice", "Bob", "Carol", "Dave"])
|
||
seed_order = list(state["seed_order"])
|
||
completed = self.settle_to_completion()
|
||
self.assertEqual(completed["status"], "completed")
|
||
self.assertTrue(completed["champion_key"])
|
||
rows = self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT battle_type, player_a, player_b, result_json
|
||
FROM autochess_battles
|
||
WHERE battle_type != 'pve'
|
||
ORDER BY battle_id
|
||
"""
|
||
).fetchall()
|
||
semifinals = [row for row in rows if row["battle_type"] == "semifinal_bo5"]
|
||
finals = [row for row in rows if row["battle_type"] == "final_bo5"]
|
||
self.assertEqual(
|
||
{
|
||
frozenset((row["player_a"], row["player_b"]))
|
||
for row in semifinals
|
||
},
|
||
{
|
||
frozenset((seed_order[0], seed_order[3])),
|
||
frozenset((seed_order[1], seed_order[2])),
|
||
},
|
||
)
|
||
self.assertEqual(len(finals), 1)
|
||
for row in [*semifinals, *finals]:
|
||
result = json.loads(row["result_json"])
|
||
self.assertEqual(result["series_format"], "BO5")
|
||
self.assertRegex(result["score"], r"^3-[0-2]$")
|
||
self.assertTrue(3 <= len(result["maps"]) <= 5)
|
||
for item in result["maps"]:
|
||
self.assertRegex(item["score"], r"^(?:16-(?:6|9|12|14)|19-17|22-20)$")
|
||
self.assertLessEqual(abs(item["fluctuation_a_bps"]), 200)
|
||
self.assertLessEqual(abs(item["fluctuation_b_bps"]), 200)
|
||
|
||
def test_five_team_round_robin_bo3_then_top_four(self):
|
||
completed = self.launch(["Alice", "Bob", "Carol", "Dave", "Eve"])
|
||
self.assertEqual(
|
||
sum(
|
||
len(stage.get("pairs", []))
|
||
for stage in completed["round_plan"]
|
||
if stage["kind"] == "group"
|
||
),
|
||
10,
|
||
)
|
||
completed = self.settle_to_completion()
|
||
self.assertEqual(len(completed["qualifiers"]), 4)
|
||
rows = self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT battle_type, player_a, player_b, result_json
|
||
FROM autochess_battles
|
||
WHERE battle_type != 'pve'
|
||
ORDER BY battle_id
|
||
"""
|
||
).fetchall()
|
||
groups = [row for row in rows if row["battle_type"] == "group_bo3"]
|
||
self.assertEqual(len(groups), 10)
|
||
self.assertEqual(
|
||
len(
|
||
{
|
||
frozenset((row["player_a"], row["player_b"]))
|
||
for row in groups
|
||
}
|
||
),
|
||
10,
|
||
)
|
||
for row in groups:
|
||
result = json.loads(row["result_json"])
|
||
self.assertEqual(result["series_format"], "BO3")
|
||
self.assertRegex(result["score"], r"^2-[01]$")
|
||
self.assertEqual(
|
||
sum(row["battle_type"] == "semifinal_bo5" for row in rows),
|
||
2,
|
||
)
|
||
self.assertEqual(
|
||
sum(row["battle_type"] == "final_bo5" for row in rows),
|
||
1,
|
||
)
|
||
|
||
def test_three_humans_plus_ai_has_taffy_host_and_turn_handoffs(self):
|
||
self.command("/自走棋开始")
|
||
self.command("/自走棋报名", sender="Bob")
|
||
self.assertIn(
|
||
"先有且仅有 3 名真人",
|
||
self.command("/自走棋加AI战队"),
|
||
)
|
||
self.command("/自走棋报名", sender="Carol")
|
||
self.assertIn(
|
||
"只有主持人",
|
||
self.command("/自走棋加AI战队", sender="Bob"),
|
||
)
|
||
self.assertIn("第 4 种子", self.command("/自走棋加AI战队"))
|
||
self.assertIn(
|
||
"已经锁定参赛名单",
|
||
self.command("/自走棋报名", sender="Dave"),
|
||
)
|
||
launch = self.command("/自走棋开赛")
|
||
state = self.state()
|
||
|
||
self.assertIn("Taffy：欢迎大家来到第 1 届 IEM 盐田 Major", launch)
|
||
self.assertIn("3 人+1 AI 测试赛", launch)
|
||
self.assertIn("1v4、2v3", launch)
|
||
self.assertIn("小组赛 BO3，淘汰赛 BO5", launch)
|
||
self.assertIn("常规时间 15-15 进入加时", launch)
|
||
self.assertIn("alice，请做好赛前准备", launch)
|
||
self.assertTrue(state["ai_test_mode"])
|
||
self.assertEqual(state["seed_order"][-1], AUTOCHESS.AI_PERSON_KEY)
|
||
self.assertTrue(
|
||
state["players"][AUTOCHESS.AI_PERSON_KEY]["is_ai"]
|
||
)
|
||
semifinal = state["round_plan"][-2]
|
||
self.assertEqual(
|
||
semifinal["pairs"],
|
||
[
|
||
[state["seed_order"][0], AUTOCHESS.AI_PERSON_KEY],
|
||
[state["seed_order"][1], state["seed_order"][2]],
|
||
],
|
||
)
|
||
|
||
self.assertIn(
|
||
"当前轮到 alice",
|
||
self.command("/选1", sender="Bob"),
|
||
)
|
||
self.assertIn(
|
||
"alice 准备完成！到 bob 做准备了",
|
||
self.command("/结束回合"),
|
||
)
|
||
self.assertIn(
|
||
"当前轮到 bob",
|
||
self.command("/结束回合", sender="Carol"),
|
||
)
|
||
self.assertIn(
|
||
"bob 准备完成！到 carol 做准备了",
|
||
self.command("/结束回合", sender="Bob"),
|
||
)
|
||
self.assertIn(
|
||
"全员准备完毕",
|
||
self.command("/结束回合", sender="Carol"),
|
||
)
|
||
settled = self.command("/自走棋结算")
|
||
self.assertIn("Taffy：AI 战队 自动运营", settled)
|
||
self.assertIn("Taffy：bob，请做好赛前准备", settled)
|
||
state = self.state()
|
||
self.assertEqual(
|
||
state["preparation_order"][0],
|
||
state["seed_order"][1],
|
||
)
|
||
self.assertEqual(
|
||
self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT COUNT(*) FROM autochess_story_logs
|
||
WHERE story_type = 'ai_operation'
|
||
"""
|
||
).fetchone()[0],
|
||
1,
|
||
)
|
||
self.assertEqual(
|
||
self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT COUNT(*) FROM autochess_story_logs
|
||
WHERE story_text NOT LIKE 'Taffy：%'
|
||
"""
|
||
).fetchone()[0],
|
||
0,
|
||
)
|
||
|
||
def test_identical_final_creates_persisted_overtime_stories(self):
|
||
state = self.launch(["Alice", "Bob"])
|
||
for _ in range(3):
|
||
self.command("/自走棋强制结算")
|
||
state = self.state()
|
||
self.assertEqual(AUTOCHESS._current_stage(state)["kind"], "final")
|
||
first, second = state["seed_order"]
|
||
state["players"][second]["board"] = copy.deepcopy(
|
||
state["players"][first]["board"]
|
||
)
|
||
state["players"][first]["tactic"] = "稳定"
|
||
state["players"][second]["tactic"] = "稳定"
|
||
self.save_state(state)
|
||
|
||
with mock.patch.object(AUTOCHESS, "_hash_int", return_value=0):
|
||
response = self.command(
|
||
"/自走棋强制结算",
|
||
event_id="forced-overtime-final",
|
||
)
|
||
self.assertIn("19-17", response)
|
||
final = self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT result_json FROM autochess_battles
|
||
WHERE battle_type = 'final_bo5'
|
||
"""
|
||
).fetchone()
|
||
result = json.loads(final["result_json"])
|
||
self.assertEqual(result["score"], "3-0")
|
||
self.assertEqual(
|
||
[item["score"] for item in result["maps"]],
|
||
["19-17", "19-17", "19-17"],
|
||
)
|
||
overtime = self.engine.ledger.connection.execute(
|
||
"""
|
||
SELECT story_text, facts_json FROM autochess_story_logs
|
||
WHERE story_type = 'overtime'
|
||
ORDER BY story_id
|
||
"""
|
||
).fetchall()
|
||
self.assertEqual(len(overtime), 3)
|
||
for row in overtime:
|
||
self.assertIn("常规时间战成 15-15", row["story_text"])
|
||
self.assertEqual(json.loads(row["facts_json"])["overtime_count"], 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|