1408 lines
59 KiB
Python
1408 lines
59 KiB
Python
"""声音从实际样张到明确确认；报告和案例沿各自公共入口验证。"""
|
||
|
||
import hashlib
|
||
import itertools
|
||
import json
|
||
from copy import deepcopy
|
||
from dataclasses import replace
|
||
|
||
import pytest
|
||
from test_两阶段写手 import 生成环境 as _生成环境夹具
|
||
from test_完整文学审校 import 发起 as 发起全文审校
|
||
from test_完整文学审校 import 文学环境 as _文学环境夹具
|
||
from test_完整文学审校 import 片结果 as 合成片结果
|
||
from test_完整文学审校 import 跑 as 跑全文审校
|
||
|
||
from muse.作品规划.接口 import 档案保存
|
||
from muse.共享.错误 import Muse错误
|
||
from muse.审校修订.接口 import (
|
||
声音定位,
|
||
声音提案请求,
|
||
声音样张引用,
|
||
审校错误,
|
||
案例复核请求,
|
||
案例定位,
|
||
案例来源,
|
||
案例规则请求,
|
||
案例采集请求,
|
||
读取内置规则,
|
||
)
|
||
from muse.正式变更.接口 import 固定哈希
|
||
from muse.正文写作.接口 import 文本节点, 正文草稿, 段落
|
||
from muse.资料研究.接口 import 导入请求
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
|
||
|
||
@pytest.fixture
|
||
def 全文生成环境(内置结构测试库, tmp_path):
|
||
# 生成环境正文直接建档案结构，必须复用已导入内置结构的种子库。
|
||
yield from _生成环境夹具.__wrapped__(内置结构测试库, tmp_path)
|
||
|
||
|
||
@pytest.fixture
|
||
def 全文报告环境(全文生成环境):
|
||
return _文学环境夹具.__wrapped__(全文生成环境)
|
||
|
||
|
||
@pytest.fixture
|
||
def 声音环境(章后环境):
|
||
env = 章后环境
|
||
装配, 身份 = env["装配"], env["作者"]
|
||
作品 = 装配.要求作品()
|
||
结构 = 作品.新档案结构(身份, "voice-work", "work_core", 1)
|
||
作品.保存档案(
|
||
身份, "voice-work-create", 档案保存("voice-work", 0, {"名称": "合成声音作品"}, 结构)
|
||
)
|
||
作品.添加章节(身份, "voice-ch-create", "voice-work", "voice-ch", "样张", 预期目录版本=0)
|
||
装配.要求正文().保存人工(
|
||
身份,
|
||
"voice-text",
|
||
"voice-ch",
|
||
0,
|
||
正文草稿((段落("v1", (文本节点("甲走进门。雨落下来。他没有再说话。"),)),)),
|
||
)
|
||
文 = 装配.要求正文().读取正文(身份, "voice-ch")
|
||
请求 = 声音提案请求(
|
||
"voice-work",
|
||
0,
|
||
"style",
|
||
1,
|
||
{"句式画像": "短句叙述", "达标样张": "甲走进门。"},
|
||
(声音样张引用("document", "voice-ch", 文["revision"], 文["document_hash"], "main"),),
|
||
verbal_tics=("没有再说话",),
|
||
protected_expressions=("雨落下来。",),
|
||
)
|
||
return 装配, 身份, 请求, 文
|
||
|
||
|
||
def _提出(服务, 身份, 命令, 请求):
|
||
回执 = 服务.提出声音(身份, 命令, 请求)
|
||
r = 回执["results"][0]
|
||
loc = 声音定位(
|
||
请求.work_id, r["proposal_id"], 1, r["proposal_hash"], 请求.expected_voice_revision
|
||
)
|
||
return 回执, loc
|
||
|
||
|
||
def _确认(服务, 身份, 命令, loc):
|
||
审阅 = 服务.打开声音审阅(身份, loc)
|
||
return 服务.决定声音(身份, 命令, loc, 审阅ID=审阅["review_id"], 审阅哈希=审阅["review_hash"])
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w22-22c001",
|
||
environment="隔离PG与合成样张",
|
||
given="本书真实正文样张、声音未决提案和有效/无效审阅",
|
||
when="经真实S04/S01/B05/B03公共用例保存和确认声音",
|
||
then=["候选无当前投影；真实S01确认后才可读取声音，原文不变"],
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音候选不当确认且真实审阅后才有当前投影__22c001(声音环境):
|
||
装配, 身份, 请求, 文 = 声音环境
|
||
服务 = 装配.要求审校()
|
||
assert 服务.当前声音(身份, 请求.work_id) is None
|
||
回执, loc = _提出(服务, 身份, "voice-proposal", 请求)
|
||
assert 回执 == 服务.提出声音(身份, "voice-proposal", 请求)
|
||
assert 服务.当前声音(身份, 请求.work_id) is None
|
||
for 假审阅 in ("missing", "", "00000000-0000-4000-8000-000000000000"):
|
||
with pytest.raises(Muse错误):
|
||
服务.决定声音(身份, "bad-review", loc, 审阅ID=假审阅, 审阅哈希="0" * 64)
|
||
assert 服务.当前声音(身份, 请求.work_id) is None
|
||
结果 = _确认(服务, 身份, "voice-confirm", loc)
|
||
assert 结果["action"] == "adopt" and 结果["results"][0]["revision"] == 1
|
||
声音 = 服务.当前声音(身份, 请求.work_id, 字段用途="aiContext")
|
||
assert 声音["narrator"] == 请求.narrator
|
||
assert 声音["protected_expressions"] == ["雨落下来。"]
|
||
assert 声音["revision"] == 1
|
||
assert 装配.要求正文().读取正文(身份, "voice-ch") == 文
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w22-22c002",
|
||
environment="隔离PG与合成样张",
|
||
given="无依据口癖/保护/样句、错哈希、第二人物声音或缺叙述者",
|
||
when="经真实S04/S01/B05/B03公共用例保存和确认声音",
|
||
then=["拒绝不完整/未grounding提案，不生成声音版本"],
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
@pytest.mark.parametrize("坏处", ["口癖", "保护", "样句", "哈希", "人物资产", "叙述者"])
|
||
def test_声音缺依据或重复人物归属不能创建提案__22c002(声音环境, 坏处):
|
||
from pydantic import ValidationError
|
||
|
||
装配, 身份, 请求, _ = 声音环境
|
||
with pytest.raises((Muse错误, ValidationError)):
|
||
if 坏处 == "口癖":
|
||
请求 = replace(请求, verbal_tics=("样张中不存在",))
|
||
elif 坏处 == "保护":
|
||
请求 = replace(请求, protected_expressions=("虚构保护句",))
|
||
elif 坏处 == "样句":
|
||
请求 = replace(请求, narrator={"达标样张": "虚构的达标样张"})
|
||
elif 坏处 == "哈希":
|
||
请求 = replace(请求, samples=(replace(请求.samples[0], source_hash="0" * 64),))
|
||
elif 坏处 == "人物资产":
|
||
声音提案请求(
|
||
work_id="voice-work",
|
||
expected_voice_revision=0,
|
||
schema_id="style",
|
||
base_version=1,
|
||
narrator={"句式画像": "短句"},
|
||
samples=请求.samples,
|
||
characters={"甲": "再造角色声"},
|
||
)
|
||
else:
|
||
请求 = replace(请求, narrator={})
|
||
装配.要求审校().提出声音(身份, "bad-voice", 请求)
|
||
assert 装配.要求审校().当前声音(身份, "voice-work") is None
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w22-22c003",
|
||
environment="隔离PG与合成样张",
|
||
given="同一声音基线上的两份已审提案",
|
||
when="经真实S04/S01/B05/B03公共用例保存和确认声音",
|
||
then=["另一声音已确认后拒绝旧提案，不覆盖新当前版本"],
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音确认不能覆盖已推进的当前版本__22c003(声音环境):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
服务 = 装配.要求审校()
|
||
_, 一 = _提出(服务, 身份, "voice-a", 请求)
|
||
_, 二 = _提出(服务, 身份, "voice-b", replace(请求, narrator={"句式画像": "长短交替"}))
|
||
旧审阅 = 服务.打开声音审阅(身份, 二)
|
||
_确认(服务, 身份, "adopt-a", 一)
|
||
with pytest.raises(Muse错误, match="基线已变化"):
|
||
服务.决定声音(
|
||
身份, "adopt-b", 二, 审阅ID=旧审阅["review_id"], 审阅哈希=旧审阅["review_hash"]
|
||
)
|
||
assert 服务.当前声音(身份, "voice-work")["narrator"] == 请求.narrator
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w22-22c004",
|
||
environment="隔离PG与合成样张",
|
||
given="作者显式选入的B03资料样张，analysis许可存在/缺失",
|
||
when="经真实S04/S01/B05/B03公共用例保存和确认声音",
|
||
then=["核真实用途与源版本，未许可不出当前声音；合法来源可经明确确认"],
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
@pytest.mark.parametrize("授权", [True, False])
|
||
def test_显式资料样张核来源用途再进入声音__22c004(声音环境, 授权):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
with 装配.要求数据库().连接() as 连, 连.transaction():
|
||
源 = 装配.要求资料().导入(
|
||
连,
|
||
身份.作者,
|
||
导入请求(
|
||
"note",
|
||
"作者选择的样张",
|
||
"synthetic://voice-source",
|
||
"甲走进门。雨落下来。他没有再说话。",
|
||
("analysis",) if 授权 else (),
|
||
),
|
||
)
|
||
请求 = replace(
|
||
请求, samples=(声音样张引用("source", 源.source_id, 源.revision, 源.content_hash),)
|
||
)
|
||
if not 授权:
|
||
with pytest.raises(Muse错误):
|
||
装配.要求审校().提出声音(身份, "unlicensed-voice", 请求)
|
||
else:
|
||
_, loc = _提出(装配.要求审校(), 身份, "source-voice", 请求)
|
||
_确认(装配.要求审校(), 身份, "source-voice-adopt", loc)
|
||
assert 装配.要求审校().当前声音(身份, "voice-work")["revision"] == 1
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-a724239fa69b",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音样张引用另一作品正文时基线强制拒绝__a72423(声音环境):
|
||
装配, 身份, 请求, 文 = 声音环境
|
||
作品 = 装配.要求作品()
|
||
结构 = 作品.新档案结构(身份, "voice-work-b", "work_core", 1)
|
||
作品.保存档案(
|
||
身份, "voice-work-b-create", 档案保存("voice-work-b", 0, {"名称": "另一作品"}, 结构)
|
||
)
|
||
作品.添加章节(
|
||
身份, "voice-ch-b-create", "voice-work-b", "voice-ch-b", "别气样张", 预期目录版本=0
|
||
)
|
||
装配.要求正文().保存人工(
|
||
身份,
|
||
"voice-text-b",
|
||
"voice-ch-b",
|
||
0,
|
||
正文草稿((段落("vb-p1", (文本节点("另一作品的正文。"),)),)),
|
||
)
|
||
文B = 装配.要求正文().读取正文(身份, "voice-ch-b")
|
||
错配 = replace(
|
||
请求,
|
||
samples=(
|
||
声音样张引用("document", "voice-ch-b", 文B["revision"], 文B["document_hash"], "main"),
|
||
),
|
||
)
|
||
with pytest.raises(审校错误, match="样张的作品或正文哈希不符"):
|
||
装配.要求审校().提出声音(身份, "voice-mismatch-baseline", 错配)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-21ed6b89ab91",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音预防输入作品不一致同样拒绝__21ed6b(声音环境):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
作品 = 装配.要求作品()
|
||
结构 = 作品.新档案结构(身份, "voice-work-c", "work_core", 1)
|
||
作品.保存档案(
|
||
身份, "voice-work-c-create", 档案保存("voice-work-c", 0, {"名称": "第三作品"}, 结构)
|
||
)
|
||
作品.添加章节(
|
||
身份, "voice-ch-c-create", "voice-work-c", "voice-ch-c", "预防样张", 预期目录版本=0
|
||
)
|
||
装配.要求正文().保存人工(
|
||
身份,
|
||
"voice-text-c",
|
||
"voice-ch-c",
|
||
0,
|
||
正文草稿((段落("vc-p1", (文本节点("预防输入正文。"),)),)),
|
||
)
|
||
文C = 装配.要求正文().读取正文(身份, "voice-ch-c")
|
||
错配 = replace(
|
||
请求,
|
||
verbal_tics=("预防输入正文",),
|
||
samples=(
|
||
声音样张引用("document", "voice-ch-c", 文C["revision"], 文C["document_hash"], "main"),
|
||
),
|
||
)
|
||
with pytest.raises(审校错误, match="作品或正文哈希"):
|
||
装配.要求审校().提出声音(身份, "voice-mismatch-prevention", 错配)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-ab52c9241b8e",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音提案口癖必须来自真实样张__ab52c9(声音环境):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
with pytest.raises(Muse错误):
|
||
装配.要求审校().提出声音(
|
||
身份, "voice-tic-mismatch", replace(请求, verbal_tics=("样张中不存在",))
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-30b54339e6dd",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音保护表达必须来自真实样张__30b543(声音环境):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
with pytest.raises(Muse错误):
|
||
装配.要求审校().提出声音(
|
||
身份, "voice-protect-mismatch", replace(请求, protected_expressions=("虚构保护句",))
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-3d99ae0ba955",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音决定需要真实审阅回执__3d99ae(声音环境):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
服务 = 装配.要求审校()
|
||
回执, loc = _提出(服务, 身份, "voice-real-review", 请求)
|
||
for 假审阅 in ("missing", "", "00000000-0000-4000-8000-000000000000"):
|
||
with pytest.raises(Muse错误):
|
||
服务.决定声音(身份, "voice-fake-review", loc, 审阅ID=假审阅, 审阅哈希="0" * 64)
|
||
assert 服务.当前声音(身份, 请求.work_id) is None
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-8b9b9e08b9ac",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音提案来源哈希漂移在提出时拒绝__8b9b9e(声音环境):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
with pytest.raises(Muse错误):
|
||
装配.要求审校().提出声音(
|
||
身份,
|
||
"voice-hash-drift",
|
||
replace(请求, samples=(replace(请求.samples[0], source_hash="0" * 64),)),
|
||
)
|
||
assert 装配.要求审校().当前声音(身份, "voice-work") is None
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-3abe1c197cb2",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_声音候选不进入当前投影__3abe1c(声音环境):
|
||
装配, 身份, 请求, _ = 声音环境
|
||
服务 = 装配.要求审校()
|
||
assert 服务.当前声音(身份, 请求.work_id) is None
|
||
_提出(服务, 身份, "voice-no-projection", 请求)
|
||
assert 服务.当前声音(身份, 请求.work_id) is None
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-ab359a56afb0",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_规则候选保持候选且例证跨作品可回查__ab359a(四类案例):
|
||
装配, 身份, 建立 = 四类案例
|
||
服务 = 装配.要求审校()
|
||
引用 = 建立()
|
||
结果 = 服务.提出案例规则(身份, 案例规则请求(_规则定义("candidate-ab359a"), 引用))
|
||
assert 结果["status"] == "candidate"
|
||
库 = 服务.读取规则库(身份, 预览候选=True)
|
||
规则 = 库["rules"]["candidate-ab359a"]
|
||
assert 规则["case_card_ids"] == sorted(c.case_id for c in 引用)
|
||
# 四类案例夹具按 work-a/work-b 交替建立；例证作品分组可从已确认案例回查。
|
||
作品集 = {
|
||
服务.读取案例(身份, c.case_id)["payload"]["evidence"]["before"]["work_group"] for c in 引用
|
||
}
|
||
assert 作品集 == {"work-a", "work-b"}
|
||
|
||
|
||
@pytest.fixture
|
||
def 案例环境(章后环境):
|
||
装配, 身份 = 章后环境["装配"], 章后环境["作者"]
|
||
|
||
def 请求(
|
||
*,
|
||
文本="值得注意的是，门外下雨了。",
|
||
来源="synthetic:card-backfill-l002-001",
|
||
用途=("sample",),
|
||
作品="work-a",
|
||
起点=0,
|
||
终点=None,
|
||
保全=True,
|
||
片段哈希=None,
|
||
来源哈希=None,
|
||
):
|
||
with 装配.要求数据库().连接() as 连, 连.transaction():
|
||
源 = 装配.要求资料().导入(
|
||
连,
|
||
身份.作者,
|
||
导入请求("reference", "合成案例来源", 来源, 文本, 用途, 作品),
|
||
)
|
||
end = len(文本) if 终点 is None else 终点
|
||
定位 = 案例来源(
|
||
"source",
|
||
源.source_id,
|
||
源.revision,
|
||
来源哈希 or 源.content_hash,
|
||
起点,
|
||
end,
|
||
片段哈希 or hashlib.sha256(文本[起点:end].encode()).hexdigest(),
|
||
)
|
||
return 案例采集请求(定位, preserve_text=保全), 源, 文本
|
||
|
||
return 装配, 身份, 请求
|
||
|
||
|
||
def _采集(装配, 身份, 命令, 请求):
|
||
结果 = 装配.要求审校().采集案例(身份, 命令, 请求)["results"][0]
|
||
return 案例定位(结果["case_id"], 结果["revision"], 结果["case_hash"])
|
||
|
||
|
||
def _复核(定位, 标签="sf"):
|
||
return 案例复核请求(
|
||
定位,
|
||
标签,
|
||
"retain" if 标签 == "snf" else "repair",
|
||
"作者确认：此处无功能" if 标签 == "sf" else "作者确认案例边界。",
|
||
)
|
||
|
||
|
||
def _确认案例(装配, 身份, 定位, 命令="confirm-case", 标签="sf"):
|
||
服务 = 装配.要求审校()
|
||
请求 = _复核(定位, 标签)
|
||
审阅 = 服务.打开案例审阅(身份, 请求)
|
||
return 服务.决定案例(
|
||
身份,
|
||
命令,
|
||
请求,
|
||
审阅ID=审阅["review_id"],
|
||
审阅哈希=审阅["review_hash"],
|
||
)
|
||
|
||
|
||
def _规则定义(rule_id="candidate-l999", version=1):
|
||
return {
|
||
"id": rule_id,
|
||
"name": "待评估元话语",
|
||
"layer": "lexical",
|
||
"carrier_scope": "narration",
|
||
"trigger": {"type": "model_judgment", "criteria": "上下文功能判断"},
|
||
"default_disposition": "candidate",
|
||
"function_check": ["是否承担叙事功能"],
|
||
"fix_hint": "先仲裁再决定",
|
||
"version": version,
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def 四类案例(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
轮次 = itertools.count()
|
||
|
||
def 建立(*, 标签=("sf", "snf", "boundary", "regression"), 作品=("work-a", "work-b")):
|
||
# 采集与确认命令按调用轮次唯一：同命令重放返回历史回执，其定位在确认后已过期。
|
||
轮 = next(轮次)
|
||
结果 = []
|
||
for i, kind in enumerate(标签):
|
||
文本 = f"第{轮}轮第{i}份例证。前半句作为证据，后半句保留语境。"
|
||
请求, _, _ = 生成(
|
||
文本=文本,
|
||
来源=f"synthetic:rule-source:{轮}-{i}",
|
||
作品=作品[i % len(作品)],
|
||
终点=7,
|
||
)
|
||
定位 = _采集(装配, 身份, f"sample-case-{轮}-{i}", 请求)
|
||
_确认案例(装配, 身份, 定位, f"confirm-{轮}-{i}", kind)
|
||
已存 = 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
结果.append(案例定位(定位.case_id, 已存["revision"], 已存["case_hash"]))
|
||
return tuple(结果)
|
||
|
||
return 装配, 身份, 建立
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-31eae78cbd8c",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_发布案例改用公开请求合同读回__31eae7(案例环境):
|
||
from pydantic import TypeAdapter
|
||
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成()
|
||
定位 = _采集(装配, 身份, "card-backfill-l002-001", 请求)
|
||
案例 = 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
重验 = TypeAdapter(案例采集请求).validate_python(案例["payload"]["request"])
|
||
assert 重验 == 请求 and 案例["case_hash"] == 定位.case_hash
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-9b9153c89f65",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_回填先为待复核且明确确认后才可作样例__9b9153(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成()
|
||
定位 = _采集(装配, 身份, "backfill-shadow", 请求)
|
||
服务 = 装配.要求审校()
|
||
assert 服务.读取案例(身份, 定位.case_id)["state"] == "shadow"
|
||
assert not 服务.读取案例(身份, 定位.case_id)["sampleable"]
|
||
_确认案例(装配, 身份, 定位)
|
||
已确认 = 服务.读取案例(身份, 定位.case_id)
|
||
assert 已确认["state"] == "confirmed" and 已确认["sampleable"]
|
||
assert 已确认["payload"]["label"] == "sf"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-07715d77085a",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_未授权来源仅留哈希且不能确认__07715d(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, 文本 = 生成(用途=("analysis",))
|
||
定位 = _采集(装配, 身份, "unlicensed-card", 请求)
|
||
服务 = 装配.要求审校()
|
||
案例 = 服务.读取案例(身份, 定位.case_id)
|
||
证据 = 案例["payload"]["evidence"]["before"]
|
||
assert 证据["excerpt"] is None and 证据["context"] is None
|
||
assert 证据["excerpt_hash"] == hashlib.sha256(文本.encode()).hexdigest()
|
||
with pytest.raises(Muse错误, match="哈希元信息"):
|
||
服务.打开案例审阅(身份, _复核(定位))
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-3bb0818a8c8d",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_只有哈希的案例请求拒绝偷注原文__3bb081(案例环境):
|
||
from dataclasses import asdict
|
||
|
||
from pydantic import TypeAdapter, ValidationError
|
||
|
||
_, _, 生成 = 案例环境
|
||
请求, _, _ = 生成(用途=("analysis",))
|
||
with pytest.raises(ValidationError):
|
||
TypeAdapter(案例采集请求).validate_python(
|
||
{**asdict(请求), "excerpt": "偷偷保存的第三方原文"}
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-876da9487f56",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_规则始终是候选且缺反证不能建立__876da9(四类案例):
|
||
装配, 身份, 建立 = 四类案例
|
||
服务 = 装配.要求审校()
|
||
引用 = 建立()
|
||
结果 = 服务.提出案例规则(身份, 案例规则请求(_规则定义(), 引用))
|
||
assert 结果["status"] == "candidate"
|
||
assert 服务.读取规则库(身份)["rules"] == {}
|
||
单边 = 建立(标签=("sf", "sf", "sf", "sf"))
|
||
with pytest.raises(Muse错误, match="四类"):
|
||
服务.提出案例规则(身份, 案例规则请求(_规则定义("candidate-l998"), 单边))
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-aff5d8c29022",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_实时反馈必须同时绑定前后版本和决定回执__aff5d8(案例环境):
|
||
from pydantic import ValidationError
|
||
|
||
_, _, 生成 = 案例环境
|
||
请求, _, _ = 生成()
|
||
with pytest.raises((Muse错误, ValidationError)):
|
||
案例采集请求(请求.before, mode="live_feedback")
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-8246112bdea3",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_内置发布输入导入后与候选读回指纹一致__824611(章后环境):
|
||
装配, 身份 = 章后环境["装配"], 章后环境["作者"]
|
||
内置 = 读取内置规则()
|
||
装配.要求审校().导入规则候选(身份)
|
||
已存 = 装配.要求审校().读取规则库(身份, 预览候选=True)
|
||
# 正式合同：导入按规则不可变版本持久化其引用的例证；发布输入中未被任何规则引用
|
||
# （反向引用为空）的事故锚点留在安装包发布输入内，不进入规则版本，也不改变指纹。
|
||
引用 = {ref for r in 内置["rules"].values() for refs in r["samples"].values() for ref in refs}
|
||
未引用 = {ref for ref, s in 内置["samples"].items() if not s["rules"]}
|
||
assert 未引用 == set(内置["samples"]) - set(已存["samples"])
|
||
assert 已存["rules"] == 内置["rules"]
|
||
assert 已存["samples"] == {ref: 内置["samples"][ref] for ref in 引用}
|
||
assert 已存["fingerprint"] == 内置["fingerprint"]
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-179b3c21692c",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_规则库失败关闭且不回退内置种子__179b3c(章后环境, monkeypatch):
|
||
from psycopg import OperationalError
|
||
|
||
服务 = 章后环境["装配"].要求审校()
|
||
|
||
def 断库(*args, **kwargs):
|
||
raise OperationalError("connection refused")
|
||
|
||
monkeypatch.setattr(type(服务.数据库), "连接", 断库)
|
||
with pytest.raises(OperationalError, match="connection refused"):
|
||
服务.读取规则库(章后环境["作者"], 预览候选=True)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-e88dcd446de0",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_规则缺少正例时在入库前失败关闭__e88dcd(章后环境):
|
||
库 = deepcopy(读取内置规则())
|
||
库["rules"]["l001"]["samples"]["sf"] = []
|
||
with pytest.raises(Muse错误, match="四类例证"):
|
||
章后环境["装配"].要求审校().导入规则候选(章后环境["作者"], 规则库=库)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-ac1163ea3657",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_规则与样例读回还原不可变载荷哈希__ac1163(章后环境):
|
||
装配, 身份 = 章后环境["装配"], 章后环境["作者"]
|
||
服务 = 装配.要求审校()
|
||
服务.导入规则候选(身份)
|
||
库 = 服务.读取规则库(身份, 预览候选=True)
|
||
for rule_id, rule in 库["rules"].items():
|
||
版本 = 服务.读取规则版本(身份, rule_id, rule["version"])
|
||
assert 版本["payload"]["rule"] == rule
|
||
assert 版本["payload_hash"] == 固定哈希(版本["payload"])
|
||
assert set(版本["payload"]["samples"]) == {
|
||
ref for refs in rule["samples"].values() for ref in refs
|
||
}
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-64a9f29d1b89",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_真实隔离库候选读回不需要旧烟雾开关__64a9f2(章后环境):
|
||
服务, 身份 = 章后环境["装配"].要求审校(), 章后环境["作者"]
|
||
导入 = 服务.导入规则候选(身份)
|
||
读回 = 服务.读取规则库(身份, 预览候选=True)
|
||
assert 导入["persisted"] and len(读回["rules"]) == len(导入["results"])
|
||
assert all(rule["status"] == "candidate" for rule in 读回["rules"].values())
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-a0fd532aef44",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_首次发布输入为每条规则建立独立版本__a0fd53(章后环境):
|
||
服务, 身份 = 章后环境["装配"].要求审校(), 章后环境["作者"]
|
||
结果 = 服务.导入规则候选(身份)
|
||
assert 结果["persisted"] and all(not r["duplicate"] for r in 结果["results"])
|
||
assert len({r["version_id"] for r in 结果["results"]}) == len(读取内置规则()["rules"])
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-5ddea6a3d165",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_同内容第二次导入幂等且不增加版本__5ddea6(章后环境):
|
||
服务, 身份 = 章后环境["装配"].要求审校(), 章后环境["作者"]
|
||
首次 = 服务.导入规则候选(身份)
|
||
再次 = 服务.导入规则候选(身份)
|
||
assert all(r["duplicate"] for r in 再次["results"])
|
||
assert [r["version_id"] for r in 再次["results"]] == [r["version_id"] for r in 首次["results"]]
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-ab996bb220ec",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_变更规则只能追加递增候选版本__ab996b(章后环境):
|
||
服务, 身份 = 章后环境["装配"].要求审校(), 章后环境["作者"]
|
||
服务.导入规则候选(身份)
|
||
库 = deepcopy(读取内置规则())
|
||
库["rules"]["l001"]["fix_hint"] = "更新后的修复提示"
|
||
with pytest.raises(Muse错误, match="递增版本"):
|
||
服务.导入规则候选(身份, 规则库=库)
|
||
库["rules"]["l001"]["version"] = 2
|
||
服务.导入规则候选(身份, 规则库=库)
|
||
assert 服务.读取规则版本(身份, "l001", 1)["version"] == 1
|
||
assert 服务.读取规则版本(身份, "l001", 2)["payload"]["rule"]["fix_hint"] == "更新后的修复提示"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-bc74312a5f4f",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_作者额外候选不被种子强制覆盖__bc7431(四类案例):
|
||
装配, 身份, 建立 = 四类案例
|
||
服务 = 装配.要求审校()
|
||
服务.提出案例规则(身份, 案例规则请求(_规则定义("z999"), 建立()))
|
||
服务.导入规则候选(身份)
|
||
读回 = 服务.读取规则库(身份, 预览候选=True)
|
||
assert "z999" in 读回["rules"] and 读回["rules"]["z999"]["status"] == "candidate"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-59c4d824b968",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_规则预览只返回计划且不触碰数据库__59c4d8(章后环境):
|
||
服务, 身份 = 章后环境["装配"].要求审校(), 章后环境["作者"]
|
||
计划 = 服务.导入规则候选(身份, 预览=True)
|
||
assert 计划 == {
|
||
"persisted": False,
|
||
"fingerprint": 读取内置规则()["fingerprint"],
|
||
"count": len(读取内置规则()["rules"]),
|
||
}
|
||
assert 服务.读取规则库(身份, 预览候选=True)["rules"] == {}
|
||
|
||
|
||
def _实时反馈请求(案例环境):
|
||
装配, 身份, _ = 案例环境
|
||
正文 = 装配.要求正文()
|
||
原文, 候选 = "候选正文", "候选正文。"
|
||
正文.保存人工(
|
||
身份,
|
||
"feedback-before",
|
||
"ch-1",
|
||
0,
|
||
正文草稿((段落("feedback-p1", (文本节点(原文),)),)),
|
||
)
|
||
前 = 正文.读取正文(身份, "ch-1")
|
||
回执 = 正文.保存人工(
|
||
身份,
|
||
"run-7",
|
||
"ch-1",
|
||
1,
|
||
正文草稿((段落("feedback-p1", (文本节点(候选),)),)),
|
||
)
|
||
后 = 正文.读取正文(身份, "ch-1")
|
||
before = 案例来源(
|
||
"document",
|
||
"ch-1",
|
||
1,
|
||
前["document_hash"],
|
||
0,
|
||
len(原文),
|
||
hashlib.sha256(原文.encode()).hexdigest(),
|
||
"main",
|
||
)
|
||
after = 案例来源(
|
||
"document",
|
||
"ch-1",
|
||
2,
|
||
后["document_hash"],
|
||
0,
|
||
len(候选),
|
||
hashlib.sha256(候选.encode()).hexdigest(),
|
||
"main",
|
||
)
|
||
return (
|
||
装配,
|
||
身份,
|
||
案例采集请求(
|
||
before,
|
||
after,
|
||
mode="live_feedback",
|
||
preserve_text=True,
|
||
feedback_command_id=回执["command_id"],
|
||
),
|
||
回执,
|
||
)
|
||
|
||
|
||
def _配置审校CLI(装配, 身份, tmp_path):
|
||
from muse.配置 import 服务配置
|
||
|
||
口令 = tmp_path / "cli-password"
|
||
口令.write_text("synthetic-cli-password")
|
||
口令.chmod(0o600)
|
||
装配.配置 = replace(
|
||
装配.配置,
|
||
HTTP=服务配置(
|
||
str(口令),
|
||
作者ID=身份.作者,
|
||
公开地址="http://testserver",
|
||
允许来源=("http://testserver",),
|
||
),
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-b3ddcb88c6f9",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_来源哈希必须由所登记的原文字节决定__b3ddcb(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成(文本="值得注意的是。", 来源哈希="0" * 64)
|
||
with pytest.raises(Muse错误, match="来源哈希"):
|
||
装配.要求审校().采集案例(身份, "wrong-source-hash", 请求)
|
||
assert 装配.要求审校().列出案例(身份) == []
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-7720243828bc",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_多来源清单按稳定身份排序且研究授权只显示哈希__772024(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
for i, 文本 in enumerate(("值得注意的是。", "嘴角微微上扬。")):
|
||
请求, _, _ = 生成(文本=文本, 来源=f"synthetic:inventory:{i}", 用途=("analysis",))
|
||
_采集(装配, 身份, f"inventory-{1 - i}", 请求)
|
||
清单 = 装配.要求审校().列出案例(身份)
|
||
assert len(清单) == len({c["case_id"] for c in 清单}) == 2
|
||
assert [c["case_id"] for c in 清单] == sorted(c["case_id"] for c in 清单)
|
||
assert all(c["state"] == "shadow" for c in 清单)
|
||
assert all(c["payload"]["evidence"]["before"]["excerpt"] is None for c in 清单)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-31b308b1132c",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_研究授权回填可以从同一数据库身份重放__31b308(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
文本 = "第一行。值得注意的是，门外下雨了。\n"
|
||
起点 = 文本.index("值得")
|
||
请求, _, _ = 生成(文本=文本, 用途=("analysis",), 起点=起点, 终点=len(文本) - 1)
|
||
定位 = _采集(装配, 身份, "research-replay", 请求)
|
||
首次 = 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
再次 = 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
assert 首次 == 再次 and 首次["state"] == "shadow"
|
||
assert 首次["payload"]["request"]["before"]["start"] == 起点
|
||
assert 首次["payload"]["evidence"]["before"]["excerpt"] is None
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-5f50720bdba1",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_可作样例来源保留确切片段上下文与位置__5f5072(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
文本 = "她抬头。嘴角微微上扬。\n"
|
||
起点, 终点 = 文本.index("嘴角"), 文本.index("。\n") + 1
|
||
请求, _, _ = 生成(文本=文本, 来源="synthetic:owned", 起点=起点, 终点=终点)
|
||
定位 = _采集(装配, 身份, "owned-backfill", 请求)
|
||
证据 = 装配.要求审校().读取案例(身份, 定位.case_id)["payload"]["evidence"]["before"]
|
||
assert 证据["excerpt"] == "嘴角微微上扬。"
|
||
assert "她抬头" in 证据["context"]
|
||
assert 请求.before.start == 起点 and 请求.before.end == 终点
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-81d43ffe316c",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_同一采集命令重放去重且只留一个案例__81d43f(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成(文本="值得注意的是。", 来源="synthetic:same")
|
||
首次 = _采集(装配, 身份, "same-pattern", 请求)
|
||
重放 = _采集(装配, 身份, "same-pattern", 请求)
|
||
assert 重放 == 首次 and len(装配.要求审校().列出案例(身份)) == 1
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-1062c87c6b5e",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_反馈案例绑定实际创作决定回执__1062c8(案例环境):
|
||
装配, 身份, 请求, 回执 = _实时反馈请求(案例环境)
|
||
定位 = _采集(装配, 身份, "feedback-case", 请求)
|
||
案例 = 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
assert 案例["payload"]["request"]["mode"] == "live_feedback"
|
||
assert 案例["payload"]["feedback"] == {
|
||
key: 回执[key] for key in ("receipt_id", "command_id", "action")
|
||
}
|
||
assert 案例["state"] == "shadow"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-2a5b8f6ffec2",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_未授权案例数据库也不存原文且不能确认__2a5b8f(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, 文本 = 生成(文本="值得注意的是。", 来源="synthetic:third-party", 用途=("analysis",))
|
||
定位 = _采集(装配, 身份, "unlicensed-persist", 请求)
|
||
with pytest.raises(Muse错误, match="哈希元信息"):
|
||
_确认案例(装配, 身份, 定位)
|
||
with 装配.要求数据库().连接(只读=True) as 连:
|
||
载荷 = 连.execute(
|
||
"SELECT payload::text FROM muse_quality_case_version WHERE case_id=%s", (定位.case_id,)
|
||
).fetchone()[0]
|
||
assert 文本 not in 载荷
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-055c73ace6f4",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_未变来源每次读回都重验为当前可用__055c73(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成()
|
||
定位 = _采集(装配, 身份, "verified-source", 请求)
|
||
_确认案例(装配, 身份, 定位)
|
||
读回 = 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
assert 读回["sampleable"] and 读回["unavailable_sources"] == []
|
||
assert 读回 == 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-7464250e5d21",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_来源推进不改写案例但使旧审阅失效__746425(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, 文本 = 生成(来源="synthetic:changed-source")
|
||
定位 = _采集(装配, 身份, "source-before-change", 请求)
|
||
服务 = 装配.要求审校()
|
||
复核 = _复核(定位)
|
||
审阅 = 服务.打开案例审阅(身份, 复核)
|
||
with 装配.要求数据库().连接() as 连, 连.transaction():
|
||
装配.要求资料().导入(
|
||
连,
|
||
身份.作者,
|
||
导入请求(
|
||
"reference",
|
||
"合成案例来源",
|
||
"synthetic:changed-source",
|
||
文本 + "已经改成另一版正文。",
|
||
("sample",),
|
||
"work-a",
|
||
),
|
||
)
|
||
with pytest.raises(Muse错误):
|
||
服务.决定案例(
|
||
身份,
|
||
"stale-source-review",
|
||
复核,
|
||
审阅ID=审阅["review_id"],
|
||
审阅哈希=审阅["review_hash"],
|
||
)
|
||
assert 服务.读取案例(身份, 定位.case_id)["state"] == "shadow"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-c83b1ea9504a",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_篡改片段锚点在案例入库前拒绝__c83b1e(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
文本 = "值得注意的是。"
|
||
请求, _, _ = 生成(文本=文本, 起点=1, 片段哈希=hashlib.sha256(文本.encode()).hexdigest())
|
||
with pytest.raises(Muse错误, match="片段或来源哈希"):
|
||
装配.要求审校().采集案例(身份, "tampered-anchor", 请求)
|
||
assert 装配.要求审校().列出案例(身份) == []
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-7774c6a20238",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_来源授权不可用时读回失败关闭__7774c6(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, 来源, _ = 生成(来源="synthetic:unavailable")
|
||
定位 = _采集(装配, 身份, "unavailable-source", 请求)
|
||
with 装配.要求数据库().连接() as 连, 连.transaction():
|
||
连.execute(
|
||
"UPDATE muse_source SET authorized_uses=%s::jsonb WHERE source_id=%s",
|
||
('["analysis"]', 来源.source_id),
|
||
)
|
||
读回 = 装配.要求审校().读取案例(身份, 定位.case_id)
|
||
assert 读回["unavailable_sources"] == ["before"] and not 读回["sampleable"]
|
||
assert 读回["payload"]["evidence"]["before"]["excerpt"] is None
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-1c36821e1438",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_案例清单对同来源多案例各只读回一次__1c3682(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
文本 = "值得注意的是。嘴角微微上扬。"
|
||
请求一, 来源, _ = 生成(文本=文本, 来源="synthetic:shared-source", 终点=9)
|
||
请求二 = 案例采集请求(
|
||
replace(
|
||
请求一.before,
|
||
start=9,
|
||
end=len(文本),
|
||
excerpt_hash=hashlib.sha256(文本[9:].encode()).hexdigest(),
|
||
),
|
||
preserve_text=True,
|
||
)
|
||
for i, 请求 in enumerate((请求一, 请求二)):
|
||
定位 = _采集(装配, 身份, f"shared-source-{i}", 请求)
|
||
_确认案例(装配, 身份, 定位, f"confirm-shared-{i}")
|
||
清单 = 装配.要求审校().列出案例(身份)
|
||
assert len(清单) == 2 and all(c["sampleable"] for c in 清单)
|
||
assert {c["payload"]["request"]["before"]["ref_id"] for c in 清单} == {来源.source_id}
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-5a6e2469583a",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_采集回执可以直接固定下一步审阅定位__5a6e24(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成(文本="值得注意的是。", 来源="synthetic:owned-wrapper")
|
||
定位 = _采集(装配, 身份, "receipt-wrapper", 请求)
|
||
# S01 审阅回执结构：候选身份在快照内以候选ID/候选版本/候选哈希固定。
|
||
审阅 = 装配.要求审校().打开案例审阅(身份, _复核(定位))
|
||
快照 = 审阅["snapshot"]
|
||
assert 快照["候选ID"] == 定位.case_id
|
||
assert 快照["候选版本"] == 定位.revision
|
||
assert 快照["候选哈希"] == 定位.case_hash
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-98a3969a2ce3",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_案例确认要求真实审阅回执__98a396(案例环境):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成()
|
||
定位 = _采集(装配, 身份, "verified-review", 请求)
|
||
with pytest.raises(Muse错误):
|
||
装配.要求审校().决定案例(
|
||
身份,
|
||
"fake-case-review",
|
||
_复核(定位),
|
||
审阅ID="00000000-0000-4000-8000-000000000000",
|
||
审阅哈希="0" * 64,
|
||
)
|
||
_确认案例(装配, 身份, 定位, "verified-confirm")
|
||
assert 装配.要求审校().读取案例(身份, 定位.case_id)["state"] == "confirmed"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-f21ddb209c73",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_未知载体字段不能混入案例合同__f21ddb(案例环境):
|
||
from dataclasses import asdict
|
||
|
||
from pydantic import TypeAdapter, ValidationError
|
||
|
||
_, _, 生成 = 案例环境
|
||
请求, _, _ = 生成()
|
||
with pytest.raises(ValidationError):
|
||
TypeAdapter(案例采集请求).validate_python({**asdict(请求), "carrier": "unknown"})
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-41e8763c05b4",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_案例只在人工复核后投影成规则样例__41e876(四类案例):
|
||
装配, 身份, 建立 = 四类案例
|
||
引用 = 建立()
|
||
服务 = 装配.要求审校()
|
||
服务.提出案例规则(身份, 案例规则请求(_规则定义(), 引用))
|
||
库 = 服务.读取规则库(身份, 预览候选=True)
|
||
for sample in 库["samples"].values():
|
||
assert sample["case_card_id"] in {c.case_id for c in 引用}
|
||
assert sample["source"] == "licensed" and sample["carrier"] == "mixed"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-b1ebdb636f6f",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_CLI采集后自动进入数据库案例清单__b1ebdb(案例环境, tmp_path):
|
||
from muse.接入.cli.审校命令 import 运行审校命令
|
||
|
||
装配, 身份, 生成 = 案例环境
|
||
_配置审校CLI(装配, 身份, tmp_path)
|
||
请求, _, _ = 生成(来源="synthetic:cli-inventory")
|
||
路径 = tmp_path / "capture.json"
|
||
from dataclasses import asdict
|
||
|
||
路径.write_text(
|
||
json.dumps({"command_id": "cli-capture", "request": asdict(请求)}, ensure_ascii=False)
|
||
)
|
||
回执 = 运行审校命令(装配, "案例采集", str(路径))
|
||
清单 = 运行审校命令(装配, "案例清单", None)
|
||
assert [c["case_id"] for c in 清单] == [回执["results"][0]["case_id"]]
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-95a9024e5a4c",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_CLI拒绝从文件扫描原文暗中生成案例__95a902(案例环境, tmp_path):
|
||
from muse.共享.错误 import 配置错误
|
||
from muse.接入.cli.审校命令 import 运行审校命令
|
||
|
||
装配, 身份, _ = 案例环境
|
||
_配置审校CLI(装配, 身份, tmp_path)
|
||
路径 = tmp_path / "work.txt"
|
||
路径.write_text("值得注意的是。")
|
||
with pytest.raises(配置错误, match="未知审校动作"):
|
||
运行审校命令(装配, "scan", str(路径))
|
||
assert 装配.要求审校().列出案例(身份) == []
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-5bb941f5815a",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_CLI实时反馈默认持久化真实决定回执__5bb941(案例环境, tmp_path):
|
||
from dataclasses import asdict
|
||
|
||
from muse.接入.cli.审校命令 import 运行审校命令
|
||
|
||
装配, 身份, 请求, 回执 = _实时反馈请求(案例环境)
|
||
_配置审校CLI(装配, 身份, tmp_path)
|
||
路径 = tmp_path / "feedback.yaml"
|
||
路径.write_text(
|
||
json.dumps({"command_id": "cli-feedback", "request": asdict(请求)}, ensure_ascii=False)
|
||
)
|
||
定位 = 运行审校命令(装配, "案例采集", str(路径))["results"][0]
|
||
读回 = 运行审校命令(装配, "案例", 定位["case_id"])
|
||
assert 读回["payload"]["feedback"]["command_id"] == 回执["command_id"] == "run-7"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-4b6ea1536967",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_案例重验直接读数据库不生成仓外边车文件__4b6ea1(案例环境, tmp_path):
|
||
装配, 身份, 生成 = 案例环境
|
||
请求, _, _ = 生成(来源="synthetic:no-sidecar")
|
||
定位 = _采集(装配, 身份, "no-sidecar", 请求)
|
||
装配.要求审校().读取案例(身份, 定位.case_id)
|
||
assert not list(tmp_path.glob("*.inventory.json"))
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-fc782c51428f",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_案例CLI不暴露确认投影或启用旧命令__fc782c(案例环境, tmp_path):
|
||
from muse.共享.错误 import 配置错误
|
||
from muse.接入.cli.审校命令 import 运行审校命令
|
||
|
||
装配, 身份, _ = 案例环境
|
||
_配置审校CLI(装配, 身份, tmp_path)
|
||
目标 = tmp_path / "request.json"
|
||
目标.write_text("{}")
|
||
for 旧命令 in ("confirm", "project", "activate"):
|
||
with pytest.raises(配置错误, match="未知审校动作"):
|
||
运行审校命令(装配, 旧命令, str(目标))
|
||
|
||
|
||
@pytest.fixture
|
||
def 诊断环境(章后环境):
|
||
装配, 身份 = 章后环境["装配"], 章后环境["作者"]
|
||
文本 = "值得注意的是，窗外正在下雨。"
|
||
装配.要求正文().保存人工(
|
||
身份,
|
||
"diagnosis-source",
|
||
"ch-2",
|
||
0,
|
||
正文草稿((段落("diagnosis-p1", (文本节点(文本),)),)),
|
||
)
|
||
正文 = 装配.要求正文().读取正文(身份, "ch-2")
|
||
目标 = {
|
||
"work_id": "post-work",
|
||
"chapter_id": "ch-2",
|
||
"branch_id": "main",
|
||
"revision": 正文["revision"],
|
||
"document_hash": 正文["document_hash"],
|
||
}
|
||
return 装配, 身份, 目标
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-71e761cc843b",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_诊断正文哈希不匹配时在报告入库前拒绝__71e761(诊断环境):
|
||
装配, 身份, 目标 = 诊断环境
|
||
with pytest.raises(审校错误, match="哈希不符"):
|
||
装配.要求审校().生成诊断(
|
||
身份,
|
||
{
|
||
**目标,
|
||
"document_hash": "0" * 64,
|
||
},
|
||
)
|
||
with 装配.要求数据库().连接(只读=True) as 连:
|
||
assert 连.execute("SELECT count(*) FROM muse_diagnosis").fetchone()[0] == 0
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-4c7dc3fdcabe",
|
||
environment="隔离 PostgreSQL，检测替身",
|
||
when="持久化、重验或查询诊断、声音、规则与案例。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_CLI默认持久化诊断并可按身份读回__4c7dc3(诊断环境, tmp_path):
|
||
from pydantic import TypeAdapter
|
||
|
||
from muse.接入.cli.审校命令 import 运行审校命令
|
||
from muse.接入.审校操作 import 诊断请求
|
||
|
||
装配, 身份, 目标 = 诊断环境
|
||
_配置审校CLI(装配, 身份, tmp_path)
|
||
请求 = TypeAdapter(诊断请求).validate_python({"target": 目标, "preview_candidates": True})
|
||
路径 = tmp_path / "diagnosis.json"
|
||
路径.write_text(json.dumps(请求.model_dump(mode="json"), ensure_ascii=False))
|
||
结果 = 运行审校命令(装配, "诊断", str(路径))
|
||
assert 结果["report"]["persisted"] is True
|
||
assert 运行审校命令(装配, "诊断报告", 结果["diagnosis_id"])["report"] == 结果["report"]
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-873f1d93b7df",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_CLI诊断默认持久化且同一作者可读回__873f1d(诊断环境, tmp_path):
|
||
from pydantic import TypeAdapter
|
||
|
||
from muse.接入.cli.审校命令 import 运行审校命令
|
||
from muse.接入.审校操作 import 诊断请求
|
||
|
||
装配, 身份, 目标 = 诊断环境
|
||
_配置审校CLI(装配, 身份, tmp_path)
|
||
装配.要求审校().导入规则候选(身份)
|
||
请求 = TypeAdapter(诊断请求).validate_python({"target": 目标, "preview_candidates": True})
|
||
路径 = tmp_path / "diag-cli.json"
|
||
路径.write_text(json.dumps(请求.model_dump(mode="json"), ensure_ascii=False))
|
||
结果 = 运行审校命令(装配, "诊断", str(路径))
|
||
assert 结果["report"]["persisted"] is True
|
||
|
||
读回 = 运行审校命令(装配, "诊断报告", 结果["diagnosis_id"])
|
||
assert 读回["diagnosis_id"] == 结果["diagnosis_id"]
|
||
assert 读回["stale"] is False and 读回["stale_reasons"] == []
|
||
报告 = 读回["report"]
|
||
assert 报告 == 结果["report"]
|
||
assert 报告["target"] == 目标 and 报告["work_id"] == "post-work"
|
||
assert 报告["schema_version"] == "diagnosis-v1"
|
||
assert 报告["mode"] == "candidate_preview" and 报告["persisted"] is True
|
||
assert 报告["text_hash"] == hashlib.sha256("值得注意的是，窗外正在下雨。".encode()).hexdigest()
|
||
assert 报告["coverage"]["text_length"] == len("值得注意的是，窗外正在下雨。")
|
||
规则库 = 读取内置规则()
|
||
assert 报告["rule_library_version"] == 规则库["fingerprint"]
|
||
assert 报告["findings"]
|
||
for 发现 in 报告["findings"]:
|
||
assert 发现["rule_version"] == 规则库["rules"][发现["rule_id"]]["version"]
|
||
旧配置 = 装配.配置
|
||
assert 旧配置.HTTP is not None
|
||
try:
|
||
装配.配置 = replace(旧配置, HTTP=replace(旧配置.HTTP, 作者ID="synthetic-other-author"))
|
||
with pytest.raises(审校错误, match="诊断不存在、不属于当前作者或哈希不符") as 错误:
|
||
运行审校命令(装配, "诊断报告", 结果["diagnosis_id"])
|
||
assert "值得注意的是" not in json.dumps(错误.value.呈现(), ensure_ascii=False)
|
||
assert 结果["diagnosis_id"] not in json.dumps(错误.value.呈现(), ensure_ascii=False)
|
||
finally:
|
||
装配.配置 = 旧配置
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-93daa381c03a",
|
||
environment="离线，固定样张、规则和四类反例",
|
||
when="生成声音草案、投影保护项或验证规则候选。",
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_规则验证CLI无离线旗标必须显式评测配置__93daa3(章后环境, tmp_path):
|
||
from pydantic import TypeAdapter
|
||
|
||
from muse.共享.错误 import 配置错误
|
||
from muse.接入.cli.经验命令 import 运行经验命令
|
||
from muse.接入.经验操作 import 规则改进验证输入
|
||
|
||
装配, 身份 = 章后环境["装配"], 章后环境["作者"]
|
||
_配置审校CLI(装配, 身份, tmp_path)
|
||
输入 = TypeAdapter(规则改进验证输入).validate_python(
|
||
{
|
||
"command_id": "cli-rule-verify",
|
||
"selection": {
|
||
"improvement_id": "11111111-1111-4111-8111-111111111111",
|
||
"revision": 1,
|
||
"content_hash": "0" * 64,
|
||
},
|
||
"request": {
|
||
"dataset_version_id": "22222222-2222-4222-8222-222222222222",
|
||
"rule_id": "l001",
|
||
"rule_version": 1,
|
||
"rule_hash": "1" * 64,
|
||
},
|
||
}
|
||
)
|
||
目标 = tmp_path / "rule-verify.json"
|
||
目标.write_text(json.dumps(输入.model_dump(mode="json"), ensure_ascii=False), encoding="utf-8")
|
||
with pytest.raises(配置错误, match="评测配置"):
|
||
运行经验命令(装配, "申请规则验证", str(目标))
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w22-22c005",
|
||
environment="隔离PG与合成模型回执",
|
||
given="真实逐片审校任务完成回执与隔离 CLI 配置",
|
||
when="任务完成后经服务与统一 CLI 读回同一报告并核对逐片证据",
|
||
then=["全文报告由实际逐片回执持久化并原样读回；审校不改写原文"],
|
||
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
|
||
)
|
||
def test_全文报告由实际逐片回执持久化并经CLI原样读回__22c005(全文报告环境, tmp_path):
|
||
from muse.任务运行.接口 import 任务状态
|
||
from muse.接入.cli.审校命令 import 运行审校命令
|
||
|
||
环境 = 全文报告环境
|
||
原正文 = 环境["审校正文"]
|
||
任务ID, 分片 = 发起全文审校(环境)
|
||
环境["剧本"].extend({"类型": "文本", "文本": 合成片结果(片, 有声音=False)} for 片 in 分片)
|
||
assert 跑全文审校(环境, 任务ID).状态 == 任务状态.已完成
|
||
服务读回 = (
|
||
环境["装配"].要求审校().读取文学工作面(环境["审校作者"], 环境["装配"].任务运行, 任务ID)
|
||
)
|
||
_配置审校CLI(环境["装配"], 环境["审校作者"], tmp_path)
|
||
assert 运行审校命令(环境["装配"], "报告", 任务ID) == 服务读回
|
||
assert 服务读回["report"]["persisted"] and 服务读回["report"]["model_verified"]
|
||
assert len(服务读回["report"]["evidence"]) == len(分片)
|
||
assert 环境["正文"].读取正文(环境["审校作者"], "ch-2") == 原正文
|