d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
"""交付文件排他发布失败恢复与榜单数据库时刻读回。"""
|
|
|
|
import os
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from muse.共享.错误 import 配置错误
|
|
from muse.接入.cli.交付命令 import _排他发布
|
|
from muse.资料研究.存储 import 资料存储
|
|
from muse.资料研究.榜单快照 import 采集榜单快照
|
|
|
|
|
|
@pytest.mark.case_id("NC-O10-008")
|
|
def test_完整文件才排他发布且不覆盖旧目标(tmp_path):
|
|
path = tmp_path / "成品.txt"
|
|
_排他发布(path, b"complete")
|
|
assert path.read_bytes() == b"complete"
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
with pytest.raises(配置错误):
|
|
_排他发布(path, b"replacement")
|
|
assert path.read_bytes() == b"complete"
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
@pytest.mark.case_id("NC-O10-009")
|
|
def test_写入失败不留下残目标并允许重试(tmp_path, monkeypatch):
|
|
path = tmp_path / "成品.txt"
|
|
real_sync = os.fsync
|
|
|
|
def fail(fd):
|
|
raise OSError("合成写盘失败")
|
|
|
|
monkeypatch.setattr(os, "fsync", fail)
|
|
with pytest.raises(配置错误):
|
|
_排他发布(path, b"partial")
|
|
assert list(tmp_path.iterdir()) == []
|
|
monkeypatch.setattr(os, "fsync", real_sync)
|
|
_排他发布(path, b"complete")
|
|
assert path.read_bytes() == b"complete"
|
|
|
|
|
|
@pytest.mark.case_id("NC-O10-010")
|
|
def test_榜单采集响应与列表复用数据库真实时刻():
|
|
captured = datetime(2026, 9, 17, 3, 4, 5, 678901, tzinfo=UTC)
|
|
queries, snapshots = [], []
|
|
|
|
class Connection:
|
|
def cursor(self, **kwargs):
|
|
return self
|
|
|
|
def execute(self, query, params):
|
|
queries.append(query)
|
|
if query.startswith("INSERT"):
|
|
snapshots.append(
|
|
{
|
|
"snapshot_id": params[0],
|
|
"ranking_id": params[1],
|
|
"captured_at": captured,
|
|
"status": params[2],
|
|
}
|
|
)
|
|
return self
|
|
|
|
def fetchone(self):
|
|
if queries[-1].startswith("INSERT"):
|
|
assert queries[-1].endswith("RETURNING captured_at")
|
|
return {"captured_at": captured}
|
|
return {"url": "https://example.invalid", "采集方式": "html"}
|
|
|
|
def fetchall(self):
|
|
return snapshots
|
|
|
|
conn = Connection()
|
|
reader = SimpleNamespace(
|
|
抓取=lambda *a: SimpleNamespace(成功=True, items=({"作品": "合成作品"},))
|
|
)
|
|
result = 采集榜单快照(conn, "ranking", reader)
|
|
listed = 资料存储(conn).列出快照("ranking")[0]
|
|
assert datetime.fromisoformat(result["captured_at"]) == listed["captured_at"] == captured
|
|
assert result["snapshot_id"] == listed["snapshot_id"]
|
|
assert result["条目数"] == 1
|