feat: data reload system
This commit is contained in:
bridge
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user