d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
63 lines
3.0 KiB
Python
63 lines
3.0 KiB
Python
"""隔离PG与合成HTTP验收原调用复用；需外部环境授权，绝不调用真实模型。"""
|
||
|
||
from uuid import UUID
|
||
|
||
import pytest
|
||
import test_评测执行与失败收敛 as 运行测试
|
||
|
||
from muse.效果评测.接口 import 评测错误
|
||
|
||
# 只用隔离PG与合成HTTP：不挂 网络 标记，否则数据库测试与验收数据库都够不到它。
|
||
pytestmark = [pytest.mark.数据库]
|
||
执行环境 = 运行测试.执行环境
|
||
|
||
|
||
@pytest.mark.case_id("NC-O03-REUSE-PERSISTED")
|
||
def test_原调用复用保留来源与费用归属且停止后拒读(执行环境):
|
||
env = 执行环境
|
||
service = env["app"].要求评测()
|
||
req = env["request"].model_copy(update={"effect_dependency_version": "effect-dependency-v2"})
|
||
source = service.创建实验(env["actor"], "reuse-source", req)
|
||
source_env = {**env, "exp": source, "request": req}
|
||
运行测试._启动执行(source_env)
|
||
运行测试._运行就绪(source_env)
|
||
service.推进实验(env["actor"], source["experiment_id"])
|
||
运行测试._运行就绪(source_env)
|
||
source_report = service.读取实验报告(env["actor"], source["experiment_id"])
|
||
assert source_report["runtime"]["units"] == 3
|
||
assert source_report["cost"]["sent_calls"] == 3
|
||
old_call_ids = {
|
||
call["call_id"]
|
||
for sample in source_report["samples"]
|
||
for unit in sample["units"]
|
||
for call in unit["calls"]
|
||
}
|
||
before = len(env["received"])
|
||
newer_req = req.model_copy(update={"reuse_experiments": (UUID(source["experiment_id"]),)})
|
||
target = service.创建实验(env["actor"], "reuse-target", newer_req)
|
||
target_env = {**env, "exp": target, "request": newer_req}
|
||
运行测试._启动执行(target_env)
|
||
运行测试._运行就绪(target_env)
|
||
service.推进实验(env["actor"], target["experiment_id"])
|
||
运行测试._运行就绪(target_env)
|
||
work = service.读取执行工作面(env["actor"], target["experiment_id"])
|
||
assert work["state"] == "completed"
|
||
writers = [u for u in work["units"] if u["kind"] == "generation"]
|
||
assert len(writers) == 2 and all(u.get("reuse") for u in writers)
|
||
assert all(u["cost"] is None for u in writers)
|
||
assert all(u["reuse"]["source_call_id"] in old_call_ids for u in writers)
|
||
report = service.读取实验报告(env["actor"], target["experiment_id"])
|
||
# 新盲化方向相同时评审也可复用；方向不同必须发一次新的独立请求。
|
||
new_calls = len(env["received"]) - before
|
||
assert new_calls in {0, 1}
|
||
assert report["cost"]["sent_calls"] == new_calls
|
||
assert all(
|
||
not u["calls"] for sample in report["samples"] for u in sample["units"] if u.get("reuse")
|
||
)
|
||
assert (
|
||
service.读取实验报告(env["actor"], source["experiment_id"])["cost"] == source_report["cost"]
|
||
)
|
||
service.取消实验(env["actor"], source["experiment_id"], "stop-reuse-source")
|
||
with pytest.raises(评测错误, match="原实验已停止"):
|
||
service.读取执行工作面(env["actor"], target["experiment_id"])
|