muse-agent-example/tests/单元/test_上下文任务范围形状.py

139 lines
4.6 KiB
Python

"""任务范围新节(目标章细纲、事实系统版本与用途绑定)的离线形状校验。
只测恢复冻结范围对冻结输入的判定:形状与身份不一致必须失败关闭,
不连数据库、不调用模型。
"""
import pytest
from muse.上下文.接口 import 上下文服务, 上下文错误
from muse.任务运行.模型 import 任务快照, 任务状态, 执行计划
from muse.共享.调用身份 import 用途
from muse.基础设施.数据库.连接 import 数据库工厂
from muse.配置 import 数据库引用
def 服务() -> 上下文服务:
工厂 = 数据库工厂(数据库引用("环境变量", "MUSE_TEST_DATABASE_URL"), 用途.生产)
return 上下文服务(工厂, None)
def 快照(范围: dict) -> 任务快照:
return 任务快照(
"task-shape",
任务状态.待运行,
"ctx-author",
"hash",
{
"作品ID": 范围["work_id"],
"执行用途": "production",
"内容用途": "generation",
"冻结上下文": {"source_scope": 范围},
},
执行计划("shape-flow", "1", ()),
(),
0,
)
def 文稿(编号: str) -> dict:
return {
"chapter_id": 编号,
"branch_id": "main",
"revision": 1,
"document_id": "doc-" + 编号,
"document_hash": "hash-" + 编号,
"position": 1,
}
def 任务范围() -> dict:
"""含历史文稿、目标章细纲与事实系统版本的完整任务范围。"""
return {
"work_id": "ctx-work",
"as_of": 2,
"source_ids": ["doc-ch-1", "plan-target", "obj-fact"],
"documents": [文稿("ch-1")],
"directory_revision": 3,
"authorized_by": "ctx-author",
"run_purpose": "production",
"content_purpose": "generation",
"target_plan": {
"chapter_id": "ch-3",
"plan_id": "plan-target",
"revision": 1,
"content_hash": "plan-hash",
},
"facts": {"system_revision": 2},
}
def test_完整任务范围原样恢复__d1a2f3() -> None:
范围 = 任务范围()
assert 服务().恢复冻结范围(快照(范围)) == 范围
def test_无历史文稿但绑定细纲与事实仍可恢复__e4b5c6() -> None:
范围 = {**任务范围(), "documents": [], "source_ids": ["plan-target", "obj-fact"]}
assert 服务().恢复冻结范围(快照(范围)) == 范围
@pytest.mark.parametrize(
"改动",
[
{"target_plan": {"chapter_id": "ch-3", "plan_id": "plan-target", "revision": 0}},
{"target_plan": {"chapter_id": "ch-3", "plan_id": "plan-target", "revision": True}},
{"target_plan": {"chapter_id": "", "plan_id": "plan-target", "revision": 1}},
{"facts": {"system_revision": -1}},
{"facts": {"system_revision": True}},
{"facts": {}},
{"documents": [], "target_plan": None, "facts": None},
],
ids=[
"zero-plan-revision",
"boolean-plan-revision",
"empty-chapter",
"negative-system-revision",
"boolean-system-revision",
"missing-system-revision",
"empty-scope",
],
)
def test_非法细纲事实形状或空范围拒绝__f7a8b9(改动: dict) -> None:
with pytest.raises(上下文错误) as 异常:
服务().恢复冻结范围(快照({**任务范围(), **改动}))
assert 异常.value.错误码 == "CONTEXT_SCOPE_INVALID"
def test_细纲身份不在来源清单拒绝__a3c4d5() -> None:
范围 = {**任务范围(), "source_ids": ["doc-ch-1", "obj-fact"]}
with pytest.raises(上下文错误) as 异常:
服务().恢复冻结范围(快照(范围))
assert 异常.value.错误码 == "CONTEXT_SCOPE_INVALID"
@pytest.mark.parametrize(
"改动",
[
{"run_purpose": None},
{"content_purpose": "planning"},
{"run_purpose": "production", "content_purpose": "extraction"},
],
ids=["missing-run-purpose", "content-mismatch", "content-unbound"],
)
def test_新节必须绑定与任务一致的用途__b6c7d8(改动: dict) -> None:
范围 = {**任务范围(), **改动}
if 改动.get("run_purpose") is None:
del 范围["run_purpose"]
with pytest.raises(上下文错误) as 异常:
服务().恢复冻结范围(快照(范围))
assert 异常.value.错误码 == "CONTEXT_SCOPE_INVALID"
def test_仅历史文稿的范围不要求用途键__c9d0e1() -> None:
范围 = 任务范围()
for 键 in ("run_purpose", "content_purpose", "target_plan", "facts"):
del 范围[键]
范围["source_ids"] = ["doc-ch-1"]
assert 服务().恢复冻结范围(快照(范围)) == 范围