d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
"""榜单快照：具名采集、失败保留失败态、观察差异不推断因果。"""
|
|
|
|
import pytest
|
|
|
|
from muse.资料研究.接口 import 资料服务
|
|
from muse.资料研究.模型 import 榜单读取结果, 资料错误
|
|
|
|
pytestmark = pytest.mark.数据库
|
|
|
|
|
|
class 合成榜单读取器:
|
|
"""按剧本返回的合成读取器；不访问网络。"""
|
|
|
|
def __init__(self, 剧本: list[榜单读取结果]):
|
|
self.剧本 = list(剧本)
|
|
|
|
def 抓取(self, url: str, 方式: str) -> 榜单读取结果:
|
|
return self.剧本.pop(0)
|
|
|
|
|
|
@pytest.fixture
|
|
def 服务():
|
|
return 资料服务()
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w17-w17b01",
|
|
environment="真实隔离 PG",
|
|
given="隔离数据库",
|
|
when="失败保留失败态不伪造；成功快照带条目；观察差异不含 run 归因",
|
|
then=["失败态保留", "快照可查", "差异为观察陈述"],
|
|
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
|
|
)
|
|
def test_采集成功失败与观察差异__17b01e(服务, 应用测试库):
|
|
from muse.共享.调用身份 import 用途
|
|
|
|
库 = 应用测试库[用途.生产]
|
|
with 库.连接() as 连, 连.transaction():
|
|
服务.登记榜单(连, "rank-qd", "起点月票榜", "https://example.com/rank", "html")
|
|
# 第一次：失败保留失败态，不伪造数据。
|
|
失败 = 服务.采集榜单快照(
|
|
连, "rank-qd", 合成榜单读取器([榜单读取结果(成功=False, 失败原因="上游 503")])
|
|
)
|
|
assert 失败["status"] == "failed" and 失败["failure_reason"] == "上游 503"
|
|
快照集 = 服务.列出快照(连, "rank-qd")
|
|
assert 快照集[0]["status"] == "failed" and 快照集[0]["items"] is None
|
|
# 第二、三次：成功快照带时刻与条目。
|
|
第一 = 服务.采集榜单快照(
|
|
连,
|
|
"rank-qd",
|
|
合成榜单读取器(
|
|
[
|
|
榜单读取结果(
|
|
成功=True,
|
|
items=(
|
|
{"作品": "甲", "名次": 1},
|
|
{"作品": "乙", "名次": 2},
|
|
),
|
|
)
|
|
]
|
|
),
|
|
)
|
|
第二 = 服务.采集榜单快照(
|
|
连,
|
|
"rank-qd",
|
|
合成榜单读取器(
|
|
[
|
|
榜单读取结果(
|
|
成功=True,
|
|
items=(
|
|
{"作品": "乙", "名次": 1},
|
|
{"作品": "丙", "名次": 2},
|
|
{"作品": "甲", "名次": 3},
|
|
),
|
|
)
|
|
]
|
|
),
|
|
)
|
|
assert 第一["status"] == "ok" and 第二["条目数"] == 3
|
|
差异 = 服务.快照观察差异(连, "rank-qd")
|
|
变化 = {d["作品"]: d["变化"] for d in 差异["差异"]}
|
|
assert 变化["乙"] == 1 and 变化["甲"] == -2 and 变化["丙"] is None
|
|
# 观察差异不携带任何 run/效果归因字段。
|
|
assert all("run" not in str(键).lower() for 键 in 差异)
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w17-w17b02",
|
|
environment="真实隔离 PG",
|
|
given="隔离数据库",
|
|
when="未登记的榜种不能采集",
|
|
then=["未登记拒绝"],
|
|
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
|
|
)
|
|
def test_未登记榜单拒绝采集__17b02f(服务, 应用测试库):
|
|
from muse.共享.调用身份 import 用途
|
|
|
|
库 = 应用测试库[用途.生产]
|
|
with 库.连接() as 连:
|
|
with pytest.raises(资料错误, match="榜单未登记"):
|
|
服务.采集榜单快照(连, "rank-none", 合成榜单读取器([]))
|