From 47ad330b355d52f054fd981a665dc31b4f1c5680 Mon Sep 17 00:00:00 2001 From: bridge Date: Sun, 18 Jan 2026 16:53:24 +0800 Subject: [PATCH] feat: data reload system --- pyproject.toml | 1 + tests/test_history.py | 446 +++++++++++++++++++----------------------- 2 files changed, 207 insertions(+), 240 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 20754d3..7b42f29 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,4 +2,5 @@ pythonpath = "." testpaths = ["tests"] asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" norecursedirs = ["tmp", "assets", "node_modules", "dist", "build", "web", "tools"] diff --git a/tests/test_history.py b/tests/test_history.py index 3a1b7f8..3c88867 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -1,8 +1,10 @@ import pytest +import json from unittest.mock import MagicMock, AsyncMock, patch from pathlib import Path -from src.classes.history import HistoryManager -from src.classes.region import CityRegion, NormalRegion, CultivateRegion, Region + +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 @@ -11,268 +13,232 @@ 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 # 假设这些全局字典在模块层级 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 -def test_world_set_history(base_world): - """测试 world.set_history 方法和 static_info 中的历史显示""" - # 初始状态：无历史 - assert base_world.history == "" - static_info = base_world.static_info - assert "历史" not in static_info - - # 设置历史 - history_text = "这是一段测试历史文本：修仙界曾发生大战，许多宗门覆灭。" +# --- 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 == history_text - - # 验证 static_info 包含历史 - static_info = base_world.static_info - assert "历史" in static_info - assert static_info["历史"] == history_text + assert base_world.history.text == history_text -@pytest.mark.asyncio -async def test_history_influence(base_world): - # --- Setup Test Data --- - - # 1. Regions - city_region = CityRegion(id=1, name="OldCity", desc="Old Desc") - normal_region = NormalRegion(id=2, name="OldWild", desc="Old Wild Desc") - cult_region = CultivateRegion(id=3, name="OldCave", desc="Old Cave Desc") - # 假设 ID 4 是宗门驻地在地图上的区域对象 - sect_region_obj = SectRegion(id=4, name="OldSectHQ", desc="Old Sect HQ Desc", sect_name="OldSect", sect_id=1) - - base_world.map.regions = { - 1: city_region, - 2: normal_region, - 3: cult_region, - 4: sect_region_obj - } - - # 2. Sects - sect = Sect( - id=1, - name="OldSect", - desc="Old Sect Desc", - member_act_style="Old Style", - alignment=Alignment.RIGHTEOUS, - headquarter=SectHeadQuarter(name="OldHQ", desc="Old HQ Desc", image=None), - technique_names=[] - ) + # 记录差分：验证 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"] == "新名字" - # 3. Techniques - tech = Technique( - id=101, - name="OldTech", - attribute=TechniqueAttribute.GOLD, - grade=TechniqueGrade.LOWER, - desc="Old Tech Desc", - weight=1.0, - condition="" - ) - - # 4. Weapons & Auxiliaries - weapon = Weapon( - id=201, - name="OldSword", - weapon_type=WeaponType.SWORD, - realm=Realm.Qi_Refinement, - desc="Old Sword Desc" - ) - aux = Auxiliary( - id=301, - name="OldOrb", - realm=Realm.Qi_Refinement, - desc="Old Orb Desc" - ) + # 第二次记录：验证合并 (Merge) + base_world.record_modification("sects", "1", {"desc": "新描述"}) + assert base_world.history.modifications["sects"]["1"]["name"] == "新名字" # 旧属性保留 + assert base_world.history.modifications["sects"]["1"]["desc"] == "新描述" # 新属性添加 - # --- Patch Global Registries --- - with patch.dict(technique_module.techniques_by_id, {101: tech}, clear=True), \ - patch.dict(technique_module.techniques_by_name, {"OldTech": tech}, clear=True), \ - patch.dict(weapon_module.weapons_by_name, {"OldSword": weapon}, clear=True), \ - patch.dict(sect_module.sects_by_id, {1: sect}, clear=True), \ - patch.dict(sect_module.sects_by_name, {"OldSect": sect}, clear=True), \ - patch.object(ItemRegistry, "_items_by_id", {201: weapon, 301: aux}): - - # --- Prepare LLM Mock Responses --- - # Map Task Response - map_response = { - "city_regions_change": {"1": {"name": "NewCity", "desc": "New Desc"}}, - "normal_regions_change": {"2": {"name": "NewWild", "desc": "New Wild Desc"}}, - "cultivate_regions_change": {"3": {"name": "NewCave", "desc": "New Cave Desc"}} - } - - # Sect Task Response - sect_response = { - "sects_change": {"1": {"name": "NewSect", "desc": "New Sect Desc"}}, - "sect_regions_change": {"4": {"name": "NewSectHQ", "desc": "New Sect HQ Desc"}} - } - - # Item Task Response - item_response = { - "techniques_change": {"101": {"name": "NewTech", "desc": "New Tech Desc"}}, - "weapons_change": {"201": {"name": "NewSword", "desc": "New Sword Desc"}}, - "auxiliarys_change": {"301": {"name": "NewOrb", "desc": "New Orb Desc"}} - } + # 第三次记录：验证覆盖 (Override) + base_world.record_modification("sects", "1", {"name": "更新的名字"}) + assert base_world.history.modifications["sects"]["1"]["name"] == "更新的名字" - def side_effect(**kwargs): - task_name = kwargs.get("task_name") - if task_name == "history_influence_map": - return map_response - elif task_name == "history_influence_sect": - return sect_response - elif task_name == "history_influence_item": - return item_response - return {} - # --- Instantiate Manager & Mock Internal Methods --- - manager = HistoryManager(base_world) - manager._read_csv = MagicMock(return_value="dummy,csv,content") - - # 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" - - # 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 +# --- 2. 修改记录行为测试 (Plan 2) --- - # 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 == 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） +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) - manager._read_csv = MagicMock(return_value="dummy,csv,content") - map_response = { - "city_regions_change": {"1": {"name": "灵气城", "desc": "充满灵气的城市"}}, - } - - def side_effect(**kwargs): - task_name = kwargs.get("task_name") - if 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 - - await manager.apply_history_influence(history_text) - - # 4. 验证影响已应用 - assert city_region.name == "灵气城" - assert city_region.desc == "充满灵气的城市" - - # 5. 验证 history 仍然保留 - assert base_world.history == history_text - - # 6. 验证 static_info 中仍包含历史 - static_info = base_world.static_info - assert "历史" in static_info - assert static_info["历史"] == history_text + # Execute Modification + # 模拟 _update_obj_attrs 的调用 + changes = {"name": "NewSect", "desc": "NewDesc"} + manager._update_obj_attrs(sect, changes, category="sects", id_str="1") -def test_history_persistence_in_save_load(base_world, tmp_path): - """测试 history 在保存和加载时的持久化""" + # 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 + + +# --- 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 - # 设置历史 - history_text = "修仙界的远古历史：曾有强者飞升，留下诸多传承。" - base_world.set_history(history_text) + # 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) - existed_sects = [] + save_path = tmp_path / "empty_history.json" + save_game(base_world, simulator, [], save_path) - # 保存游戏 - save_path = tmp_path / "test_history_save.json" - success, _ = save_game(base_world, simulator, existed_sects, save_path) - assert success, "保存游戏应该成功" - - # 验证保存文件中包含历史 - import json + # 读档应不报错 + 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: - save_data = json.load(f) + data = json.load(f) - world_data = save_data.get("world", {}) - assert "history" in world_data, "保存数据应该包含 history 字段" - assert world_data["history"] == history_text, "保存的历史文本应该正确" + # 注入一个格式奇怪的 modifications + data["world"]["history"] = { + "text": "Partial", + "modifications": { + "sects": { + "invalid_id": {"name": "ShouldNotCrash"} # ID 不是数字，但 key 是 str + }, + "unknown_category": { # 未知类别 + "1": {"name": "???"} + } + } + } - # 加载游戏 - loaded_world, loaded_sim, loaded_sects = load_game(save_path) + 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" - # 验证历史被正确恢复 - assert loaded_world.history == history_text, "加载的世界应该包含历史" - - # 验证 static_info 中包含历史 - static_info = loaded_world.static_info - assert "历史" in static_info, "加载后的 static_info 应该包含历史" - assert static_info["历史"] == history_text, "加载后的历史文本应该正确" + # 确保未知类别被加载（作为数据），但在 apply 时被忽略（不报错） + assert "unknown_category" in loaded_world_2.history.modifications +