muse-agent-example/tests/迁移/test_旧载荷归一.py
zizi cf0f4fb985 W20–W24:保存知识方法、迁移框架、审校修订与案例行为的集成成果
接续 88dd570,保存 W20–W24 已实现的共享接口、业务入口、迁移、工作台、测试与文档。
W20/W22/W23 保持 in_progress,W21/W24 保持 verified;此提交不宣称方法或规则正式启用、多轮返修、真实角色评测完成。

W25 新增实验、标定、逐调用交付与角色执行及其迁移/测试/索引留在实施工作树,原有私人和旧实现保留项不纳入。

验证:离线 571、前端 33 通过;PG 469 项通过、2 项浏览器未启用,2 项误带入的 W25 用例已移出本提交;最终任务与交付边界 37 项通过。make 检查、最终类型、84 项资源及 diff 检查通过。未重跑浏览器或 Pi 宿主,不以合成调用认证外部模型效果。
独立整体审查四维通过;证据保存在 R2-20260909/提交W20-W24。
2026-09-13 16:24:42 +08:00

223 lines
7.7 KiB
Python

"""旧载荷形状、端点身份和来源声明的纯合同;不运行旧确认实现。"""
from copy import deepcopy
import pytest
from 建立映射 import 旧快照, 映射配置
from 知识记录分流 import 分流知识记录
def _输入(载荷=None, *, 关系=False):
源 = {
"source": {
"system": "pg",
"database": "synthetic-t5",
"table": "muse_knowledge_draft",
"id": "1",
"revision": "1",
},
"record": {
"work_id": 12,
"source_id": 17473,
"source_type": "chapter_extract",
"status": "pending",
"source_status": "active",
"source_action_policy": "allowed",
"draft_type": "relation" if 关系 else "entity",
"draft_payload": 载荷
if 载荷 is not None
else {"type": "character", "name": "林深", "fields": {"身份": "驾驶员"}},
},
}
域 = {"system": "pg", "database": "synthetic-t5"}
映射 = {
"works": [
{
**域,
"legacy_id": 12,
"use": "author",
"target_ref": "new-work",
"author_id": "author",
}
],
"sources": [
{
**域,
"legacy_id": 17473,
"source_type": "chapter_extract",
"work_id": 12,
"target_ref": "new-chapter",
"revision": "1",
"chapters": {"1": "chapter:new-chapter"},
}
],
"objects": [
{
**域,
"work_id": 12,
"use": "author",
"scope": "new-work",
"refs": refs,
"target_ref": id_,
}
for refs, id_ in (
(["name:林深", "name:11"], "a"),
(["entity:11"], "b"),
(["entity:12", "name:茧"], "c"),
)
],
"schemas": {
"character": {
"type": "object",
"properties": {"身份": {"type": "string"}},
"additionalProperties": False,
},
"character_relation": {
"type": "object",
"properties": {},
"additionalProperties": False,
},
},
}
return 源, 映射
def _分流(源, 映射):
快照 = 旧快照.从载荷(源)
结果 = 分流知识记录(快照, 映射配置(映射))
assert 结果.original == 快照.导出()
return 结果
@pytest.mark.parametrize(
"键,别名,值",
[
("brief", "description", "机师"),
("description", "描述", "机师"),
("fields", "attributes", {"身份": "驾驶员"}),
("fields", "字段", {"身份": "驾驶员"}),
("evidence", "证据", "证据句"),
],
)
def test_载荷别名同值承接冲突不吞原值__210001(键, 别名, 值):
源, 映射 = _输入()
载荷 = 源["record"]["draft_payload"]
载荷.update({键: 值, 别名: deepcopy(值)})
assert _分流(源, 映射).disposition == "待转换"
载荷[别名] = {"身份": "异值"} if isinstance(值, dict) else "异值"
assert _分流(源, 映射).reason_code == "field_conflict"
@pytest.mark.parametrize(
"值,原因",
[
(None, "field_type_mismatch"),
(True, "field_type_mismatch"),
([], "field_type_mismatch"),
({}, "missing_endpoint"),
({"id": 11}, "unknown_fields"),
({"name": " "}, "field_type_mismatch"),
({"name": "林深", "名称": "异名"}, "field_conflict"),
(" ", "field_type_mismatch"),
],
)
def test_端点坏形状保全且错误固定到侧__210002(值, 原因):
源, 映射 = _输入(
{"type": "驾驶", "source": 值, "target": {"name": "茧"}, "description": "不应出现在错误中"},
关系=True,
)
结果 = _分流(源, 映射)
assert 结果.disposition == "隔离" and 结果.reason_code == 原因
assert 结果.detail == "relation.source"
@pytest.mark.parametrize(
"源值,目标", [("11", "a"), (11, "b"), ({"name": " 林深 ", "名称": " 林深 "}, "a")]
)
def test_名称引用不猜数字身份且全提示必须一致__210003(源值, 目标):
源, 映射 = _输入(
{"type": "驾驶", "source": 源值, "target": "茧", "description": "首次实战驾驶"}, 关系=True
)
结果 = _分流(源, 映射)
assert 结果.disposition == "待转换" and 结果.content["source"] == 目标
assert 结果.content["target"] == "c" and 结果.content["brief"] == "首次实战驾驶"
if 目标 == "a":
源["record"]["source_entity_id"] = 11
冲突 = _分流(源, 映射)
assert 冲突.reason_code == "endpoint_conflict" and 冲突.detail == "relation.source"
@pytest.mark.parametrize(
"改动,原因",
[
("hint输入", "unknown_fields"),
("未知来源键", "unknown_fields"),
("来源bool", "field_type_mismatch"),
("其他用途", "unknown_source"),
("来源冲突", "field_conflict"),
("属性非对象", "field_type_mismatch"),
],
)
def test_来源提示不容纳未知字段或扩展授权__210004(改动, 原因):
源, 映射 = _输入()
载荷 = 源["record"]["draft_payload"]
载荷["source"] = {"workId": 12, "chapterId": 17473, "chapter": 1}
if 改动 == "hint输入":
载荷["source_hint"] = {}
elif 改动 == "未知来源键":
载荷["source"]["unknown"] = 1
elif 改动 == "来源bool":
载荷["source"]["chapterId"] = True
elif 改动 == "其他用途":
映射["works"][0]["use"] = "reference"
elif 改动 == "来源冲突":
载荷["出处"] = {"workId": 13}
else:
载荷.pop("fields")
载荷["attributes"] = "不能包成sourcePayload"
assert _分流(源, 映射).reason_code == 原因
@pytest.mark.parametrize(
"章序,有定位,原因",
[
("1", True, "classified"),
(2, True, "scope_conflict"),
(True, True, "field_type_mismatch"),
(1, False, "missing_location"),
],
)
def test_章序只取明确一致声明且不能猜定位__210005(章序, 有定位, 原因):
源, 映射 = _输入()
载荷 = 源["record"]["draft_payload"]
载荷.update(source={"workId": 12, "chapterId": 17473, "chapter": 1}, 章序=章序)
if not 有定位:
映射["sources"][0].pop("chapters")
结果 = _分流(源, 映射)
assert 结果.reason_code == 原因
if 原因 == "classified":
assert 结果.references[0]["locator"] == "chapter:new-chapter"
assert 结果.content["source_hint"] == 载荷["source"]
@pytest.mark.parametrize("证据", [None, "", " ", True, 12])
def test_证据只保原码点而不宽松转型__210006(证据):
源, 映射 = _输入()
源["record"]["draft_payload"]["evidence"] = 证据
结果 = _分流(源, 映射)
if 证据 is None or isinstance(证据, str):
assert 结果.disposition == "待转换" and 结果.content["source_hint"]["evidence"] == 证据
else:
assert 结果.reason_code == "field_type_mismatch"
@pytest.mark.parametrize("边", ["甲方", "乙方"])
@pytest.mark.parametrize("值", [True, ["不得出现在错误中的旧原值"], {}])
def test_关系字段坏端点不被动态结构错误遮蔽__210007(边, 值):
源, 映射 = _输入({"type": "驾驶", "source": 11, "target": 12, "fields": {边: 值}}, 关系=True)
映射["schemas"]["character_relation"]["properties"] = {边: {"type": "string"}}
结果 = _分流(源, 映射)
assert 结果.reason_code == "field_type_mismatch"
assert 结果.detail == ("relation.source" if 边 == "甲方" else "relation.target")