74 lines
3.3 KiB
Python
74 lines
3.3 KiB
Python
# ============================================================
|
||
# Plan 批准/驳回审计黄金测试（plan.md §3.1 / 矩阵 110 行剩余项）
|
||
# 覆盖：DRAFT->APPROVED 批准（追加版本、regenCount 不变、证据引用）、
|
||
# DRAFT->FAILED 驳回、非法流转拒绝、历史不可变。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from server.agent_core.plan_runtime import (
|
||
PlanStore,
|
||
PlanTransitionError,
|
||
)
|
||
|
||
|
||
def _store(tmp_path: Path) -> PlanStore:
|
||
return PlanStore(str(tmp_path / "plans.json"))
|
||
|
||
|
||
def test_approve_appends_status_version(tmp_path: Path):
|
||
"""批准：DRAFT->APPROVED 追加新版本，regenCount 不变，证据引用记录 status。"""
|
||
store = _store(tmp_path)
|
||
plan = store.create(plan_id="p-approve", layer="L0", parent_id=None,
|
||
inputs={"prompt": "x"}, payload={"goal": "g"})
|
||
assert plan.status == "DRAFT" and plan.version == 1 and plan.regenCount == 0
|
||
|
||
approved = store.transition_status("p-approve", new_status="APPROVED",
|
||
created_by="USER", note="计划员批准")
|
||
assert approved.status == "APPROVED"
|
||
assert approved.version == 2
|
||
assert approved.regenCount == 0 # 状态流转不消耗重生额度
|
||
assert approved.evidenceRefs[-1] == "status:APPROVED"
|
||
# 历史不可变：v1 仍为 DRAFT
|
||
assert store.node_at("p-approve", 1).status == "DRAFT"
|
||
|
||
|
||
def test_reject_appends_failed_version(tmp_path: Path):
|
||
"""驳回：DRAFT->FAILED 追加新版本，携带 note 入 payload。"""
|
||
store = _store(tmp_path)
|
||
store.create(plan_id="p-reject", layer="L0", parent_id=None,
|
||
inputs={"prompt": "x"}, payload={"goal": "g"})
|
||
rejected = store.transition_status("p-reject", new_status="FAILED",
|
||
created_by="USER", note="驳回：数据不足")
|
||
assert rejected.status == "FAILED"
|
||
assert rejected.version == 2
|
||
assert rejected.payload["_note"][-1]["status"] == "FAILED"
|
||
assert rejected.payload["_note"][-1]["note"] == "驳回：数据不足"
|
||
|
||
|
||
def test_illegal_transition_rejected(tmp_path: Path):
|
||
"""非法流转：APPROVED->DRAFT 必须拒绝（PlanTransitionError）。"""
|
||
store = _store(tmp_path)
|
||
store.create(plan_id="p-illegal", layer="L0", parent_id=None,
|
||
inputs={"prompt": "x"}, payload={"goal": "g"})
|
||
store.transition_status("p-illegal", new_status="APPROVED")
|
||
with pytest.raises(PlanTransitionError):
|
||
store.transition_status("p-illegal", new_status="DRAFT")
|
||
|
||
|
||
def test_transition_chain_approve_then_done(tmp_path: Path):
|
||
"""连续流转：DRAFT->APPROVED->DONE 合法，每步追加版本且历史不变量保持。"""
|
||
store = _store(tmp_path)
|
||
store.create(plan_id="p-chain", layer="L0", parent_id=None,
|
||
inputs={"prompt": "x"}, payload={"goal": "g"})
|
||
store.transition_status("p-chain", new_status="APPROVED")
|
||
done = store.transition_status("p-chain", new_status="DONE")
|
||
assert done.status == "DONE"
|
||
assert done.version == 3
|
||
versions = store.versions("p-chain")
|
||
assert [v.status for v in versions] == ["DRAFT", "APPROVED", "DONE"]
|
||
assert [v.regenCount for v in versions] == [0, 0, 0]
|