164 lines
6.0 KiB
Python
164 lines
6.0 KiB
Python
"""全书工作面批量读取与归属校验的独立机制验证。"""
|
||
|
||
from dataclasses import asdict
|
||
|
||
import pytest
|
||
|
||
from muse.交付连载.全书整理 import 整理全书工作面
|
||
from muse.作品规划.模型 import 作品错误
|
||
from muse.正文写作.模型 import 正文草稿, 正文错误, 段落
|
||
from muse.正文写作.正文格式 import 可见文本哈希, 正文结构哈希
|
||
|
||
|
||
class 模拟连:
|
||
def __init__(self, chapters, *, author="author-1", work_id="work-1", missing=(), corrupt=()):
|
||
self.chapters = chapters
|
||
self.author = author
|
||
self.work_id = work_id
|
||
self.missing = set(missing)
|
||
self.corrupt = set(corrupt)
|
||
self.queries = []
|
||
|
||
def cursor(self, **_):
|
||
return self
|
||
|
||
def execute(self, sql, params=()):
|
||
self.queries.append((sql, params))
|
||
if "FROM muse_work w" in sql or "c.chapter_id=ANY(%s)" in sql:
|
||
# 批量读取章节归属
|
||
ids = params[0]
|
||
rows = []
|
||
for cid in ids:
|
||
if cid.startswith("nonexistent"):
|
||
continue
|
||
a = "wrong-author" if cid.startswith("wrong-author") else self.author
|
||
w = "wrong-work" if cid.startswith("wrong-work") else self.work_id
|
||
rows.append(
|
||
{
|
||
"chapter_id": cid,
|
||
"work_id": w,
|
||
"title": f"Title {cid}",
|
||
"position": 1,
|
||
"author_id": a,
|
||
"directory_revision": 1,
|
||
}
|
||
)
|
||
self._rows = rows
|
||
return self
|
||
|
||
if "FROM muse_document d JOIN muse_document_version v" in sql:
|
||
# 批量读取当前正文
|
||
ids = params[0]
|
||
branch = params[1]
|
||
rows = []
|
||
for cid in ids:
|
||
if cid in self.missing:
|
||
continue
|
||
draft = 正文草稿((段落(f"{cid}:p0", ()),))
|
||
doc_hash = "bad_hash" if cid in self.corrupt else 正文结构哈希(draft)
|
||
vis_hash = 可见文本哈希(draft)
|
||
rows.append(
|
||
{
|
||
"document_id": f"doc-{cid}",
|
||
"chapter_id": cid,
|
||
"branch_id": branch,
|
||
"revision": 1,
|
||
"document": asdict(draft),
|
||
"document_hash": doc_hash,
|
||
"visible_text_hash": vis_hash,
|
||
"restored_from": None,
|
||
}
|
||
)
|
||
self._rows = rows
|
||
return self
|
||
|
||
self._rows = []
|
||
return self
|
||
|
||
def fetchall(self):
|
||
return getattr(self, "_rows", [])
|
||
|
||
def fetchone(self):
|
||
rows = getattr(self, "_rows", [])
|
||
return rows[0] if rows else None
|
||
|
||
|
||
@pytest.mark.case_id("TC-DELIVERY-BATCH-AUTH-CHECK")
|
||
def test_批量核对章节归属异常分支():
|
||
from muse.作品规划.接口 import 批量核对章节归属
|
||
|
||
# 1. 空列表直接返回
|
||
连 = 模拟连([])
|
||
assert 批量核对章节归属(连, "author-1", []) == {}
|
||
|
||
# 2. 章节不存在整批拒绝
|
||
连 = 模拟连([])
|
||
with pytest.raises(作品错误) as exc:
|
||
批量核对章节归属(连, "author-1", ["c1", "nonexistent-1"])
|
||
assert exc.value.错误码 == "CHAPTER_NOT_FOUND"
|
||
|
||
# 3. 章节不属于当前作者整批拒绝
|
||
with pytest.raises(作品错误) as exc:
|
||
批量核对章节归属(连, "author-1", ["c1", "wrong-author-c2"])
|
||
assert exc.value.错误码 == "SCOPE_DENIED"
|
||
assert "章节不属于当前作者" in exc.value.说明
|
||
|
||
# 4. 章节不属于指定作品整批拒绝
|
||
with pytest.raises(作品错误) as exc:
|
||
批量核对章节归属(连, "author-1", ["c1", "wrong-work-c2"], work_id="work-1")
|
||
assert exc.value.错误码 == "SCOPE_DENIED"
|
||
assert "章节不属于指定作品" in exc.value.说明
|
||
|
||
|
||
@pytest.mark.case_id("TC-DELIVERY-BATCH-CORRUPT-DETECT")
|
||
def test_批量读取正文损坏则报错():
|
||
from muse.正文写作.接口 import 批量读取当前正文依据
|
||
|
||
连 = 模拟连(["c1"], corrupt=["c1"])
|
||
with pytest.raises(正文错误, match="已保存正文与其结构或文本哈希不一致"):
|
||
批量读取当前正文依据(连, "author-1", "work-1", ["c1"])
|
||
|
||
|
||
@pytest.mark.case_id("TC-DELIVERY-BATCH-READ-WORKBENCH")
|
||
def test_全书批量读取查询次数为常数上界():
|
||
from muse.正文写作.接口 import 批量读取当前正文依据
|
||
|
||
# 100 章批量获取，SQL 查询次数恒定为 1 次批量归属 + 1 次批量正文
|
||
c_ids = [f"ch-{i}" for i in range(100)]
|
||
连 = 模拟连(c_ids)
|
||
结果 = 批量读取当前正文依据(连, "author-1", "work-1", c_ids)
|
||
assert len(结果) == 100
|
||
assert len(连.queries) == 2 # 1次批量核对归属 + 1次批量正文读取，而非 200 次！
|
||
|
||
|
||
@pytest.mark.case_id("TC-DELIVERY-WORKBENCH-MEM-SEPARATED")
|
||
def test_全书工作面组装正确区分状态():
|
||
# 模拟包含正常章、缺正文章、空正文章
|
||
directory = {
|
||
"revision": 5,
|
||
"chapters": [
|
||
{"chapter_id": "c1", "title": "第一章", "position": 1},
|
||
{"chapter_id": "c2", "title": "第二章", "position": 2},
|
||
],
|
||
}
|
||
# c1 正常，c2 缺失
|
||
draft = 正文草稿((段落("c1:p0", ()),))
|
||
body_map = {
|
||
"c1": {
|
||
"work_id": "work-1",
|
||
"document_id": "doc-c1",
|
||
"chapter_id": "c1",
|
||
"revision": 1,
|
||
"current_revision": 1,
|
||
"document_hash": 正文结构哈希(draft),
|
||
"document": draft,
|
||
}
|
||
}
|
||
res = 整理全书工作面("work-1", directory, body_map)
|
||
assert res["work_id"] == "work-1"
|
||
assert res["directory_revision"] == 5
|
||
assert len(res["chapters"]) == 2
|
||
assert res["chapters"][0]["state"] == "empty" # 空内容
|
||
assert res["chapters"][1]["state"] == "missing" # 缺正文
|
||
assert not res["can_freeze"] # 有 problem 不能冻结
|