Release alpha 0.6.6 autochess flow
check / local-gate (push) Failing after 40s

This commit is contained in:
JWeBot Contributors
2026-10-15 17:10:28 +08:00
parent 9fff57134f
commit a74147e7f1
64 changed files with 1838 additions and 1060 deletions
+106 -436
View File
@@ -1,15 +1,11 @@
"""Behavioral coverage for the CS Major autochess extension."""
"""Behavioral coverage for the CS Major autochess 0.6.6 flow."""
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
@@ -18,31 +14,31 @@ 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.engine = MODULE.EdgeApplication(config(Path(self.temp.name)))
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:
def command(self, question, *, sender="Alice", event_id=None):
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, {})
response = self.engine.accept(
{
"event_id": event_id or f"autochess-{self.counter}",
"group_fingerprint": "group-a",
"sender": sender,
"text": "@JWeBot " + question,
"mention_bot": True,
"confidence": 1.0,
"sender_confidence": 1.0,
"body_confidence": 1.0,
}
)
self.assertEqual(response["status"], "reply")
return list(response.get("texts") or [response["text"]])
def state(self) -> dict:
def state(self):
row = self.engine.ledger.connection.execute(
"""
SELECT state_json FROM autochess_runs
@@ -54,56 +50,40 @@ class Release064AutochessTests(unittest.TestCase):
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 start_two_human_two_ai(self):
self.command("/自走棋开始")
self.command("/自走棋报名", sender="Bob")
self.command("/自走棋加AI战队")
return self.command("/自走棋开赛")
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 finish_draft(self):
self.command("/选1")
self.command("/选1", sender="Bob")
self.command("/选1", sender="Bob")
self.command("/选1")
self.command("/选1")
return self.command("/选1", sender="Bob")
def settle_to_completion(self) -> dict:
def ready_all(self):
state = self.state()
while state["status"] == "active":
self.command("/自走棋强制结算")
state = self.state()
return state
replies = []
for key in list(state.get("ready_keys") or []):
replies.extend(
self.command(
"/结束回合",
sender=state["players"][key]["display_name"],
)
)
return replies
def test_static_pool_parser_and_attribute_ranges(self):
def test_static_pool_parser_and_coach_pool(self):
self.assertEqual(len(AUTOCHESS.PLAYER_POOL), 60)
self.assertEqual(len({card["id"] for card in AUTOCHESS.PLAYER_POOL}), 60)
self.assertEqual(len(AUTOCHESS.COACH_POOL), 10)
self.assertEqual(
len({card["id"] for card in AUTOCHESS.PLAYER_POOL}),
60,
len({coach["id"] for coach in AUTOCHESS.COACH_POOL}),
10,
)
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", ""),
@@ -117,391 +97,81 @@ class Release064AutochessTests(unittest.TestCase):
("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("/自走棋开赛")
def test_launch_starts_multimessage_coach_draft_not_full_rosters(self):
response = self.start_two_human_two_ai()
state = self.state()
self.assertEqual(len(state["round_plan"]), 5)
self.assertGreaterEqual(len(response), 5)
self.assertIn("第一轮选秀", "\n".join(response))
self.assertIn("1. zonic", "\n".join(response))
self.assertEqual(state["phase"], "draft_coach_1")
self.assertEqual(state["ai_test_variant"], "two_plus_two")
self.assertEqual(
[stage["kind"] for stage in state["round_plan"]],
["pve", "pve", "pve", "semifinal", "final"],
state["seed_order"][2:],
[AUTOCHESS.AI_PERSON_KEY, AUTOCHESS.AI_SECOND_PERSON_KEY],
)
shops = []
for player in state["players"].values():
self.assertEqual(player["board"], [])
self.assertEqual(player["coach_id"], "")
def test_three_round_draft_training_and_group_intro(self):
self.start_two_human_two_ai()
final_draft = self.finish_draft()
state = self.state()
self.assertIn("三轮选秀结束", "\n".join(final_draft))
self.assertEqual(state["phase"], "lineup_prep")
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)
self.assertEqual(len(player["bench"]), 3)
self.assertTrue(player["coach_id"])
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"]
self.command("/结束回合")
training = self.command("/结束回合", sender="Bob")
self.assertIn("Team Falcons", "\n".join(training))
self.assertIn("Natus Vincere", "\n".join(training))
self.assertEqual(self.state()["phase"], "training_manage")
tactic = self.command("/战术残局")
self.assertIn("战术设为 残局", tactic)
coach = self.command("/升级教练")
self.assertIn("教练升到 2 级", coach)
self.command("/结束回合")
group = self.command("/结束回合", sender="Bob")
self.assertIn("正赛小组赛", "\n".join(group))
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)
self.assertEqual(state["phase"], "group_ready")
self.assertEqual(len(state["group_matches"]), 6)
def test_full_tournament_completes_and_only_active_playoff_players_ready(self):
self.start_two_human_two_ai()
self.finish_draft()
seen_playoff_prompt = False
for _ in range(80):
state = self.state()
if state["status"] == "completed":
break
ready_keys = list(state.get("ready_keys") or [])
self.assertTrue(ready_keys, state["phase"])
if state["phase"] in {"semifinal_ready", "final_ready"}:
pair = set(state["current_match"]["pair"])
self.assertEqual(set(ready_keys), pair & set(AUTOCHESS._human_keys(state)))
seen_playoff_prompt = True
self.ready_all()
else:
self.fail("tournament did not complete")
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)
self.assertTrue(seen_playoff_prompt)
self.assertEqual(state["status"], "completed")
self.assertEqual(state["phase"], "completed")
self.assertTrue(state["champion_key"])
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):
def test_multimessage_response_replays_identically(self):
self.command("/自走棋开始")
self.assertIn(
"2 名或 3 名真人",
self.command("/自走棋加AI战队"),
)
self.command("/自走棋报名", sender="Bob")
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)
self.command("/自走棋加AI战队")
first = self.command("/自走棋开赛", event_id="same-launch")
replay = self.command("/自走棋开赛", event_id="same-launch")
self.assertEqual(first, replay)
self.assertGreater(len(first), 1)
if __name__ == "__main__":