404 lines
17 KiB
Python
404 lines
17 KiB
Python
import pytest
|
||
import json
|
||
from unittest.mock import MagicMock, AsyncMock, patch
|
||
from pathlib import Path
|
||
|
||
from src.classes.history import HistoryManager, History
|
||
from src.classes.region import CityRegion, NormalRegion, CultivateRegion
|
||
from src.classes.sect_region import SectRegion
|
||
from src.classes.technique import Technique, TechniqueAttribute, TechniqueGrade
|
||
from src.classes.weapon import Weapon, WeaponType
|
||
from src.classes.auxiliary import Auxiliary
|
||
from src.classes.cultivation import Realm
|
||
from src.classes.item_registry import ItemRegistry
|
||
from src.classes.sect import Sect, SectHeadQuarter
|
||
from src.classes.alignment import Alignment
|
||
from src.sim.load.load_game import apply_history_modifications
|
||
|
||
# 保存原始方法，因为 conftest 中全局 mock 了它，导致部分集成测试失效
|
||
# 我们需要在这个变量中保存引用，以便在特定测试中恢复它
|
||
_real_apply_history_influence = HistoryManager.apply_history_influence
|
||
|
||
# 假设这些全局字典在模块层级
|
||
from src.classes import technique as technique_module
|
||
from src.classes import weapon as weapon_module
|
||
from src.classes import sect as sect_module
|
||
from src.classes import auxiliary as auxiliary_module
|
||
|
||
# --- 1. 基础数据结构测试 (Plan 1) ---
|
||
|
||
def test_world_history_structure(base_world):
|
||
"""
|
||
目标：验证 World 与 History dataclass 的交互是否符合预期。
|
||
"""
|
||
# 初始化：验证 world.history 是否自动初始化为空的 History 对象
|
||
assert isinstance(base_world.history, History)
|
||
assert base_world.history.text == ""
|
||
assert base_world.history.modifications == {}
|
||
|
||
# 设置文本：验证 text 更新
|
||
history_text = "修仙界风云变幻"
|
||
base_world.set_history(history_text)
|
||
assert base_world.history.text == history_text
|
||
|
||
# 记录差分：验证 record_modification
|
||
# 第一次记录
|
||
base_world.record_modification("sects", "1", {"name": "新名字"})
|
||
assert "sects" in base_world.history.modifications
|
||
assert "1" in base_world.history.modifications["sects"]
|
||
assert base_world.history.modifications["sects"]["1"]["name"] == "新名字"
|
||
|
||
# 第二次记录：验证合并 (Merge)
|
||
base_world.record_modification("sects", "1", {"desc": "新描述"})
|
||
assert base_world.history.modifications["sects"]["1"]["name"] == "新名字" # 旧属性保留
|
||
assert base_world.history.modifications["sects"]["1"]["desc"] == "新描述" # 新属性添加
|
||
|
||
# 第三次记录：验证覆盖 (Override)
|
||
base_world.record_modification("sects", "1", {"name": "更新的名字"})
|
||
assert base_world.history.modifications["sects"]["1"]["name"] == "更新的名字"
|
||
|
||
|
||
# --- 2. 修改记录行为测试 (Plan 2) ---
|
||
|
||
def test_history_manager_records_changes(base_world):
|
||
"""
|
||
目标：验证 HistoryManager 在修改对象时，是否自动产生“留痕”。
|
||
"""
|
||
# Setup
|
||
sect = Sect(id=1, name="OldSect", desc="OldDesc", member_act_style="", alignment=Alignment.RIGHTEOUS, headquarter=None, technique_names=[])
|
||
manager = HistoryManager(base_world)
|
||
|
||
# Execute Modification
|
||
# 模拟 _update_obj_attrs 的调用
|
||
changes = {"name": "NewSect", "desc": "NewDesc"}
|
||
manager._update_obj_attrs(sect, changes, category="sects", id_str="1")
|
||
|
||
# Verify Object Updated
|
||
assert sect.name == "NewSect"
|
||
assert sect.desc == "NewDesc"
|
||
|
||
# Verify History Recorded
|
||
assert "sects" in base_world.history.modifications
|
||
assert "1" in base_world.history.modifications["sects"]
|
||
recorded_change = base_world.history.modifications["sects"]["1"]
|
||
assert recorded_change["name"] == "NewSect"
|
||
assert recorded_change["desc"] == "NewDesc"
|
||
|
||
|
||
# --- 3. 差分回放逻辑测试 (Plan 3) ---
|
||
|
||
def test_apply_history_modifications_logic(base_world):
|
||
"""
|
||
目标：验证 apply_history_modifications 函数能否将数据字典正确应用到静态对象上。
|
||
"""
|
||
# Setup Objects
|
||
sect = Sect(id=1, name="OriginalSect", desc="OriginalDesc", member_act_style="", alignment=Alignment.RIGHTEOUS, headquarter=None, technique_names=[])
|
||
weapon = Weapon(id=101, name="OriginalSword", weapon_type=WeaponType.SWORD, realm=Realm.Qi_Refinement, desc="OriginalDesc")
|
||
|
||
# Patch Global Registries
|
||
with patch.dict(sect_module.sects_by_id, {1: sect}, clear=True), \
|
||
patch.dict(sect_module.sects_by_name, {"OriginalSect": sect}, clear=True), \
|
||
patch.object(ItemRegistry, "get", return_value=weapon), \
|
||
patch.dict(weapon_module.weapons_by_name, {"OriginalSword": weapon}, clear=True):
|
||
|
||
# Construct Modifications
|
||
modifications = {
|
||
"sects": {
|
||
"1": {"name": "ReplayedSect", "desc": "ReplayedDesc"},
|
||
"999": {"name": "GhostSect"} # 不存在的 ID，应忽略
|
||
},
|
||
"weapons": {
|
||
"101": {"name": "ReplayedSword"}
|
||
}
|
||
}
|
||
|
||
# Execute Replay
|
||
apply_history_modifications(base_world, modifications)
|
||
|
||
# Verify Sect Updated
|
||
assert sect.name == "ReplayedSect"
|
||
assert sect.desc == "ReplayedDesc"
|
||
|
||
# Verify Sect Index Synced (Old name removed, new name added)
|
||
assert "OriginalSect" not in sect_module.sects_by_name
|
||
assert "ReplayedSect" in sect_module.sects_by_name
|
||
assert sect_module.sects_by_name["ReplayedSect"] == sect
|
||
|
||
# Verify Weapon Updated
|
||
assert weapon.name == "ReplayedSword"
|
||
assert weapon.desc == "OriginalDesc" # Unchanged
|
||
|
||
# Verify Weapon Index Synced
|
||
assert "OriginalSword" not in weapon_module.weapons_by_name
|
||
assert "ReplayedSword" in weapon_module.weapons_by_name
|
||
|
||
|
||
# @pytest.mark.asyncio
|
||
# async def test_history_influence_complex_verification_WIP(base_world):
|
||
# """
|
||
# TODO: 该测试代码片段在合并过程中被隔离，缺少上下文变量 (manager, city_region 等)。
|
||
# 需要重新组织数据准备代码后启用。
|
||
# """
|
||
# # Mock call_llm_with_task_name
|
||
# # with patch("src.classes.history.call_llm_with_task_name", new_callable=AsyncMock) as mock_llm:
|
||
# # mock_llm.side_effect = side_effect
|
||
# #
|
||
# # # --- Execute ---
|
||
# # history_text = "Some history text"
|
||
# # await manager.apply_history_influence(history_text)
|
||
# #
|
||
# # # --- Assertions ---
|
||
# #
|
||
# # # 0. World history 未自动设置（需要外部调用 set_history）
|
||
# # # 注意：apply_history_influence 只应用影响，不设置 history 属性
|
||
# # # history 属性应该在调用前由外部设置
|
||
# #
|
||
# # # 1. LLM Called 3 times
|
||
# # assert mock_llm.call_count == 3
|
||
# #
|
||
# # # 2. Map Regions Updated
|
||
# # assert city_region.name == "NewCity"
|
||
# # assert city_region.desc == "New Desc"
|
||
# # assert normal_region.name == "NewWild"
|
||
# # assert normal_region.desc == "New Wild Desc"
|
||
# # assert cult_region.name == "NewCave"
|
||
# # assert cult_region.desc == "New Cave Desc"
|
||
# #
|
||
# # # 2.1 resolve_query can find regions by new names
|
||
# # from src.utils.resolution import resolve_query
|
||
# # from src.classes.region import Region
|
||
# # assert resolve_query("NewCity", base_world, expected_types=[Region]).obj == city_region
|
||
# # assert resolve_query("NewWild", base_world, expected_types=[Region]).obj == normal_region
|
||
# # assert resolve_query("NewCave", base_world, expected_types=[Region]).obj == cult_region
|
||
# # # Old names should no longer resolve (region objects have new names)
|
||
# # assert resolve_query("OldCity", base_world, expected_types=[Region]).obj is None
|
||
# # assert resolve_query("OldWild", base_world, expected_types=[Region]).obj is None
|
||
# # assert resolve_query("OldCave", base_world, expected_types=[Region]).obj is None
|
||
# #
|
||
# # # 3. Sect & Sect Region Updated
|
||
# # assert sect.name == "NewSect"
|
||
# # assert sect.desc == "New Sect Desc"
|
||
# # assert sect_region_obj.name == "NewSectHQ" # 地图上的对象被更新
|
||
# # assert sect_region_obj.desc == "New Sect HQ Desc"
|
||
# #
|
||
# # # 4. Sect Index Synced
|
||
# # assert "NewSect" in sect_module.sects_by_name
|
||
# # assert "OldSect" not in sect_module.sects_by_name
|
||
# # assert sect_module.sects_by_name["NewSect"] == sect
|
||
# #
|
||
# # # 5. Technique Updated & Index Synced
|
||
# # assert tech.name == "NewTech"
|
||
# # assert tech.desc == "New Tech Desc"
|
||
# # assert "NewTech" in technique_module.techniques_by_name
|
||
# # assert "OldTech" not in technique_module.techniques_by_name
|
||
# # assert technique_module.techniques_by_name["NewTech"] == tech
|
||
# #
|
||
# # # 6. Weapon Updated & Index Synced
|
||
# # assert weapon.name == "NewSword"
|
||
# # assert weapon.desc == "New Sword Desc"
|
||
# # assert "NewSword" in weapon_module.weapons_by_name
|
||
# # assert "OldSword" not in weapon_module.weapons_by_name
|
||
# # assert weapon_module.weapons_by_name["NewSword"] == weapon
|
||
# #
|
||
# # # 7. Auxiliary Updated
|
||
# # assert aux.name == "NewOrb"
|
||
# # assert aux.desc == "New Orb Desc"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_history_workflow_integration(base_world):
|
||
"""测试完整的历史工作流程：设置历史 -> 应用影响"""
|
||
# 准备测试数据
|
||
city_region = CityRegion(id=1, name="测试城", desc="旧描述")
|
||
base_world.map.regions = {1: city_region}
|
||
|
||
# 模拟初始化时的完整流程
|
||
history_text = "这片大陆曾经历过灵气复苏，修仙宗门林立。"
|
||
|
||
# 1. 先设置 history（模拟 init_game_async 中的调用）
|
||
base_world.set_history(history_text)
|
||
assert base_world.history.text == history_text
|
||
|
||
# 2. 验证 static_info 中包含历史
|
||
static_info = base_world.static_info
|
||
assert "历史" in static_info
|
||
assert static_info["历史"] == history_text
|
||
|
||
# 3. 应用历史影响（模拟 HistoryManager.apply_history_influence）
|
||
manager = HistoryManager(base_world)
|
||
manager._read_csv = MagicMock(return_value="dummy,csv,content")
|
||
|
||
map_response = {
|
||
"city_regions_change": {"1": {"name": "灵气城", "desc": "充满灵气的城市"}},
|
||
}
|
||
# TODO: 这里只设置了变量，没有实际调用 assert 或 mock side_effect 进行验证
|
||
# 可能是测试还没写完，暂时保持原样
|
||
|
||
|
||
# --- 4. 集成测试：存读档全流程 (Plan 4) ---
|
||
|
||
def test_save_load_integration_with_history(base_world, tmp_path):
|
||
"""
|
||
目标：模拟真实游戏场景，验证“修改 -> 存档 -> 重置 -> 读档 -> 还原”的闭环。
|
||
"""
|
||
from src.sim.save.save_game import save_game
|
||
from src.sim.load.load_game import load_game
|
||
from src.sim.simulator import Simulator
|
||
|
||
# 1. Setup Initial State
|
||
sect = Sect(id=1, name="OriginalSect", desc="OriginalDesc", member_act_style="", alignment=Alignment.RIGHTEOUS, headquarter=None, technique_names=[])
|
||
|
||
# Patch 全局状态，模拟游戏运行环境
|
||
with patch.dict(sect_module.sects_by_id, {1: sect}, clear=True), \
|
||
patch.dict(sect_module.sects_by_name, {"OriginalSect": sect}, clear=True):
|
||
|
||
# 2. Apply Changes & Record History
|
||
history_text = "History Text"
|
||
base_world.set_history(history_text)
|
||
|
||
# 模拟 HistoryManager 的修改操作
|
||
sect.name = "ModifiedSect"
|
||
base_world.record_modification("sects", "1", {"name": "ModifiedSect"})
|
||
# 此时内存中是 ModifiedSect
|
||
|
||
simulator = Simulator(base_world)
|
||
existed_sects = [sect]
|
||
|
||
# 3. Save Game
|
||
save_path = tmp_path / "integration_save.json"
|
||
save_game(base_world, simulator, existed_sects, save_path)
|
||
|
||
# 4. Reset Memory (Simulate Restart)
|
||
# 将对象重置为原始状态，模拟重新加载配置文件的过程
|
||
sect.name = "OriginalSect"
|
||
if "ModifiedSect" in sect_module.sects_by_name:
|
||
del sect_module.sects_by_name["ModifiedSect"]
|
||
sect_module.sects_by_name["OriginalSect"] = sect
|
||
|
||
assert sect.name == "OriginalSect" # 确认重置成功
|
||
|
||
# 5. Load Game
|
||
# load_game 会调用 apply_history_modifications
|
||
# 注意：load_game 内部会导入 sects_by_id，我们已经在 patch 中了，所以 load_game 会看到我们 patch 的 dict
|
||
# 但 load_game 会重建 world，所以我们需要验证 loaded_world
|
||
|
||
loaded_world, _, _ = load_game(save_path)
|
||
|
||
# 6. Verify
|
||
# 验证历史文本
|
||
assert loaded_world.history.text == history_text
|
||
|
||
# 验证 Modifications 数据存在
|
||
assert "sects" in loaded_world.history.modifications
|
||
assert loaded_world.history.modifications["sects"]["1"]["name"] == "ModifiedSect"
|
||
|
||
# 核心验证：内存中的对象是否变回了 ModifiedSect
|
||
# 因为我们 patch 了全局字典，load_game 回放时修改的就是这个全局字典里的 sect 对象
|
||
assert sect.name == "ModifiedSect"
|
||
assert "ModifiedSect" in sect_module.sects_by_name
|
||
assert "OriginalSect" not in sect_module.sects_by_name
|
||
|
||
|
||
# --- 5. 边界情况测试 (Plan 5) ---
|
||
|
||
def test_history_boundary_cases(base_world, tmp_path):
|
||
"""
|
||
目标：边界情况测试
|
||
"""
|
||
from src.sim.save.save_game import save_game
|
||
from src.sim.load.load_game import load_game
|
||
from src.sim.simulator import Simulator
|
||
|
||
# Case 1: Empty History
|
||
# 保存一个没有历史修改的存档
|
||
simulator = Simulator(base_world)
|
||
save_path = tmp_path / "empty_history.json"
|
||
save_game(base_world, simulator, [], save_path)
|
||
|
||
# 读档应不报错
|
||
loaded_world, _, _ = load_game(save_path)
|
||
assert loaded_world.history.text == ""
|
||
assert loaded_world.history.modifications == {}
|
||
|
||
# Case 2: Corrupted/Partial Modification Data (Manual JSON edit)
|
||
# 创建一个手动修改的 JSON，模拟数据损坏或旧版本残留
|
||
with open(save_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
# 注入一个格式奇怪的 modifications
|
||
data["world"]["history"] = {
|
||
"text": "Partial",
|
||
"modifications": {
|
||
"sects": {
|
||
"invalid_id": {"name": "ShouldNotCrash"} # ID 不是数字，但 key 是 str
|
||
},
|
||
"unknown_category": { # 未知类别
|
||
"1": {"name": "???"}
|
||
}
|
||
}
|
||
}
|
||
|
||
with open(save_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f)
|
||
|
||
# 读档应鲁棒处理，不崩溃
|
||
loaded_world_2, _, _ = load_game(save_path)
|
||
assert loaded_world_2.history.text == "Partial"
|
||
|
||
# 验证 static_info 中包含历史 (pr-35)
|
||
static_info = loaded_world_2.static_info
|
||
assert "历史" in static_info, "加载后的 static_info 应该包含历史"
|
||
assert static_info["历史"] == "Partial", "加载后的历史文本应该正确"
|
||
|
||
# 确保未知类别被加载（作为数据），但在 apply 时被忽略（不报错） (main)
|
||
assert "unknown_category" in loaded_world_2.history.modifications
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_move_to_region_after_history_rename(base_world, dummy_avatar):
|
||
"""
|
||
测试 MoveToRegion action 在 history 修改 region 名称后能正确工作。
|
||
|
||
这是对以下失败场景的回归测试：
|
||
WARNING - 非法动作: Avatar(name=苏梦蝶) 的动作 MoveToRegion
|
||
参数={'region': '沧澜潮汐城'} 无法启动，原因=无法解析区域: 沧澜潮汐城
|
||
"""
|
||
from src.classes.action.move_to_region import MoveToRegion
|
||
|
||
# 准备：创建一个城市区域
|
||
city_region = CityRegion(id=304, name="沧澜城", desc="原始描述")
|
||
base_world.map.regions = {304: city_region}
|
||
|
||
# 模拟 history 修改了城市名称（如 LLM 返回的新名称）
|
||
manager = HistoryManager(base_world)
|
||
manager._read_csv = MagicMock(return_value="dummy")
|
||
|
||
map_response = {
|
||
"city_regions_change": {"304": {"name": "沧澜潮汐城", "desc": "新描述"}}
|
||
}
|
||
|
||
def side_effect(**kwargs):
|
||
if kwargs.get("task_name") == "history_influence_map":
|
||
return map_response
|
||
return {}
|
||
|
||
with patch("src.classes.history.call_llm_with_task_name", new_callable=AsyncMock) as mock_llm:
|
||
mock_llm.side_effect = side_effect
|
||
|
||
# 临时恢复真实的 apply_history_influence 方法，因为 conftest 把它 mock 掉了
|
||
with patch.object(HistoryManager, 'apply_history_influence', new=_real_apply_history_influence):
|
||
await manager.apply_history_influence("测试历史")
|
||
|
||
# 验证名称已修改
|
||
assert city_region.name == "沧澜潮汐城"
|
||
|
||
# 核心测试：MoveToRegion 使用新名称应该能成功
|
||
move_action = MoveToRegion(dummy_avatar, base_world)
|
||
can_start, reason = move_action.can_start("沧澜潮汐城")
|
||
|
||
assert can_start is True, f"MoveToRegion 应该能解析新名称，但失败了: {reason}"
|
||
|
||
# 旧名称应该无法解析（因为 region 对象的 name 已经变了）
|
||
can_start_old, reason_old = move_action.can_start("沧澜城")
|
||
assert can_start_old is False, "旧名称不应该能解析"
|
||
assert "无法解析区域" in reason_old
|