148 lines
8.6 KiB
Python
148 lines
8.6 KiB
Python
# ============================================================
|
||
# M2 状态治理黄金测试（moduleId: golden-m2-state, 发版门禁 §14.3）
|
||
# 固化三个核心不变式：
|
||
# S1 Explore 沙盒隔离：方案对比绝不污染主干世界（§5.1 铁律）
|
||
# S2 Checkpoint 往返：建档 → 破坏 → 回滚 → 完整恢复（§4.3）
|
||
# S3 审计哈希链：正常链校验通过；篡改任意字段即断链（§3.6）
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import copy # 世界对照拷贝
|
||
import json # 深比较辅助
|
||
|
||
from server.agent_core.audit import write_audit # 审计写入
|
||
from server.agent_core.registry import verify_audit_chain # 链校验
|
||
from server.aps_domain.scenario import compare_scenarios # 方案对比（沙盒）
|
||
from server.state.checkpoints import CheckpointStore # 快照仓（用临时路径实例化）
|
||
from server.state.seed import seed_world # 种子世界
|
||
|
||
|
||
def _next_id_factory():
|
||
"""独立发号器（用例内存态）。"""
|
||
counters: dict[str, int] = {} # 计数器
|
||
def next_id(kind: str) -> int: # 闭包发号
|
||
counters[kind] = counters.get(kind, 0) + 1 # 自增
|
||
return counters[kind] # 返回
|
||
return next_id # 闭包
|
||
|
||
|
||
# ---------------- S1：沙盒隔离 ----------------
|
||
def test_scenario_compare_does_not_mutate_world():
|
||
"""方案对比在深拷贝沙盒运行：主干世界的所有表必须逐字节不变。"""
|
||
world = seed_world() # 主干世界
|
||
before = json.dumps(world, ensure_ascii=False, sort_keys=True) # 运行前指纹
|
||
text, block = compare_scenarios(world) # 执行三策略沙盒对比
|
||
after = json.dumps(world, ensure_ascii=False, sort_keys=True) # 运行后指纹
|
||
assert before == after, "方案对比污染了主干世界（违反 §5.1 Explore 铁律）" # 逐字节一致
|
||
assert len(block.props["cards"]) == 7 # 七张方案卡（含齐套/技能/换型/战役）
|
||
for card in block.props["cards"]: # 逐卡校验结构
|
||
assert card["kpi"]["woCount"] > 0 # 沙盒确实完成了排产
|
||
assert card["strategy"] in (
|
||
"DELIVERY_FIRST", "KITTING_FIRST", "SKILL_FIRST", "CAPACITY_BALANCE",
|
||
"CHANGEOVER_MIN", "CAMPAIGN", "COMPREHENSIVE",
|
||
) # 策略合法
|
||
assert "totalChangeoverMin" in card["kpi"]
|
||
assert "沙盒" in text # 文案明示未影响当前方案
|
||
|
||
|
||
# ---------------- S2：Checkpoint 往返 ----------------
|
||
def test_checkpoint_roundtrip(tmp_path):
|
||
"""建档 → 破坏世界 → 从档恢复：恢复结果与建档时刻逐字节一致。"""
|
||
store = CheckpointStore(path=str(tmp_path / "ckpt.json")) # 临时路径快照仓（隔离生产文件）
|
||
world = seed_world() # 初始世界
|
||
baseline = copy.deepcopy(world) # 建档时刻的对照副本
|
||
meta = store.create(world, label="测试基线", reason="manual") # 建档
|
||
assert meta["pairId"] # 拿到快照 ID
|
||
world["salesOrders"].clear() # 破坏：清空订单
|
||
world["scheduleVersions"].append({"id": 999}) # 破坏：塞入脏版本
|
||
pair = store.get(meta["pairId"]) # 取回快照
|
||
assert pair is not None # 快照存在
|
||
restored = pair["world"] # 快照中的世界
|
||
assert json.dumps(restored, ensure_ascii=False, sort_keys=True) == \
|
||
json.dumps(baseline, ensure_ascii=False, sort_keys=True), "快照未完整保留建档时刻状态" # 逐字节恢复
|
||
# 快照与运行态隔离：破坏运行态不应影响已存快照（深拷贝验证）
|
||
assert len(restored["salesOrders"]) == 7 # 快照内订单完好
|
||
|
||
|
||
# ---------------- S2b：容量淘汰 ----------------
|
||
def test_checkpoint_capacity_eviction(tmp_path):
|
||
"""快照数超过上限时淘汰最旧（防文件膨胀）。"""
|
||
store = CheckpointStore(path=str(tmp_path / "ckpt.json")) # 临时快照仓
|
||
world = seed_world() # 世界
|
||
ids = [store.create(world, label=f"c{i}", reason="manual")["pairId"] for i in range(25)] # 建 25 档（上限 20）
|
||
assert len(store.pairs) == 20 # 只保留 20 个
|
||
assert store.get(ids[0]) is None # 最旧的已被淘汰
|
||
assert store.get(ids[-1]) is not None # 最新的仍在
|
||
|
||
|
||
# ---------------- S3：审计哈希链 ----------------
|
||
def test_audit_chain_verify_and_tamper_detection():
|
||
"""正常链校验通过；篡改任一历史事件的任一字段必须被检出。"""
|
||
world = seed_world() # 世界（审计表随种子创建）
|
||
nid = _next_id_factory() # 发号器
|
||
for i in range(5): # 连续写 5 条审计
|
||
write_audit(world, nid, actor="tester", category="TOOL", action=f"op{i}",
|
||
target={"type": "T", "id": i}, power="P0", rationale={})
|
||
events = world["auditEvents"] # 审计链
|
||
assert verify_audit_chain(events)["ok"] is True # 完整链校验通过
|
||
tampered = copy.deepcopy(events) # 拷贝后篡改
|
||
tampered[2]["actor"] = "hacker" # 篡改第 3 条的行为人
|
||
result = verify_audit_chain(tampered) # 重新校验
|
||
assert result["ok"] is False # 必须检出断链
|
||
assert result["checked"] == 2 # 断点定位在被篡改处
|
||
|
||
|
||
# ---------------- S2c：成对快照（对话侧 + 世界侧）与成对回滚 ----------------
|
||
def test_checkpoint_pair_roundtrip_world_and_conversation(tmp_path):
|
||
"""矩阵 52 行：快照包含完整对话（workspace+messages）和世界；
|
||
破坏两侧后回滚，两侧均逐字段一致。"""
|
||
store = CheckpointStore(path=str(tmp_path / "ckpt.json"))
|
||
world = seed_world()
|
||
world_baseline = json.dumps(world, ensure_ascii=False, sort_keys=True)
|
||
conversation_side = {
|
||
"workspace": {
|
||
"projects": [{"id": "p1", "name": "项目A"}],
|
||
"sessions": [{"id": "s1", "projectId": "p1", "title": "话题1", "status": "running"}],
|
||
"files": [],
|
||
"messages": {"s1": [{"role": "user", "content": "试排一版"}]},
|
||
"activeProjectId": "p1",
|
||
"activeSessionId": "s1",
|
||
"worldKey": "default",
|
||
},
|
||
"plans": {"plan_x": [{"version": 1, "layer": "L0", "status": "DRAFT"}]},
|
||
}
|
||
side_baseline = json.dumps(conversation_side, ensure_ascii=False, sort_keys=True)
|
||
|
||
meta = store.create(world, label="成对基线", reason="manual",
|
||
conversation_note="对话摘要",
|
||
conversation_side=conversation_side)
|
||
assert meta["pairId"]
|
||
|
||
# 破坏两侧
|
||
world["salesOrders"].clear()
|
||
conversation_side["workspace"]["messages"]["s1"].append({"role": "assistant", "content": "被污染"})
|
||
conversation_side["plans"]["plan_x"].append({"version": 2, "layer": "L0", "status": "DRAFT"})
|
||
|
||
pair = store.get(meta["pairId"])
|
||
assert pair is not None
|
||
# 世界侧完整恢复
|
||
assert json.dumps(pair["world"], ensure_ascii=False, sort_keys=True) == world_baseline
|
||
# 对话侧完整恢复
|
||
assert json.dumps(pair["conversationSide"], ensure_ascii=False, sort_keys=True) == side_baseline
|
||
assert pair["conversationSide"]["workspace"]["messages"]["s1"] == [{"role": "user", "content": "试排一版"}]
|
||
# 快照与运行态隔离（后续破坏不影响已存快照）
|
||
world["salesOrders"].append({"id": "broken"})
|
||
assert len(pair["world"]["salesOrders"]) == 7
|
||
|
||
|
||
def test_checkpoint_pair_without_conversation_side_is_backward_compatible(tmp_path):
|
||
"""不传 conversation_side 时保持旧行为（仅 world + note），不报错。"""
|
||
store = CheckpointStore(path=str(tmp_path / "ckpt.json"))
|
||
world = seed_world()
|
||
meta = store.create(world, label="旧式基线", reason="manual", conversation_note="只有摘要")
|
||
pair = store.get(meta["pairId"])
|
||
assert pair is not None
|
||
assert pair["conversationNote"] == "只有摘要"
|
||
assert pair.get("conversationSide") is None
|
||
assert json.dumps(pair["world"], ensure_ascii=False, sort_keys=True) == json.dumps(world, ensure_ascii=False, sort_keys=True)
|