103 lines
4.5 KiB
Python
103 lines
4.5 KiB
Python
"""资料导入与原文版本：幂等去重、版本链、证据校验、用途授权与清理对照。"""
|
|
|
|
import pytest
|
|
|
|
from muse.共享.调用身份 import 用途
|
|
from muse.资料研究.接口 import 资料服务
|
|
from muse.资料研究.模型 import 导入请求, 资料错误
|
|
|
|
pytestmark = pytest.mark.数据库
|
|
|
|
|
|
@pytest.fixture
|
|
def 服务(应用测试库):
|
|
return 资料服务()
|
|
|
|
|
|
def test_导入去重与版本链__17a01a(服务, 应用测试库):
|
|
库 = 应用测试库[用途.生产]
|
|
请求 = 导入请求(
|
|
kind="reference",
|
|
title="封神演义",
|
|
origin="file:///资料/封神演义.txt",
|
|
content="第一回 纣王女娲宫进香。\n\n最新章节请访问 example.com\n",
|
|
authorized_uses=("analysis",),
|
|
)
|
|
with 库.连接() as 连, 连.transaction():
|
|
首次 = 服务.导入(连, "importer", 请求)
|
|
assert 首次.重复 is False and 首次.revision == 1
|
|
# 同源同内容：幂等返回既有版本，重复可辨认。
|
|
再次 = 服务.导入(连, "importer", 请求)
|
|
assert 再次.重复 is True and 再次.revision == 1 and 再次.source_id == 首次.source_id
|
|
# 同源异内容：追加新版本，旧版本仍可回读。
|
|
新版 = 服务.导入(
|
|
连,
|
|
"importer",
|
|
导入请求(
|
|
kind="reference",
|
|
title="封神演义（校对版）",
|
|
origin=请求.origin,
|
|
content="第一回 纣王女娲宫进香。（校对）\n",
|
|
authorized_uses=("analysis", "reference_reading"),
|
|
),
|
|
)
|
|
assert 新版.重复 is False and 新版.revision == 2 and 新版.source_id == 首次.source_id
|
|
版本 = 服务.读取版本(连, 首次.source_id, 1)
|
|
assert "最新章节" in 版本["content"]
|
|
# 授权用途只增不减。
|
|
源 = 服务.核对来源授权(连, 首次.source_id, "reference_reading")
|
|
assert 源["authorized_uses"] == ["analysis", "reference_reading"]
|
|
|
|
|
|
def test_证据读取与哈希校验__17a02b(服务, 应用测试库):
|
|
库 = 应用测试库[用途.生产]
|
|
文本 = "林深把伞收在了门边。雨还没停。"
|
|
with 库.连接() as 连, 连.transaction():
|
|
结果 = 服务.导入(
|
|
连,
|
|
"importer",
|
|
导入请求("note", "码头笔记", "note://local/1", 文本, ("analysis",)),
|
|
)
|
|
证据 = 服务.读取证据(连, 结果.source_id, 结果.revision, 0, 10)
|
|
assert 证据["片段"] == "林深把伞收在了门边。"
|
|
with pytest.raises(资料错误, match="区间越界"):
|
|
服务.读取证据(连, 结果.source_id, 结果.revision, 0, 999)
|
|
|
|
|
|
def test_用途未授权拒绝消费__17a03c(服务, 应用测试库):
|
|
库 = 应用测试库[用途.生产]
|
|
with 库.连接() as 连, 连.transaction():
|
|
结果 = 服务.导入(
|
|
连,
|
|
"importer",
|
|
导入请求("web", "某网页", "https://example.com/a", "网页内容", ("reference_reading",)),
|
|
)
|
|
with pytest.raises(资料错误, match="未授权用途"):
|
|
服务.核对来源授权(连, 结果.source_id, "analysis")
|
|
|
|
|
|
def test_清理候选对照且原文不变__17a04d(服务, 应用测试库):
|
|
库 = 应用测试库[用途.生产]
|
|
原文 = "第一回 纣王女娲宫进香。\n最新章节请访问 example.com\n本章完。"
|
|
with 库.连接() as 连, 连.transaction():
|
|
结果 = 服务.导入(
|
|
连,
|
|
"importer",
|
|
导入请求("reference", "样章", "file:///样章.txt", 原文, ("analysis",)),
|
|
)
|
|
候选 = 服务.生成清理候选(连, 结果.source_id, 结果.revision)
|
|
assert 候选.removals and all(项["理由"] for 项 in 候选.removals)
|
|
应用后 = 服务.应用清理候选(连, 候选.candidate_id)
|
|
assert "最新章节" not in 应用后.cleaned_text
|
|
assert "第一回" in 应用后.cleaned_text
|
|
# 原文版本不受清理影响。
|
|
版本 = 服务.读取版本(连, 结果.source_id, 结果.revision)
|
|
assert 版本["content"] == 原文
|
|
# 重复应用与处置不可改写。
|
|
with pytest.raises(资料错误, match="已处置"):
|
|
服务.应用清理候选(连, 候选.candidate_id)
|
|
新候选 = 服务.生成清理候选(连, 结果.source_id, 结果.revision)
|
|
服务.拒绝清理候选(连, 新候选.candidate_id)
|
|
with pytest.raises(资料错误, match="已处置"):
|
|
服务.应用清理候选(连, 新候选.candidate_id)
|