260 lines
10 KiB
Python
260 lines
10 KiB
Python
"""B07规则改进只登记意向；B10验证与B06审阅、启停、消费保持各自owner。"""
|
||
|
||
from dataclasses import replace
|
||
from datetime import UTC, datetime, timedelta
|
||
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.数据库
|
||
规则资格环境 = 规则测试.规则资格环境
|
||
|
||
|
||
def _规则改进(env):
|
||
author = env["author"]
|
||
service = env["app"].要求作者经验()
|
||
body = env["app"].要求正文()
|
||
candidate = body.创建人工候选(
|
||
author,
|
||
"rule-feedback-candidate",
|
||
"rule-ch",
|
||
env["target"]["revision"],
|
||
正文草稿((段落("p1", (文本节点("值得注意的是，门外的雨已经停了。"),)),)),
|
||
)["results"][0]
|
||
with env["pools"][用途.生产].连接(只读=True) as conn:
|
||
changed = 读取候选依据(conn, author.作者, candidate["candidate_id"], 1)
|
||
feedback = service.记录反馈(
|
||
author,
|
||
"rule-improvement-feedback",
|
||
反馈请求(
|
||
"rule-work",
|
||
反馈版本(
|
||
"document",
|
||
"rule-ch",
|
||
env["target"]["revision"],
|
||
env["target"]["document_hash"],
|
||
),
|
||
反馈版本("candidate", candidate["candidate_id"], 1, changed["document_hash"]),
|
||
"这处提示语需要识别，但不能改掉人物明确保留的口头禅。",
|
||
"revise",
|
||
),
|
||
)
|
||
version = env["review"].读取规则版本(author, "l002", 1)
|
||
request = 规则改进请求(
|
||
improvement_id=str(uuid4()),
|
||
expected_revision=0,
|
||
title="校准提示语规则",
|
||
rationale="依据实际正文反馈补足正例、负例与保护表达验证。",
|
||
rule_id="l002",
|
||
rule_version=1,
|
||
rule_hash=固定哈希(version["payload"]["rule"]),
|
||
feedback_ids=(str(feedback["feedback_id"]),),
|
||
)
|
||
row = service.提出规则改进(author, request)
|
||
point = 改进定位(request.improvement_id, row["revision"], row["content_hash"])
|
||
return service, request, point
|
||
|
||
|
||
def _启停请求(action, proposal, opened, revision, credential_id=""):
|
||
return 规则启停请求(
|
||
action=action,
|
||
rule_id=proposal.rule_id,
|
||
rule_version=proposal.rule_version,
|
||
rule_hash=proposal.rule_hash,
|
||
expected_activation_revision=revision,
|
||
review_id=opened["review_id"],
|
||
review_hash=opened["review_hash"],
|
||
credential_id=credential_id,
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w26-26d101",
|
||
environment="隔离PG/合成HTTP及浏览器",
|
||
given="固定请求、来源与作者；按用例环境准备独立测试输入",
|
||
when="通过实际公开入口执行并核对拒绝、状态及持久结果",
|
||
then=["固定规则验证到B06启停消费及原回执恢复"],
|
||
contract="docs/系统架构/新版设计/模块设计/B07-作者经验.md#有来源的运行观察",
|
||
)
|
||
def test_固定规则验证到B06启停消费及原回执恢复__26d101(规则资格环境, monkeypatch):
|
||
env = 规则资格环境
|
||
service, proposal, point = _规则改进(env)
|
||
published = 规则测试._publish(env, "b07-owned-rule-improvement", "owned")
|
||
direct = env["evaluate"].执行规则诊断验证(env["evaluation_identity"], published["version_id"])
|
||
proof = env["maintain"].导出规则诊断凭据(
|
||
env["maintenance_identity"],
|
||
direct["run_id"],
|
||
批准引用="rule-improvement-owner-proof",
|
||
有效期=datetime.now(UTC) + timedelta(hours=1),
|
||
)
|
||
command = 规则状态命令(
|
||
proposal.rule_id,
|
||
proposal.rule_version,
|
||
"enable",
|
||
expected_activation_revision=0,
|
||
credential_id=proof["receipt_id"],
|
||
)
|
||
opened = env["review"].打开规则状态审阅(env["author"], command)
|
||
enable = _启停请求("enable", proposal, opened, 0, proof["receipt_id"])
|
||
with pytest.raises(Muse错误, match="尚未登记同一目标"):
|
||
service.请求规则启停(env["author"], "enable-before-validation", point, enable)
|
||
with pytest.raises(Muse错误, match="审阅身份"):
|
||
replace(enable, review_id="")
|
||
|
||
validation = service.申请规则验证(
|
||
env["author"],
|
||
"validate-rule-improvement",
|
||
point,
|
||
env["evaluate"],
|
||
env["evaluation_identity"],
|
||
规则验证请求(
|
||
dataset_version_id=published["version_id"],
|
||
rule_id=proposal.rule_id,
|
||
rule_version=proposal.rule_version,
|
||
rule_hash=proposal.rule_hash,
|
||
),
|
||
)
|
||
assert set(validation["receipt"]) == {"run_id", "target", "result_hash"}
|
||
assert validation["receipt"]["run_id"] == direct["run_id"]
|
||
assert validation["receipt"]["target"] == published["target"]
|
||
assert "experiment_id" not in validation["receipt"]
|
||
assert (
|
||
service.申请规则验证(
|
||
env["author"],
|
||
"validate-rule-improvement",
|
||
point,
|
||
env["evaluate"],
|
||
env["evaluation_identity"],
|
||
规则验证请求(
|
||
published["version_id"],
|
||
proposal.rule_id,
|
||
proposal.rule_version,
|
||
proposal.rule_hash,
|
||
),
|
||
)["receipt"]
|
||
== validation["receipt"]
|
||
)
|
||
|
||
save_result = 经验存储.保存改进结果
|
||
failures = {"enable-rule": 1}
|
||
|
||
def interrupt_once(self, author, command_id, receipt):
|
||
if failures.get(command_id, 0):
|
||
failures[command_id] -= 1
|
||
raise RuntimeError("B06已经提交，B07登记前中断")
|
||
return save_result(self, author, command_id, receipt)
|
||
|
||
monkeypatch.setattr(经验存储, "保存改进结果", interrupt_once)
|
||
with pytest.raises(RuntimeError, match="登记前中断"):
|
||
service.请求规则启停(env["author"], "enable-rule", point, enable)
|
||
active = env["review"].读取规则状态(env["author"], proposal.rule_id)
|
||
assert active["current"]["state"] == "active"
|
||
pending = next(
|
||
action
|
||
for action in service.读取改进(env["author"], proposal.improvement_id)["actions"]
|
||
if action["kind"] == "enable"
|
||
)
|
||
assert pending["receipt"] is None
|
||
original = service.正式.读取回执(env["author"], "B07.owner:" + str(pending["action_id"]))
|
||
|
||
diagnosis = env["review"].生成诊断(env["author"], env["target"])
|
||
assert (
|
||
diagnosis["report"]["rule_bindings"][0]["version_id"] == published["target"]["version_id"]
|
||
)
|
||
env["evaluate"].停止规则诊断验证(
|
||
env["evaluation_identity"], direct["run_id"], "stop-after-real-consumption"
|
||
)
|
||
recovered = service.请求规则启停(env["author"], "enable-rule", point, enable)
|
||
assert recovered["receipt"] == original
|
||
|
||
disable_command = 规则状态命令(
|
||
proposal.rule_id,
|
||
proposal.rule_version,
|
||
"disable",
|
||
expected_activation_revision=1,
|
||
)
|
||
disable_opened = env["review"].打开规则状态审阅(env["author"], disable_command)
|
||
disabled = service.请求规则启停(
|
||
env["author"],
|
||
"disable-rule",
|
||
point,
|
||
_启停请求("disable", proposal, disable_opened, 1),
|
||
)
|
||
assert disabled["receipt"]["results"][0]["state"] == "disabled"
|
||
detail = service.读取改进(env["author"], proposal.improvement_id)
|
||
assert detail["target_state"] == "disabled"
|
||
assert detail["rule_state"]["current"]["activation_revision"] == 2
|
||
assert len(detail["consumption"]) == 1
|
||
assert detail["consumption"][0]["diagnosis_id"] == diagnosis["diagnosis_id"]
|
||
assert detail["consumption"][0]["binding"]["credential_id"] == proof["receipt_id"]
|
||
with pytest.raises(Muse错误, match="凭据不可用|凭据.*失效"):
|
||
env["review"].打开规则状态审阅(
|
||
env["author"],
|
||
规则状态命令(
|
||
proposal.rule_id,
|
||
proposal.rule_version,
|
||
"enable",
|
||
expected_activation_revision=2,
|
||
credential_id=proof["receipt_id"],
|
||
),
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w26-26d102",
|
||
environment="隔离PG/合成HTTP及浏览器",
|
||
given="固定请求、来源与作者；按用例环境准备独立测试输入",
|
||
when="通过实际公开入口执行并核对拒绝、状态及持久结果",
|
||
then=["异版他作者与空反馈不产生规则改进行动"],
|
||
contract="docs/系统架构/新版设计/模块设计/B07-作者经验.md#有来源的运行观察",
|
||
)
|
||
def test_异版他作者与空反馈不产生规则改进行动__26d102(规则资格环境):
|
||
env = 规则资格环境
|
||
service, proposal, point = _规则改进(env)
|
||
published = 规则测试._publish(env, "b07-rule-negative", "owned")
|
||
with pytest.raises(Muse错误, match="不是请求点名"):
|
||
service.申请规则验证(
|
||
env["author"],
|
||
"wrong-rule-version",
|
||
point,
|
||
env["evaluate"],
|
||
env["evaluation_identity"],
|
||
规则验证请求(published["version_id"], "l002", 2, proposal.rule_hash),
|
||
)
|
||
with pytest.raises(Muse错误, match="非空"):
|
||
replace(proposal, improvement_id=str(uuid4()), feedback_ids=())
|
||
stranger = replace(env["author"], 作者="other-author")
|
||
with pytest.raises(Muse错误):
|
||
service.提出规则改进(stranger, replace(proposal, improvement_id=str(uuid4())))
|
||
with pytest.raises(Muse错误):
|
||
service.读取改进(stranger, proposal.improvement_id)
|
||
first = service.读取改进(env["author"], proposal.improvement_id)
|
||
assert first["actions"] == []
|
||
revised = service.提出规则改进(
|
||
env["author"],
|
||
replace(
|
||
proposal,
|
||
expected_revision=1,
|
||
rationale="作者补充了保护表达必须保留的改进理由。",
|
||
),
|
||
)
|
||
assert revised["revision"] == 2
|
||
historical = service.读取改进(env["author"], proposal.improvement_id, revision=1)
|
||
assert historical["content_hash"] == point.content_hash
|