muse-agent-example/tests/集成/test_通用选段改写.py
zizi d909d1bd1b 后端实现与用例身份:19 包集成落地并修复收尾缺陷
实现侧:
- 上下文:任务范围拆分为 范围校验/范围授权;索引按可发现口径重建、索引新鲜度改对称差;依赖校验统一快照漂移说明。
- 知识方法:方法与材料读取口径统一;超限方法材料按可选省略,核对路径不再二次计费;删除无合同的读时重算。
- 任务运行:新增 context.usage/tool.denied 事件类型;连接池常驻并在装配生命周期内开关;调用结算与核对分列。
- 效果评测/审校修订/交付连载/作者经验/作品规划:凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。
- 资源加载:能力正文不再夹带索引用的导航注记(该注记此前进入角色与技能的模型提示)。
- 元数据:受保护骨架与代码保护属性对齐;字段校验与内置结构口径同步。
- 基础设施:环境预检进入装配生命周期;数据库连接运行期字段不参与相等比较;索引指纹归一化 jsonb 浮点。
- 删除被替代实现:7 份旧提示词模板与空壳 资料来源 读取器。

用例侧:
- 用例身份与导航元信息迁移;夹具补生命周期、同库暴露与模板封存;
- 本轮定向修复:方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
2026-09-18 01:15:00 +08:00

268 lines
10 KiB
Python

