246 lines
9.6 KiB
Python
246 lines
9.6 KiB
Python
"""合成流程驱动真实 S02、B01、B05 和隔离 PG；不调用模型、不证明完整创作链。"""
|
||
|
||
from dataclasses import asdict, replace
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
|
||
from muse.上下文.接口 import 上下文错误, 装配材料
|
||
from muse.任务运行.接口 import (
|
||
事件类型,
|
||
任务请求,
|
||
工具请求,
|
||
模型协议错误,
|
||
步骤处理器,
|
||
步骤结果,
|
||
步骤计划,
|
||
)
|
||
from muse.作品规划.接口 import 档案保存
|
||
from muse.元数据.接口 import 元数据服务, 导入内置结构
|
||
from muse.共享.调用身份 import 内容用途, 用途, 调用身份
|
||
from muse.启动 import 构建
|
||
from muse.正文写作.接口 import 换行节点, 文本节点, 正文草稿, 段落
|
||
from muse.编排.接口 import 流程定义, 流程服务, 流程登记
|
||
from muse.配置 import 应用配置
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
原文 = " 雨夜🌧，e\u0301。\n门还开着。"
|
||
前段, 后段 = 原文.split("\n")
|
||
|
||
|
||
def 原文草稿(章: str) -> 正文草稿:
|
||
"""段内换行走显式换行节点；可见文本仍与原文逐字一致。"""
|
||
return 正文草稿((段落("p-" + 章, (文本节点(前段), 换行节点(), 文本节点(后段))),))
|
||
|
||
|
||
@pytest.fixture
|
||
def 历史环境(应用测试库):
|
||
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
|
||
导入内置结构(元数据服务(连))
|
||
登记 = 流程登记()
|
||
装配 = 构建(应用配置(应用测试库[用途.生产].引用, "test"), 流程=登记)
|
||
作者 = 调用身份("history-author", None, 用途.生产, 内容用途.生成)
|
||
作品, 正文 = 装配.要求作品(), 装配.要求正文()
|
||
for work, chapters in [
|
||
("history-work", ("past", "other", "future")),
|
||
("another-work", ("foreign",)),
|
||
]:
|
||
结构 = 作品.新档案结构(作者, work, "work_core", 1)
|
||
作品.保存档案(
|
||
作者,
|
||
"create-" + work,
|
||
档案保存(work, 0, {"名称": work, "题材定位": "现实", "主题立意": "选择"}, 结构),
|
||
)
|
||
for i, chapter in enumerate(chapters):
|
||
作品.添加章节(作者, "add-" + chapter, work, chapter, chapter, 预期目录版本=i)
|
||
正文.保存人工(作者, "save-" + chapter, chapter, 0, 原文草稿(chapter))
|
||
|
||
def 执行(上下文):
|
||
工具 = 装配.要求只读工具(上下文.领取)
|
||
结果 = 工具.调用(
|
||
工具请求("history-read", "read_history_document", {"chapter_id": "past", "revision": 1})
|
||
)
|
||
材料 = {结果.来源[0].来源ID: 结果.内容["text"]}
|
||
上限 = 上下文.任务.冻结输入["冻结上下文"]["budget"]["context_bytes"]
|
||
return 步骤结果(asdict(装配材料(材料, tuple(材料), 字节上限=上限)))
|
||
|
||
登记.登记处理器(
|
||
步骤处理器(
|
||
"synthetic-history",
|
||
"1",
|
||
执行,
|
||
"history/v1",
|
||
"history/v1",
|
||
保护职责="范围核对",
|
||
允许工具=("read_history_document",),
|
||
)
|
||
)
|
||
登记.登记类型("synthetic-history", 必需保护=("范围核对",))
|
||
登记.登记恢复检查(
|
||
"synthetic-history", "1", lambda 任务: 装配.要求上下文().读取范围(任务.任务ID)
|
||
)
|
||
运行 = 装配.任务运行
|
||
assert 运行 is not None
|
||
流程服务(运行, 登记).发布(
|
||
流程定义(
|
||
"synthetic-history",
|
||
"1",
|
||
(步骤计划("read", "synthetic-history", "1", 工具=("read_history_document",)),),
|
||
)
|
||
)
|
||
|
||
def 开始(范围=None):
|
||
if 范围 is None:
|
||
范围 = 装配.要求上下文().授权历史范围(
|
||
作者, "history-work", 2, ({"chapter_id": "past", "revision": 1},)
|
||
)
|
||
id_ = 运行.创建任务(
|
||
任务请求(
|
||
"synthetic-history",
|
||
str(uuid4()),
|
||
作者.作者,
|
||
用途.生产,
|
||
内容用途.生成,
|
||
{},
|
||
"test",
|
||
"test",
|
||
{
|
||
"source_scope": 范围,
|
||
"schema_versions": {},
|
||
"authorization": "synthetic-author-choice",
|
||
"budget": {"context_bytes": 1024},
|
||
"stop_conditions": {"max_reads": 1},
|
||
},
|
||
作品ID="history-work",
|
||
),
|
||
"synthetic-history",
|
||
"1",
|
||
)
|
||
领取 = 运行.领取步骤("history-test", ["synthetic-history"], 租期秒=120)
|
||
assert 领取 is not None and 领取.任务ID == id_
|
||
return id_, 领取, 装配.要求只读工具(领取)
|
||
|
||
return 装配, 作者, 开始
|
||
|
||
|
||
def test_历史正文经真实任务读取留痕并进入材料预算__b09101(历史环境):
|
||
装配, _, 开始 = 历史环境
|
||
task_id, 领取, _ = 开始()
|
||
结果 = 装配.任务运行.执行一步(领取)
|
||
assert 结果.输出["文本"] == 原文
|
||
assert 结果.输出["字节数"] == len(原文.encode("utf-8"))
|
||
事件 = 装配.任务运行.续接事件(task_id).事件
|
||
读取 = [e for e in 事件 if e.类型 == 事件类型.工具读取]
|
||
assert len(读取) == 1
|
||
引用 = 读取[0].载荷["source_refs"][0]
|
||
assert 引用["revision"] == "1"
|
||
assert 引用["schema_hash"] and 引用["projection_version"]
|
||
assert 读取[0].载荷["tool"] == "read_history_document"
|
||
|
||
|
||
def test_正文工具按原文码点返回片段与用途__b09102(历史环境):
|
||
_, _, 开始 = 历史环境
|
||
_, _, 工具 = 开始()
|
||
结果 = 工具.调用(
|
||
工具请求(
|
||
"slice",
|
||
"read_history_document",
|
||
{"chapter_id": "past", "revision": 1, "start": 3, "end": 8},
|
||
)
|
||
)
|
||
assert 结果.内容["text"] == "🌧，e\u0301。"
|
||
assert (结果.内容["start"], 结果.内容["end"]) == (3, 8)
|
||
assert 结果.内容["run_purpose"] == "production"
|
||
assert 结果.内容["content_purpose"] == "generation"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"章节", ["foreign", "future", "other"], ids=["cross-work", "future", "unauthorized"]
|
||
)
|
||
def test_正文工具拒绝冻结范围以外的文稿__b09103(历史环境, 章节):
|
||
装配, _, 开始 = 历史环境
|
||
task_id, _, 工具 = 开始()
|
||
with pytest.raises(上下文错误) as 错误:
|
||
工具.调用(工具请求("denied", "read_history_document", {"chapter_id": 章节, "revision": 1}))
|
||
assert 错误.value.错误码 == "SCOPE_DENIED"
|
||
assert not [e for e in 装配.任务运行.续接事件(task_id).事件 if e.类型 == 事件类型.工具读取]
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"变化", ["document", "directory"], ids=["new-document", "reordered-directory"]
|
||
)
|
||
def test_正文或目录变化拒绝沿用原任务__b09104(历史环境, 变化):
|
||
装配, 作者, 开始 = 历史环境
|
||
_, _, 工具 = 开始()
|
||
if 变化 == "document":
|
||
装配.要求正文().保存人工(
|
||
作者, "revision-2", "past", 1, 正文草稿((段落("p-past", (文本节点("改过的正文"),)),))
|
||
)
|
||
else:
|
||
装配.要求作品().重排章节(
|
||
作者, "reorder", "history-work", ("future", "other", "past"), 预期目录版本=3
|
||
)
|
||
with pytest.raises(上下文错误) as 错误:
|
||
工具.调用(工具请求("stale", "read_history_document", {"chapter_id": "past", "revision": 1}))
|
||
assert 错误.value.错误码 == "SOURCE_STALE"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"参数",
|
||
[{"chapter_id": "past", "revision": 2}, {"chapter_id": "past", "revision": 1, "end": 1000}],
|
||
ids=["unapproved-version", "invalid-span"],
|
||
)
|
||
def test_未批准版本与越界片段不返回正文__b09105(历史环境, 参数):
|
||
_, _, 开始 = 历史环境
|
||
_, _, 工具 = 开始()
|
||
with pytest.raises(上下文错误) as 错误:
|
||
工具.调用(工具请求("invalid", "read_history_document", 参数))
|
||
assert 错误.value.错误码 in {"SCOPE_DENIED", "CONTEXT_INPUT_INVALID"}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"场景",
|
||
["unknown-tool", "work-id", "missing-cutoff", "wrong-author", "missing-version"],
|
||
ids=str,
|
||
)
|
||
def test_未登记工具扩权参数及不完整授权失败关闭__b09106(历史环境, 场景):
|
||
装配, 作者, 开始 = 历史环境
|
||
if 场景 == "wrong-author":
|
||
with pytest.raises(上下文错误) as 错误:
|
||
装配.要求上下文().授权历史范围(
|
||
replace(作者, 作者="another-author"),
|
||
"history-work",
|
||
2,
|
||
({"chapter_id": "past", "revision": 1},),
|
||
)
|
||
assert 错误.value.错误码 == "SCOPE_DENIED"
|
||
return
|
||
if 场景 == "missing-version":
|
||
with pytest.raises(上下文错误) as 错误:
|
||
装配.要求上下文().授权历史范围(
|
||
作者, "history-work", 2, ({"chapter_id": "past", "revision": 99},)
|
||
)
|
||
assert 错误.value.错误码 == "SOURCE_NOT_FOUND"
|
||
return
|
||
范围 = 装配.要求上下文().授权历史范围(
|
||
作者, "history-work", 2, ({"chapter_id": "past", "revision": 1},)
|
||
)
|
||
if 场景 == "missing-cutoff":
|
||
del 范围["as_of"]
|
||
with pytest.raises(上下文错误) as 错误:
|
||
开始(范围)
|
||
assert 错误.value.错误码 == "CONTEXT_SCOPE_INVALID"
|
||
return
|
||
_, _, 工具 = 开始(范围)
|
||
参数 = {"chapter_id": "past", "revision": 1}
|
||
if 场景 == "work-id":
|
||
参数["work_id"] = "another-work"
|
||
with pytest.raises(模型协议错误):
|
||
工具.调用(
|
||
工具请求(
|
||
"no-tool",
|
||
"read_unknown" if 场景 == "unknown-tool" else "read_history_document",
|
||
参数,
|
||
)
|
||
)
|