d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
"""真实进度组装与owner查询经合成游标验证固定借用和SQL上界。"""
|
|
|
|
from contextlib import contextmanager
|
|
|
|
import pytest
|
|
|
|
from muse.任务运行.接口 import 任务服务
|
|
from muse.作品规划.接口 import 作品服务, 创作进度服务
|
|
from muse.共享.标识 import 任务标识
|
|
from muse.共享.调用身份 import 内容用途, 用途, 调用身份
|
|
from muse.正式变更.接口 import 参与者目录, 正式变更服务
|
|
from muse.正文写作.接口 import 正文服务
|
|
from muse.编排.接口 import 流程登记
|
|
|
|
|
|
class 合成连接:
|
|
def __init__(self, count):
|
|
self.chapters = [
|
|
{"chapter_id": f"c-{i}", "title": f"章{i}", "position": i + 1} for i in range(count)
|
|
]
|
|
self.sql = []
|
|
self.rows = []
|
|
|
|
def cursor(self, **_):
|
|
return self
|
|
|
|
def execute(self, sql, params=()):
|
|
query = sql if isinstance(sql, str) else sql.as_string()
|
|
self.sql.append(query)
|
|
if query.startswith("SET TRANSACTION"):
|
|
self.rows = []
|
|
elif "SELECT w.*" in query:
|
|
self.rows = [{"author_id": "author"}]
|
|
elif "SELECT v.revision,v.chapters" in query:
|
|
self.rows = [(3, self.chapters)]
|
|
elif "AS visible_characters" in query:
|
|
self.rows = [
|
|
{
|
|
"chapter_id": c["chapter_id"],
|
|
"revision": 2,
|
|
"document_hash": "synthetic",
|
|
"visible_characters": 12,
|
|
}
|
|
for c in self.chapters
|
|
]
|
|
elif "AS stale" in query:
|
|
self.rows = [
|
|
{
|
|
"chapter_id": c["chapter_id"],
|
|
"candidate_id": "candidate-" + c["chapter_id"],
|
|
"revision": 1,
|
|
"stale": True,
|
|
}
|
|
for c in self.chapters
|
|
]
|
|
elif "muse_candidate_decision" in query:
|
|
assert params[0] == "author"
|
|
self.rows = []
|
|
elif "AS event_cursor" in query:
|
|
assert params == ("production", "author", "work")
|
|
self.rows = [{"task_id": "task", "state": "succeeded", "event_cursor": 7}]
|
|
else:
|
|
raise AssertionError(query)
|
|
return self
|
|
|
|
def fetchall(self):
|
|
return self.rows
|
|
|
|
def fetchone(self):
|
|
return self.rows[0] if self.rows else None
|
|
|
|
|
|
@pytest.mark.case_id("NC-O06-PROGRESS-CONSTANT")
|
|
@pytest.mark.parametrize("count", [0, 1, 500])
|
|
def test_进度完整组装恒一次借用且最多八次业务查询(count):
|
|
conn = 合成连接(count)
|
|
|
|
class Database:
|
|
用途 = 用途.生产
|
|
borrows = active = 0
|
|
|
|
@contextmanager
|
|
def 连接(self, *, 只读):
|
|
assert 只读
|
|
self.borrows += 1
|
|
self.active += 1
|
|
try:
|
|
yield conn
|
|
finally:
|
|
self.active -= 1
|
|
|
|
db = Database()
|
|
formal = 正式变更服务(db, 参与者目录())
|
|
service = 创作进度服务(作品服务(db, formal), 正文服务(db, formal), 任务服务(db, 流程登记()))
|
|
actor = 调用身份("author", 任务标识("progress"), 用途.生产, 内容用途.规划)
|
|
result = service.查询(actor, "work")
|
|
assert db.borrows == 1 and db.active == 0
|
|
assert conn.sql[0] == "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY"
|
|
assert len(conn.sql) - 1 <= 8
|
|
assert (result["total_chapters"], result["saved_chapters"], result["undecided_candidates"]) == (
|
|
count,
|
|
count,
|
|
count,
|
|
)
|
|
assert result["tasks"] == [{"task_id": "task", "state": "succeeded", "event_cursor": 7}]
|
|
for chapter in result["chapters"]:
|
|
assert chapter["visible_characters"] == 12
|
|
assert chapter["candidates"][0]["stale"]
|
|
assert chapter["candidates"][0]["decision"] == "undecided"
|
|
|
|
|
|
@pytest.mark.case_id("NC-O06-PROGRESS-AUTH-OPTIMIZED")
|
|
def test_优化模式下缺作者身份仍显式拒绝():
|
|
import subprocess
|
|
import sys
|
|
|
|
code = """
|
|
from muse.作品规划.进度查询 import 创作进度服务
|
|
from muse.作品规划.模型 import 作品错误
|
|
from muse.共享.调用身份 import 调用身份, 用途, 内容用途
|
|
identity = 调用身份(None, None, 用途.生产, 内容用途.规划)
|
|
try:
|
|
创作进度服务(None, None, None).查询(identity, 'work')
|
|
except 作品错误 as exc:
|
|
if exc.错误码 != 'SCOPE_DENIED':
|
|
raise RuntimeError(exc.错误码)
|
|
else:
|
|
raise RuntimeError('missing author was accepted')
|
|
"""
|
|
result = subprocess.run([sys.executable, "-O", "-c", code], capture_output=True, text=True)
|
|
assert result.returncode == 0, result.stderr
|