"""四类作者选段目的共享真实修订链,保持范围外并不伪造诊断。"""
from dataclasses import asdict, replace
import pytest
import test_模型修订任务 as 修订测试
from muse.共享.错误 import Muse错误
from muse.启动 import 装配审校修订
from muse.正文写作.接口 import 原文选段, 换行节点, 文本节点, 正文草稿, 段落, 段落哈希, 选段写作授权
from muse.编排.审校修订 import 发起模型修订
pytestmark = pytest.mark.数据库
生成环境 = 修订测试.生成环境
@pytest.fixture
def 选段环境(生成环境):
from pathlib import Path
env = 生成环境
装配审校修订(
env["装配"],
修订测试.写手测试.合成计价(),
Path(__file__).parents[2] / "配置/流程模板/审校修订.yaml",
)
source = 正文草稿(
(
段落("p1", (文本节点("雨声很轻。"),)),
段落("blank", ()),
段落(
"p2",
(
文本节点("雨声很轻。", ("emphasis",)),
换行节点("\r\n"),
文本节点("🙂e\u0301她只剩3枚铜钱。"),
),
),
)
)
env["正文"].保存人工(env["作者"], "selection-seed", "ch-4", 0, source)
saved = env["正文"].读取正文(env["作者"], "ch-4")
selection = 原文选段("p2", 段落哈希(source.paragraphs[2]), 0, 5, "雨声很轻。")
protection = 原文选段("p1", 段落哈希(source.paragraphs[0]), 0, 5, "雨声很轻。")
auth = 选段写作授权(
"gen-work",
"ch-4",
"main",
1,
saved["document_hash"],
"polish",
"只处理第二处雨声描写",
(selection,),
(protection,),
)
env.update(授权=auth, 原稿=saved, 源文档=source, 选段=selection)
return env
@pytest.mark.case_id(
"NC-w23-235001",
environment="隔离PG与合成HTTP",
given="带样式、重复句及空白的已保存原文",
when="按四类目的改写第二处原句",
then=["只在选段重放,范围外及保护区保持,没有诊断记录"],
contract="docs/系统架构/新版设计/模块设计/B05-正文写作.md",
)
@pytest.mark.parametrize(
"mode,text",
[
("expand", "雨声很轻,屋檐落下一滴水。"),
("compress", "雨声轻。"),
("polish", "细雨轻响。"),
("rewrite", "雨轻轻敲着屋瓦。"),
],
)
def test_四类改写只重放确切选段并保留空白样式与保护区__235001(选段环境, mode, text):
from pydantic import TypeAdapter
env = 选段环境
env["授权"] = replace(env["授权"], mode=mode)
tid = 修订测试.发起(env)
env["剧本"].append(
{
"类型": "文本",
"文本": {
"edits": [
{
"selection_id": env["选段"].selection_id,
"action": "replace",
"replacement": text,
"reason": "合成选段修改",
}
]
},
}
)
修订测试.跑(env, tid)
result = env["装配"].要求审校().读取模型修订(env["作者"], tid)
assert result["source_kind"] == "selection" and result["diagnosis_id"] is None
candidate = env["正文"].读取候选(env["作者"], result["candidate_id"])
doc = TypeAdapter(正文草稿).validate_python(candidate["document"])
assert doc.paragraphs[:2] == env["源文档"].paragraphs[:2]
assert doc.paragraphs[2].content[0] == 文本节点(text, ("emphasis",))
assert doc.paragraphs[2].content[1:] == env["源文档"].paragraphs[2].content[1:]
assert len(env["收到"]) == 1 and "selection_id" in str(env["收到"][0])
assert env["正文"].读取正文(env["作者"], "ch-4") == env["原稿"]
with env["库"].连接(只读=True) as conn:
assert conn.execute("SELECT count(*) FROM muse_diagnosis").fetchone()[0] == 0
@pytest.mark.case_id(
"NC-w23-235002",
environment="隔离PG与合成HTTP",
given="明确圈选与保护范围",
when="尝试错哈希、错原文、重叠或切开CRLF",
then=["发起前拒绝,不留任务或发送"],
contract="docs/系统架构/新版设计/模块设计/B05-正文写作.md",
)
@pytest.mark.parametrize("bad", ["protected", "hash", "text", "overlap", "crlf"])
def test_失效或受保护选段发起前拒绝且不留任务__235002(选段环境, bad):
env = 选段环境
auth, s = env["授权"], env["选段"]
if bad == "protected":
auth = replace(auth, protected_ranges=(s,))
elif bad == "hash":
auth = replace(auth, document_hash="0" * 64)
elif bad == "text":
auth = replace(auth, selections=(replace(s, original="雨声巨大。"),))
elif bad == "overlap":
auth = replace(auth, selections=(s, replace(s, start=1, original="声很轻。")))
else:
auth = replace(auth, selections=(replace(s, start=5, end=6, original="\r"),))
with pytest.raises(Muse错误):
发起模型修订(env["装配"], env["作者"], "bad-selection", auth, 配置ID="gen-config")
assert env["装配"].任务运行.读取命令任务(env["作者"].作者, "bad-selection") is None
assert not env["收到"]
@pytest.mark.case_id(
"NC-w23-235003",
environment="隔离PG与合成HTTP",
given="合法作者圈选任务",
when="模型保留原文或引入新数值",
then=["分别原文胜出或保真拒绝,无候选"],
contract="docs/系统架构/新版设计/模块设计/B05-正文写作.md",
)
@pytest.mark.parametrize(
"action,replacement,outcome",
[
("retain", "", "original"),
("replace", "她有4枚铜钱。", "rejected"),
],
)
def test_原文胜出与保真失败不生成候选__235003(选段环境, action, replacement, outcome):
env = 选段环境
tid = 修订测试.发起(env)
env["剧本"].append(
{
"类型": "文本",
"文本": {
"edits": [
{
"selection_id": env["选段"].selection_id,
"action": action,
"replacement": replacement,
"reason": "合成理由",
}
]
},
}
)
修订测试.跑(env, tid)
result = env["装配"].要求审校().读取模型修订(env["作者"], tid)
assert result["outcome"] == outcome and result["candidate_id"] is None
assert env["正文"].列出候选(env["作者"], "ch-4") == []
@pytest.mark.case_id(
"NC-w23-235004",
environment="隔离PG与合成HTTP",
given="旧整章生成接口",
when="无确切圈选请求扩写或选段修改",
then=["明确拒绝,转由带授权的选段链承接"],
contract="docs/系统架构/新版设计/模块设计/B05-正文写作.md",
)
def test_无圈选的旧整章入口不能冒充选段改写__235004(选段环境):
from muse.编排.生成正文 import 发起生成正文
env = 选段环境
for granularity in ("expand_scene", "edit_selection"):
with pytest.raises(Muse错误, match="圈选"):
发起生成正文(
env["装配"],
env["作者"],
granularity,
granularity,
work_id="gen-work",
chapter_id="ch-4",
配置ID="gen-config",
)
assert env["装配"].任务运行.读取命令任务(env["作者"].作者, granularity) is None
assert not env["收到"]
@pytest.mark.case_id(
"NC-w23-235005",
environment="隔离PG与合成HTTP",
given="同一作者与真实圈选坐标",
when="CLI定位并以HTTP和CLI重放生成请求",
then=["定位一致、同一任务,没有隐式模型调用"],
contract="docs/系统架构/新版设计/模块设计/B05-正文写作.md",
)
def test_CLI定位与HTTP生成共用同一真实选段授权__235005(选段环境, tmp_path):
import json
from fastapi.testclient import TestClient
from muse.接入.cli.审校命令 import 运行审校命令
from muse.接入.http.应用 import 创建应用
from muse.配置 import 服务配置
env = 选段环境
secret = tmp_path / "password"
secret.write_text("synthetic-selection-only")
secret.chmod(0o600)
config = replace(
env["装配"].配置,
HTTP=服务配置(
str(secret),
作者ID=env["作者"].作者,
公开地址="http://testserver",
允许来源=("http://testserver",),
),
)
env["装配"].配置 = config
auth = asdict(env["授权"])
target = {
k: auth[k] for k in ("work_id", "chapter_id", "branch_id", "revision", "document_hash")
}
payload = {
"target": target,
"selections": [
{k: v for k, v in s.items() if k != "paragraph_hash"} for s in auth["selections"]
],
}
path = tmp_path / "scope.json"
path.write_text(json.dumps(payload, ensure_ascii=False))
location = 运行审校命令(env["装配"], "选段定位", str(path))
assert location == list(auth["selections"])
http_app = 创建应用(config)
request = {"command_id": "shared-selection", "config_id": "gen-config", "request": auth}
with TestClient(http_app, headers={"origin": "http://testserver"}) as client:
http_app.state.装配 = env["装配"]
assert (
client.post("/api/v1/session", json={"password": secret.read_text()}).status_code == 200
)
assert client.post("/api/v1/selection-scope", json=payload).json() == location
response = client.post("/api/v1/revision-tasks", json=request)
assert response.status_code == 201, response.text
path.write_text(json.dumps(request, ensure_ascii=False))
assert 运行审校命令(env["装配"], "模型修订", str(path)) == response.json()
assert (
env["装配"].任务运行.读取任务(response.json()["task_id"]).冻结输入["输入"]["source_kind"]
== "selection"
)
assert not env["收到"]