d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
219 lines
9.0 KiB
Python
219 lines
9.0 KiB
Python
"""真实S01定稿、S02导出和不可变产物；不使用私人正文。"""
|
||
|
||
from dataclasses import replace
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
import test_正文候选与作者决定 as 正文测试
|
||
|
||
from muse.交付连载.接口 import 交付材料, 定稿请求, 章节版本
|
||
from muse.任务运行.接口 import 任务状态
|
||
from muse.共享.调用身份 import 用途
|
||
from muse.共享.错误 import Muse错误
|
||
from muse.启动 import 构建
|
||
from muse.编排.定稿交付 import 发起定稿导出
|
||
from muse.配置 import 应用配置
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
候选环境 = 正文测试.候选环境
|
||
|
||
|
||
@pytest.fixture
|
||
def 交付环境(候选环境):
|
||
body, author, pools = 候选环境
|
||
app = 构建(应用配置(数据库=pools[用途.生产].引用, 资源发布身份="test"))
|
||
with app.生命周期():
|
||
app.要求作品().添加章节(
|
||
author,
|
||
"second",
|
||
"work-candidate",
|
||
"chapter-second",
|
||
"第二章 雨停",
|
||
预期目录版本=1,
|
||
)
|
||
body.保存人工(
|
||
author,
|
||
"second-body",
|
||
"chapter-second",
|
||
0,
|
||
正文测试.草稿("第二章原文🙂", "保留<&>与组合字符e\u0301。"),
|
||
)
|
||
chapters = tuple(
|
||
章节版本(cid, 1, body.读取正文(author, cid)["document_hash"])
|
||
for cid in ("chapter-candidate", "chapter-second")
|
||
)
|
||
req = 定稿请求(
|
||
str(uuid4()),
|
||
"work-candidate",
|
||
2,
|
||
chapters,
|
||
交付材料("雨中旧事", "合成作者", "用于验证的简介", "梗概", "人物表"),
|
||
)
|
||
yield app, author, pools, req
|
||
|
||
|
||
def 跑导出(app, tid):
|
||
tasks = app.任务运行
|
||
while claim := tasks.领取步骤(
|
||
"export-worker",
|
||
["delivery.verify", "delivery.export"],
|
||
任务ID=tid,
|
||
作者=tasks.读取任务(tid).作者,
|
||
):
|
||
tasks.执行一步(claim)
|
||
return tasks.读取任务(tid)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w27-27a001",
|
||
environment="隔离PG",
|
||
given="本作者完整目录、确切正文版本与独立隔离数据",
|
||
when="通过定稿、S02导出、跨章候选或连载反馈实际入口执行并检查失败恢复",
|
||
then=["完整目录冻结后编辑不改变原定稿且重放唯一"],
|
||
contract="docs/系统架构/新版设计/模块设计/B08-交付连载.md",
|
||
)
|
||
def test_完整目录冻结后编辑不改变原定稿且重放唯一__27a001(交付环境):
|
||
app, author, pools, req = 交付环境
|
||
svc = app.要求交付()
|
||
with pytest.raises(Muse错误, match="完整目录"):
|
||
svc.冻结定稿(author, "missing", replace(req, chapters=req.chapters[:1]))
|
||
with pytest.raises(Muse错误, match="完整目录"):
|
||
svc.冻结定稿(author, "order", replace(req, chapters=req.chapters[::-1]))
|
||
result = svc.冻结定稿(author, "freeze", req)
|
||
app.要求正文().保存人工(
|
||
author,
|
||
"edit",
|
||
"chapter-candidate",
|
||
1,
|
||
正文测试.草稿("后来编辑"),
|
||
)
|
||
assert svc.冻结定稿(author, "freeze", req) == result
|
||
with pytest.raises(Muse错误):
|
||
svc.冻结定稿(author, "stale", replace(req, delivery_id=str(uuid4())))
|
||
read = svc.读取定稿(author, req.delivery_id)
|
||
assert "原句「保留」" in read["chapters"][0]["text"]
|
||
assert "后来编辑" not in read["chapters"][0]["text"]
|
||
with pools[用途.生产].连接(只读=True) as conn:
|
||
assert conn.execute("SELECT count(*) FROM muse_delivery").fetchone()[0] == 1
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w27-27a002",
|
||
environment="隔离PG",
|
||
given="本作者完整目录、确切正文版本与独立隔离数据",
|
||
when="通过定稿、S02导出、跨章候选或连载反馈实际入口执行并检查失败恢复",
|
||
then=["明确历史选择仍保护当前指针且不同作者拒绝"],
|
||
contract="docs/系统架构/新版设计/模块设计/B08-交付连载.md",
|
||
)
|
||
def test_明确历史选择仍保护当前指针且不同作者拒绝__27a002(交付环境):
|
||
app, author, _, req = 交付环境
|
||
app.要求正文().保存人工(
|
||
author,
|
||
"edit",
|
||
"chapter-candidate",
|
||
1,
|
||
正文测试.草稿("后来编辑"),
|
||
)
|
||
req = replace(
|
||
req,
|
||
chapters=(
|
||
replace(req.chapters[0], expected_current_revision=2),
|
||
req.chapters[1],
|
||
),
|
||
)
|
||
svc = app.要求交付()
|
||
svc.冻结定稿(author, "history", req)
|
||
assert "原句" in svc.读取定稿(author, req.delivery_id)["chapters"][0]["text"]
|
||
with pytest.raises(Muse错误):
|
||
svc.读取定稿(replace(author, 作者="stranger"), req.delivery_id)
|
||
with pytest.raises(Muse错误):
|
||
svc.冻结定稿(replace(author, 用途=用途.评测), "eval", req)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w27-27a003",
|
||
environment="隔离PG",
|
||
given="本作者完整目录、确切正文版本与独立隔离数据",
|
||
when="通过定稿、S02导出、跨章候选或连载反馈实际入口执行并检查失败恢复",
|
||
then=["真实任务四格式导出不修改正文且恢复不重发已存格式"],
|
||
contract="docs/系统架构/新版设计/模块设计/B08-交付连载.md",
|
||
)
|
||
def test_真实任务四格式导出不修改正文且恢复不重发已存格式__27a003(
|
||
交付环境,
|
||
monkeypatch,
|
||
):
|
||
from muse.交付连载 import 导出格式
|
||
|
||
app, author, pools, req = 交付环境
|
||
svc = app.要求交付()
|
||
svc.冻结定稿(author, "freeze", req)
|
||
formats = ["txt", "md", "docx", "epub"]
|
||
tid = 发起定稿导出(app, author, "export", req.delivery_id, formats)["task_id"]
|
||
assert 发起定稿导出(app, author, "export", req.delivery_id, formats)["task_id"] == tid
|
||
calls, real = [], 导出格式.导出
|
||
|
||
def fault(fmt, content):
|
||
calls.append(fmt)
|
||
if fmt == "docx":
|
||
raise RuntimeError("合成格式失败")
|
||
return real(fmt, content)
|
||
|
||
monkeypatch.setattr(导出格式, "导出", fault)
|
||
with pytest.raises(RuntimeError, match="合成格式失败"):
|
||
跑导出(app, tid)
|
||
first = app.任务运行.读取任务(tid)
|
||
assert first.状态.value == "failed"
|
||
assert calls == ["txt", "md", "docx"]
|
||
txt = svc.读取导出(author, req.delivery_id, "txt")
|
||
assert txt["verification"]["identities"] == "not_encoded"
|
||
assert txt["verification"]["chapter_ids"] is None
|
||
monkeypatch.setattr(导出格式, "导出", lambda fmt, c: (calls.append(fmt), real(fmt, c))[1])
|
||
app.任务运行.控制任务(tid, author.作者, 任务状态.已失败, "恢复", 命令ID="resume")
|
||
assert 跑导出(app, tid).状态.value == "completed"
|
||
assert calls == ["txt", "md", "docx", "docx", "epub"]
|
||
assert svc.读取导出(author, req.delivery_id, "txt")["content_hash"] == txt["content_hash"]
|
||
assert all(svc.读取导出(author, req.delivery_id, f)["data"] for f in formats)
|
||
with pools[用途.生产].连接(只读=True) as conn:
|
||
assert conn.execute("SELECT count(*) FROM muse_delivery_artifact").fetchone()[0] == 4
|
||
assert conn.execute("SELECT count(*) FROM muse_document_version").fetchone()[0] == 2
|
||
|
||
|
||
@pytest.mark.case_id("NC-O10-014")
|
||
def test_旧成品显式审计追加证明且原行不变(交付环境, monkeypatch):
|
||
from muse.交付连载 import 定稿版本
|
||
from muse.交付连载.存储 import 交付存储
|
||
from muse.交付连载.导出格式 import 导出, 校验, 格式版本
|
||
|
||
app, author, pools, req = 交付环境
|
||
svc = app.要求交付()
|
||
svc.冻结定稿(author, "freeze-audit", req)
|
||
content = svc.读取定稿(author, req.delivery_id)
|
||
data = 导出("txt", content)
|
||
legacy = 校验("txt", data, content)
|
||
with pools[用途.生产].连接() as conn, conn.transaction():
|
||
original = 交付存储(conn).写产物(
|
||
req.delivery_id, "txt", 格式版本, data, legacy["sha256"], legacy
|
||
)
|
||
with pytest.raises(Muse错误) as error:
|
||
svc.读取导出(author, req.delivery_id, "txt")
|
||
assert error.value.错误码 == "EXPORT_VERIFICATION_UPGRADE_REQUIRED"
|
||
with pytest.raises(Muse错误):
|
||
svc.审计导出(replace(author, 作者="stranger"), req.delivery_id, "txt", 补充验证=True)
|
||
receipt = svc.审计导出(author, req.delivery_id, "txt", 补充验证=True)
|
||
assert receipt["persisted"] is True
|
||
assert svc.审计导出(author, req.delivery_id, "txt", 补充验证=True) == receipt
|
||
|
||
def unavailable(*args, **kwargs):
|
||
raise AssertionError("下载不能读取原正文")
|
||
|
||
monkeypatch.setattr(定稿版本, "读取正文依据", unavailable)
|
||
downloaded = svc.读取导出(author, req.delivery_id, "txt")
|
||
assert downloaded["data"] == original["data"]
|
||
assert downloaded["verification"] == original["verification"]
|
||
assert downloaded["artifact_id"] == original["artifact_id"]
|
||
with pools[用途.生产].连接(只读=True) as conn:
|
||
assert (
|
||
conn.execute("SELECT count(*) FROM muse_delivery_artifact_validation").fetchone()[0]
|
||
== 1
|
||
)
|