183 lines
5.8 KiB
Python
183 lines
5.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from game.action_resolver import ActionResolver
|
|
from game.room_manager import RoomManager
|
|
from game.state_manager import StateManager
|
|
from game.story_engine import StoryEngine
|
|
from services.deepseek_client import DeepSeekClient
|
|
from services.persistence import SQLitePersistence
|
|
|
|
|
|
def _build_manager(db_path: str) -> RoomManager:
|
|
persistence = SQLitePersistence(db_path)
|
|
story_engine = StoryEngine()
|
|
state_manager = StateManager(persistence, story_engine)
|
|
action_resolver = ActionResolver()
|
|
deepseek_client = DeepSeekClient(api_key="")
|
|
return RoomManager(
|
|
persistence,
|
|
state_manager,
|
|
story_engine,
|
|
action_resolver,
|
|
deepseek_client,
|
|
max_players=8,
|
|
max_action_length=280,
|
|
)
|
|
|
|
|
|
def test_start_game_writes_initial_narration(test_db_path):
|
|
manager = _build_manager(test_db_path)
|
|
created = manager.create_room("Host")
|
|
room_id = created["room_id"]
|
|
host_id = created["player_id"]
|
|
|
|
manager.start_game(room_id, host_id)
|
|
snapshot = manager.get_snapshot(room_id)
|
|
|
|
ai_rows = [row for row in snapshot["messages"] if row["kind"] == "ai"]
|
|
assert ai_rows
|
|
assert "雾港旅店" in ai_rows[-1]["content"]
|
|
|
|
|
|
def test_single_player_action_auto_resolves_flag(test_db_path):
|
|
manager = _build_manager(test_db_path)
|
|
created = manager.create_room("Host")
|
|
room_id = created["room_id"]
|
|
host_id = created["player_id"]
|
|
|
|
manager.start_game(room_id, host_id)
|
|
action_result = manager.submit_action(room_id, host_id, "检查登记台")
|
|
|
|
assert action_result["should_auto_resolve"] is True
|
|
|
|
|
|
def test_multi_player_action_waits_until_all_online_submitted(test_db_path):
|
|
manager = _build_manager(test_db_path)
|
|
created = manager.create_room("Host")
|
|
room_id = created["room_id"]
|
|
host_id = created["player_id"]
|
|
|
|
join_info, _ = manager.join_room(room_id, "Mate")
|
|
mate_id = join_info["player_id"]
|
|
manager.set_ready(room_id, mate_id, True)
|
|
|
|
manager.start_game(room_id, host_id)
|
|
|
|
first = manager.submit_action(room_id, host_id, "先查看油画")
|
|
assert first["should_auto_resolve"] is False
|
|
|
|
second = manager.submit_action(room_id, mate_id, "检查登记台")
|
|
assert second["should_auto_resolve"] is True
|
|
|
|
|
|
def test_same_player_cannot_submit_twice_before_round_resolve(test_db_path):
|
|
manager = _build_manager(test_db_path)
|
|
created = manager.create_room("Host")
|
|
room_id = created["room_id"]
|
|
host_id = created["player_id"]
|
|
|
|
join_info, _ = manager.join_room(room_id, "Mate")
|
|
mate_id = join_info["player_id"]
|
|
manager.set_ready(room_id, mate_id, True)
|
|
manager.start_game(room_id, host_id)
|
|
|
|
manager.submit_action(room_id, host_id, "先查看前台账本")
|
|
with pytest.raises(ValueError, match="action_already_submitted"):
|
|
manager.submit_action(room_id, host_id, "再翻一遍前台账本")
|
|
|
|
|
|
def test_ai_failure_does_not_resolve_round_or_clear_pending_actions(test_db_path):
|
|
manager = _build_manager(test_db_path)
|
|
created = manager.create_room("Host")
|
|
room_id = created["room_id"]
|
|
host_id = created["player_id"]
|
|
manager.start_game(room_id, host_id)
|
|
|
|
manager.deepseek_client.chat_json = lambda *args, **kwargs: {
|
|
"ok": False,
|
|
"error": "request_timeout",
|
|
"detail": "Read timed out",
|
|
}
|
|
|
|
manager.submit_action(room_id, host_id, "检查登记台")
|
|
with pytest.raises(ValueError, match="ai_call_failed:request_timeout"):
|
|
manager.resolve_round(room_id, host_id)
|
|
|
|
snapshot = manager.get_snapshot(room_id)
|
|
assert snapshot["pending_actions"]
|
|
assert snapshot["game_state"]["round_no"] == 0
|
|
|
|
|
|
def test_concurrent_resolve_round_only_one_can_run(test_db_path):
|
|
manager = _build_manager(test_db_path)
|
|
created = manager.create_room("Host")
|
|
room_id = created["room_id"]
|
|
host_id = created["player_id"]
|
|
manager.start_game(room_id, host_id)
|
|
manager.submit_action(room_id, host_id, "检查登记台")
|
|
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def slow_ai(*args, **kwargs):
|
|
started.set()
|
|
release.wait(timeout=2)
|
|
return {
|
|
"ok": True,
|
|
"content": json.dumps(
|
|
{
|
|
"narration": "你翻开了登记台账本，发现被撕掉的一页。",
|
|
"npc_dialogues": [],
|
|
"state_updates": {
|
|
"danger_delta": 0,
|
|
"add_clues": [],
|
|
"set_flags": {},
|
|
"npc_changes": {},
|
|
"inventory_add": {},
|
|
"inventory_remove": {},
|
|
},
|
|
"private_player_updates": [],
|
|
"choices": ["继续调查"],
|
|
"should_end_scene": False,
|
|
"should_end_game": False,
|
|
"ending_summary": None,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
}
|
|
|
|
manager.deepseek_client.chat_json = slow_ai
|
|
|
|
outcomes: list[tuple[str, str]] = []
|
|
|
|
def runner():
|
|
try:
|
|
manager.resolve_round(room_id, host_id)
|
|
outcomes.append(("ok", ""))
|
|
except ValueError as exc:
|
|
outcomes.append(("err", str(exc)))
|
|
|
|
t1 = threading.Thread(target=runner)
|
|
t1.start()
|
|
assert started.wait(timeout=1.5)
|
|
|
|
t2 = threading.Thread(target=runner)
|
|
t2.start()
|
|
t2.join(timeout=2)
|
|
|
|
release.set()
|
|
t1.join(timeout=2)
|
|
|
|
ok_count = sum(1 for kind, _ in outcomes if kind == "ok")
|
|
err_codes = [code for kind, code in outcomes if kind == "err"]
|
|
assert ok_count == 1
|
|
assert any(code == "round_resolving" for code in err_codes)
|
|
|
|
snapshot = manager.get_snapshot(room_id)
|
|
assert snapshot["game_state"]["round_no"] == 1
|