muse-agent-example/tests/迁移/test_隔离目标台账.py
zizi d909d1bd1b 后端实现与用例身份:19 包集成落地并修复收尾缺陷
实现侧:
- 上下文:任务范围拆分为 范围校验/范围授权;索引按可发现口径重建、索引新鲜度改对称差;依赖校验统一快照漂移说明。
- 知识方法:方法与材料读取口径统一;超限方法材料按可选省略,核对路径不再二次计费;删除无合同的读时重算。
- 任务运行:新增 context.usage/tool.denied 事件类型;连接池常驻并在装配生命周期内开关;调用结算与核对分列。
- 效果评测/审校修订/交付连载/作者经验/作品规划:凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。
- 资源加载:能力正文不再夹带索引用的导航注记(该注记此前进入角色与技能的模型提示)。
- 元数据:受保护骨架与代码保护属性对齐;字段校验与内置结构口径同步。
- 基础设施:环境预检进入装配生命周期;数据库连接运行期字段不参与相等比较;索引指纹归一化 jsonb 浮点。
- 删除被替代实现:7 份旧提示词模板与空壳 资料来源 读取器。

用例侧:
- 用例身份与导航元信息迁移;夹具补生命周期、同库暴露与模板封存;
- 本轮定向修复:方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
2026-09-18 01:15:00 +08:00

2837 lines
132 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""隔离目标身份和数据库约束;全部在已授权的临时PG实例与随机新库中验证。"""
from __future__ import annotations
import hashlib
import json
import subprocess
import sys
from dataclasses import asdict
from pathlib import Path
import psycopg
import pytest
from psycopg import sql
from pydantic import TypeAdapter
from muse.作品规划.接口 import 档案保存
from muse.元数据.接口 import 元数据服务, 导入内置结构
from muse.共享.调用身份 import 内容用途, 用途, 调用身份
from muse.共享.错误 import Muse错误
from muse.启动 import 构建
from muse.基础设施.数据库.连接 import 数据库工厂
from muse.故事世界.接口 import 定位字面引文, 正文依据
from muse.正文写作.接口 import 文本节点, 正文草稿, 段落
from muse.资料研究.接口 import 导入请求, 资料服务
from muse.配置 import 应用配置, 数据库引用
from 切换检查 import 检查切换条件
from 导入新库 import 初始化目标, 导入快照, 核对隔离目标
from 建立映射 import 旧快照, 映射配置, 迁移错误
from 映射台账 import 迁移命令ID, 迁移台账
from 核对结果 import 核对迁移结果
pytestmark = pytest.mark.数据库
@pytest.fixture
def 目标方法环境(迁移目标):
库, 参数, _ = 迁移目标
维护引用 = 数据库引用("受控存储", str(Path(库.引用.位置).with_name("维护连接.txt")))
with 数据库工厂(维护引用, 用途.维护).连接() as 连, 连.transaction():
导入内置结构(元数据服务(连))
装配 = 构建(应用配置(库.引用, "synthetic"))
with 装配.生命周期():
身份 = 调用身份("migration-author", None, 用途.生产, 内容用途.规划)
with 库.连接() as 连:
来源 = 资料服务().导入(
连,
身份.作者,
导入请求(
kind="reference",
title="合成来源",
origin="seed://legacy-import",
content="第一章铺垫,第二章回收。",
authorized_uses=("analysis",),
),
)
字段 = {
"装置形态": "结构装置",
"原理": "在当前场景明确回收前文条件,保持承诺与结果可核对",
"埋设手法": "放在日常动线",
"回收点": "选择转折处",
"记忆维持": "中途轻提一次",
"间隔纪律": "隔两章",
"滥用反例": "反复铺垫却无回收",
"例证出处": "合成来源第一、二章",
}
快照 = 旧快照.从载荷(
{
"source": {
"system": "pg",
"database": "synthetic-legacy",
"table": "muse_knowledge_draft",
"id": "1",
"revision": "1",
},
"record": {
"work_id": 0,
"draft_type": "entity",
"source_type": "parse_book",
"source_id": 8,
"source_status": "active",
"source_action_policy": "allowed",
"status": "pending",
"draft_payload": {"型": "craft", "名称": "合成回收法", "字段": 字段},
},
}
)
原值 = {
"works": [
{
"system": "pg",
"database": "synthetic-legacy",
"legacy_id": 0,
"use": "method",
"target_ref": "library:migration-author",
"author_id": 身份.作者,
}
],
"sources": [
{
"system": "pg",
"database": "synthetic-legacy",
"source_type": "parse_book",
"legacy_id": 8,
"work_id": 8,
"target_ref": 来源.source_id,
"revision": "1",
}
],
"schemas": {
"craft": {
"type": "object",
"properties": {k: {"type": "string"} for k in 字段},
"additionalProperties": False,
}
},
"target_schemas": {
"B04:craft": {
"schema_id": "craft",
"base_version": 1,
"fields": {k: "fields." + k for k in 字段},
}
},
}
yield 装配, 身份, 快照, 原值, 参数
@pytest.fixture
def 迁移目标(测试数据库端点, 隔离数据库URL, tmp_path):
管理文件 = tmp_path / "管理连接.txt"
管理文件.write_text(隔离数据库URL, encoding="utf-8")
管理文件.chmod(0o600)
with psycopg.connect(隔离数据库URL) as 连:
允许 = {
"endpoint_type": "unix" if 连.info.host.startswith("/") else "tcp",
"host": 连.info.host,
"port": 连.info.port,
"system_identifier": str(
连.execute(
"SELECT system_identifier FROM pg_catalog.pg_control_system()"
).fetchone()[0]
),
"data_directory": 连.execute("SHOW data_directory").fetchone()[0],
}
允许文件 = tmp_path / "允许实例.json"
允许文件.write_text(json.dumps(允许), encoding="utf-8")
输出 = tmp_path / "目标"
管理引用 = 数据库引用("受控存储", str(管理文件))
回执 = 初始化目标(
允许实例=允许文件,
管理引用=管理引用,
维护模板=测试数据库端点[用途.维护].引用,
应用模板=测试数据库端点[用途.生产].引用,
输出目录=输出,
)
工厂 = 数据库工厂(数据库引用("受控存储", str(输出 / "应用连接.txt")), 用途.生产)
try:
yield (
工厂,
{"允许实例": 允许文件, "管理引用": 管理引用, "目标回执": 输出 / "目标回执.json"},
回执,
)
finally:
with psycopg.connect(隔离数据库URL, autocommit=True) as 连:
连.execute(
sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(回执["database"]))
)
@pytest.mark.case_id(
"NC-w21-21c001",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="受控临时实例初始化随机库",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["回执与实际实例/OID/角色/标记一致,应用不能伪造标记"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_随机隔离库回执与实际标记一致且应用不能伪造__21c001(迁移目标):
库, 参数, 回执 = 迁移目标
assert 核对隔离目标(库, **参数) == 回执
assert 回执["database"].startswith("muse_migration_")
assert 回执["role"] == "muse_app" and 回执["run_purpose"] == "production"
assert Path(库.引用.位置).stat().st_mode & 0o777 == 0o600
with 库.连接() as 连:
with pytest.raises(psycopg.errors.InsufficientPrivilege):
连.execute("UPDATE public.muse_legacy_target SET target_id=target_id")
@pytest.mark.case_id(
"NC-w21-21c002",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="回执、实例标识、角色或用途错误",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["核对拒绝而非相信参数自报隔离"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_错误回执端点实例和用途拒绝而非自报隔离__21c002(迁移目标, tmp_path):
库, 参数, 回执 = 迁移目标
for 键, 值 in (("database_oid", 0), ("system_identifier", "0"), ("role", "muse_eval")):
坏 = tmp_path / (键 + ".json")
坏.write_text(json.dumps({**回执, 键: 值}), encoding="utf-8")
with pytest.raises(迁移错误, match="target_denied"):
核对隔离目标(库, **{**参数, "目标回执": 坏})
with pytest.raises(迁移错误, match="target_denied"):
核对隔离目标(数据库工厂(库.引用, 用途.评测), **参数)
允许 = json.loads(参数["允许实例"].read_text())
参数["允许实例"].write_text(json.dumps({**允许, "system_identifier": "0"}), encoding="utf-8")
with pytest.raises(迁移错误, match="target_denied"):
核对隔离目标(库, **参数)
@pytest.mark.case_id(
"NC-w21-21c003",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="台账来源、哈希、请求及终态被改写",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["真实数据库约束拒绝,reserved不能伪映射,应用不得删原记录"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_迁移原值请求和终态不能被原地改写__21c003(迁移目标):
库, _, _ = 迁移目标
with 库.连接() as 连:
连.execute(
"""INSERT INTO public.muse_legacy_import_record
(author_id,run_purpose,source_key,source_hash,mapping_hash,original,command_id,route)
VALUES ('合成作者','production','源键',%s,%s,'{}','命令','{}')""",
("a" * 64, "b" * 64),
)
for 语句 in (
"UPDATE public.muse_legacy_import_record SET source_key='另一源'",
"UPDATE public.muse_legacy_import_record SET source_hash=repeat('c',64)",
"UPDATE public.muse_legacy_import_record SET state='mapped'",
"DELETE FROM public.muse_legacy_import_record",
):
with pytest.raises(psycopg.Error), 库.连接() as 连:
连.execute(语句)
with 库.连接() as 连:
连.execute(
"UPDATE public.muse_legacy_import_record "
"SET frozen_request='{}',state='quarantined',reason_code='合成隔离'"
)
with pytest.raises(psycopg.Error), 库.连接() as 连:
连.execute("UPDATE public.muse_legacy_import_record SET state='reserved'")
with 库.连接(只读=True) as 连:
assert 连.execute(
"SELECT state,source_key FROM public.muse_legacy_import_record"
).fetchone() == ("quarantined", "源键")
@pytest.mark.case_id(
"TC-knowledge-public-pattern",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="work=0、parse_book、craft未决卡与真实B03来源",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["真实B04未决候选、无作品0、重复导入同产物、源漂移不另造命令或启用"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_真实方法候选导入幂等且源漂移不生成新命令__21c004(目标方法环境):
装配, 身份, 快照, 原值, 参数 = 目标方法环境
for _ in range(2):
结果 = 导入快照(装配, 身份, [快照], 映射配置(原值), **参数)
assert 结果["states"] == {"mapped": 1} and 结果["production_switch_authorized"] is False
服务 = 装配.要求知识方法()
回执 = 服务.正式.读取回执(身份, 迁移命令ID(身份.作者, 快照.source.记录键))
提案 = 服务.读取方法提案(身份, 回执["target_ref"])
assert 提案["revision"] == 1 and 提案["content"]["原理"].startswith("在当前场景")
assert 提案["state"] != "enabled" and 提案["current_version"] == 0
assert 装配.要求作品().列出作品(身份) == []
数据 = 快照.导出()
数据.pop("source_hash")
数据["record"]["draft_payload"]["字段"]["原理"] = "不同源内容"
with pytest.raises(迁移错误, match="drift"):
导入快照(装配, 身份, [旧快照.从载荷(数据)], 映射配置(原值), **参数)
assert 服务.读取方法提案(身份, 回执["target_ref"])["revision"] == 1
@pytest.mark.case_id(
"NC-w21-21c005",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="S01已提交而台账完成前中断",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["按原命令和真实回执补齐,不多候选或改原结果"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_S01已提交而台账未完成时按原命令补齐__21c005(目标方法环境, monkeypatch):
装配, 身份, 快照, 原值, 参数 = 目标方法环境
原保存 = 迁移台账.保存终态
def 中断(self, key, 状态, **字段):
if 状态 == "mapped":
raise RuntimeError("合成中断")
return 原保存(self, key, 状态, **字段)
monkeypatch.setattr(迁移台账, "保存终态", 中断)
with pytest.raises(RuntimeError, match="合成中断"):
导入快照(装配, 身份, [快照], 映射配置(原值), **参数)
服务 = 装配.要求知识方法()
台账 = 迁移台账(服务.数据库, 身份, 服务.正式)
assert 台账.读取(快照.source.记录键)["state"] == "reserved"
原回执 = 服务.正式.读取回执(身份, 迁移命令ID(身份.作者, 快照.source.记录键))
monkeypatch.setattr(迁移台账, "保存终态", 原保存)
结果 = 导入快照(装配, 身份, [快照], 映射配置(原值), **参数)
assert 结果["states"] == {"mapped": 1}
assert 台账.读取(快照.source.记录键)["receipt"] == 原回执
assert 服务.读取方法提案(身份, 原回执["target_ref"])["revision"] == 1
@pytest.mark.case_id(
"NC-w21-21c006",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="映射指向不存在的真实来源版本",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["领域零候选写入,台账保原值及明确隔离原因"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_无效真实来源拒绝候选并保原值__21c006(目标方法环境):
装配, 身份, 快照, 原值, 参数 = 目标方法环境
原值["sources"][0]["revision"] = "99"
结果 = 导入快照(装配, 身份, [快照], 映射配置(原值), **参数)
assert 结果["states"] == {"quarantined": 1}
服务 = 装配.要求知识方法()
assert 服务.列出方法(身份) == []
行 = 迁移台账(服务.数据库, 身份).读取(快照.source.记录键)
assert 行["original"] == 快照.导出() and 行["receipt"] is None
assert 行["targets"] == []
def _本书输入(装配, 身份):
作品 = 装配.要求作品()
结构 = 作品.新档案结构(身份, "migration-work", "work_core", 1)
作品.保存档案(
身份, "create-migration-work", 档案保存("migration-work", 0, {"名称": "合成迁移作品"}, 结构)
)
作品.添加章节(
身份,
"create-migration-chapter",
"migration-work",
"migration-chapter",
"第一章",
预期目录版本=0,
)
正文 = 装配.要求正文()
正文.保存人工(
身份,
"save-migration-body",
"migration-chapter",
0,
正文草稿((段落("p1", (文本节点("甲经过城门。"),)),)),
)
文 = 正文.读取正文(身份, "migration-chapter")
依据 = 正文依据(
"migration-chapter",
"main",
文["revision"],
文["document_hash"],
(定位字面引文(TypeAdapter(正文草稿).validate_python(文["document"]), "甲经过城门。"),),
)
源 = 旧快照.从载荷(
{
"source": {
"system": "pg",
"database": "synthetic-legacy",
"table": "muse_knowledge_draft",
"id": "11",
"revision": "1",
},
"record": {
"work_id": 7,
"source_type": "chapter_extract",
"source_id": 70,
"status": "pending",
"source_status": "active",
"source_action_policy": "allowed",
"draft_type": "entity",
"draft_payload": {"type": "character", "name": "甲", "fields": {"身份": "驾驶员"}},
},
}
)
映射 = {
"works": [
{
"system": "pg",
"database": "synthetic-legacy",
"legacy_id": 7,
"use": "author",
"target_ref": "migration-work",
"author_id": 身份.作者,
}
],
"sources": [
{
"system": "pg",
"database": "synthetic-legacy",
"source_type": "chapter_extract",
"legacy_id": 70,
"work_id": 7,
"target_ref": "migration-chapter",
"revision": "1",
}
],
"schemas": {
"character": {
"type": "object",
"properties": {"身份": {"type": "string"}},
"additionalProperties": False,
}
},
"target_schemas": {
"B02:character": {
"schema_id": "character",
"base_version": 1,
"fields": {"名称": "name", "身份定位": "fields.身份"},
}
},
"body_bindings": [
{
"source_key": 源.source.记录键,
"source_hash": 源.源哈希,
"sources": [asdict(依据)],
"effective_from": "migration-chapter",
}
],
}
return 源, 映射
@pytest.mark.case_id(
"TC-knowledge-authored-entity",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="作者实体英文载荷、动态字段及确切正文引文",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["B02未决提案、字段别名映射与正文依据可查,不独立化、不确认成正式事实"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_本书事实只迁未决候选且保真实正文依据__21c007(目标方法环境):
装配, 身份, _, _, 参数 = 目标方法环境
源, 原值 = _本书输入(装配, 身份)
结果 = 导入快照(装配, 身份, [源], 映射配置(原值), **参数)
assert 结果["states"] == {"mapped": 1}
回执 = 装配.要求知识方法().正式.读取回执(身份, 迁移命令ID(身份.作者, 源.source.记录键))
当前 = 装配.要求故事世界().读取提案(身份, 回执["target_ref"])
assert 当前["content"] == {"名称": "甲", "身份定位": "驾驶员"}
assert 当前["independent"] is False and 当前["sources"][0]["revision"] == 1
assert 当前["sources"][0]["evidence"][0]["quote"] == "甲经过城门。"
assert 装配.要求故事世界().列出世界实体(身份, "migration-work") == []
@pytest.mark.case_id(
"NC-w21-21c009",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="CLI从受控配置取得作者并重复导入",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["命令真实可用、报告稳定私有、不输出正文、不采纳映射作者作为认证"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_迁移命令使用配置作者且报告可重放__21c009(目标方法环境, tmp_path):
装配, 身份, 源, 原值, 参数 = 目标方法环境
配置 = tmp_path / "应用.toml"
配置.write_text(
'["数据库"]\n"取值方式"="受控存储"\n"位置"='
+ json.dumps(装配.配置.数据库.位置, ensure_ascii=False)
+ '\n["资源"]\n"发布身份"="synthetic"\n["运行"]\n"用途"="production"'
'\n[HTTP]\n"作者ID"="migration-author"\n"口令文件"="口令.txt"\n',
encoding="utf-8",
)
(tmp_path / "口令.txt").write_text("synthetic-only-password", encoding="utf-8")
输入, 映射文件, 报告 = [tmp_path / n for n in ("输入.json", "映射.json", "报告.json")]
输入.write_text(json.dumps([源.导出()], ensure_ascii=False), encoding="utf-8")
映射文件.write_text(json.dumps(原值, ensure_ascii=False), encoding="utf-8")
入口 = Path(__file__).resolve().parents[2] / "数据库/旧库迁移/入口.py"
命令 = [
sys.executable,
str(入口),
"导入目标",
"--快照",
str(输入),
"--映射",
str(映射文件),
"--配置",
str(配置),
"--允许实例",
str(参数["允许实例"]),
"--管理引用",
参数["管理引用"].位置,
"--目标回执",
str(参数["目标回执"]),
"--输出",
str(报告),
]
for _ in range(2):
完成 = subprocess.run(命令, capture_output=True, text=True, timeout=30)
assert 完成.returncode == 0, 完成.stderr
assert "在当前场景" not in 完成.stdout
assert json.loads(报告.read_text())["states"] == {"mapped": 1}
assert 报告.stat().st_mode & 0o777 == 0o600
@pytest.mark.case_id(
"NC-w21-21c00a",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="源尚未登记、已真实映射及源后续版本变化",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["逐项只读对账,缺记录或源变化拒绝通过,不宣称整包或生产授权"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_只读对账核实际产物并阻止源增量冒充完成__21c00a(目标方法环境):
装配, 身份, 源, 原值, 参数 = 目标方法环境
映射 = 映射配置(原值)
缺 = 核对迁移结果(装配, 身份, [源], 映射, **参数)
assert 缺["issues"][0]["reason_code"] == "source_unaccounted"
导入快照(装配, 身份, [源], 映射, **参数)
核对 = 核对迁移结果(装配, 身份, [源], 映射, **参数)
assert 核对["data_checks_passed"] is True and 核对["package_acceptance_claimed"] is False
数据 = 源.导出()
数据.pop("source_hash")
数据["source"]["revision"] = "2"
当前 = 旧快照.从载荷(数据)
检查 = 检查切换条件(装配, 身份, [源], [当前], 映射, **参数)
assert 检查["data_checks_passed"] is False and 检查["source_delta"]["changed"]
assert 检查["production_switch_authorized"] is False
@pytest.mark.case_id(
"NC-w21-21c00b",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="两个并发同源导入",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["双唯一键竞争正确收敛,只有一个候选和一条映射"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_并发同源只产生同一候选与映射__21c00b(目标方法环境):
from concurrent.futures import ThreadPoolExecutor
装配, 身份, 源, 原值, 参数 = 目标方法环境
映射 = 映射配置(原值)
with ThreadPoolExecutor(max_workers=2) as 池:
一 = 池.submit(导入快照, 装配, 身份, [源], 映射, **参数)
二 = 池.submit(导入快照, 装配, 身份, [源], 映射, **参数)
assert 一.result()["states"] == 二.result()["states"] == {"mapped": 1}
服务 = 装配.要求知识方法()
assert len(服务.列出方法(身份)) == 1
记录 = 迁移台账(服务.数据库, 身份, 服务.正式).列出([源.source.记录键])
assert len(记录) == 1 and 记录[0]["state"] == "mapped"
@pytest.mark.case_id(
"TC-knowledge-entity-relation",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="entity形关系卡与已确认/不存在的真实目标端点",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["映为关系候选不造普通实体;端点不在目标时隔离,不自行确认"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("缺目标", [False, True])
def test_实体形关系迁为关系候选且核实目标端点__21c00c(目标方法环境, 缺目标):
from muse.故事世界.接口 import 事实定位
from muse.正式变更.接口 import 作者动作
装配, 身份, _, _, 参数 = 目标方法环境
初, 原值 = _本书输入(装配, 身份)
世界 = 装配.要求故事世界()
# 合成环境中由作者事先明确确认两个目标实体;迁移器自身不调用确认。
for id_, 名 in (("existing-a", "甲"), ("existing-b", "乙")):
回执 = 世界.保存实体提案(
身份,
"seed-" + id_,
work_id="migration-work",
object_id=id_,
expected_revision=0,
schema_id="character",
schema_version=1,
content={"名称": 名},
independent=True,
)
提案 = 世界.读取提案(身份, 回执["target_ref"])
定位 = 事实定位(
回执["target_ref"], 提案["revision"], 提案["proposal_hash"], "world:" + id_, 0
)
审阅 = 世界.打开事实审阅(身份, 定位)
世界.决定事实(
身份,
"confirm-" + id_,
定位,
作者动作.采纳,
审阅["review_id"],
审阅["review_hash"],
("object",),
)
数据 = 初.导出()
数据.pop("source_hash")
数据["record"]["draft_payload"] = {
"type": "character_relation",
"name": "甲乙同船",
"fields": {"甲方": "甲", "乙方": "乙", "关系类型": "同船"},
}
源 = 旧快照.从载荷(数据)
原值["schemas"] = {
"character_relation": {
"type": "object",
"properties": {k: {"type": "string"} for k in ("甲方", "乙方", "关系类型")},
"additionalProperties": False,
}
}
原值["target_schemas"] = {
"B02:character_relation": {
"schema_id": "character_relation",
"base_version": 1,
"fields": {"甲方": "source", "乙方": "target", "关系类型": "relation_type"},
}
}
原值["objects"] = [
{
"system": "pg",
"database": "synthetic-legacy",
"work_id": 7,
"use": "author",
"scope": "migration-work",
"refs": ["name:" + 名],
"target_ref": id_,
}
for 名, id_ in (("甲", "existing-a"), ("乙", "nonexistent" if 缺目标 else "existing-b"))
]
原值["body_bindings"][0]["source_hash"] = 源.源哈希
结果 = 导入快照(装配, 身份, [源], 映射配置(原值), **参数)
assert 结果["states"] == ({"quarantined": 1} if 缺目标 else {"mapped": 1})
assert len(世界.列出世界实体(身份, "migration-work")) == 2
if not 缺目标:
回执 = 世界.正式.读取回执(身份, 迁移命令ID(身份.作者, 源.source.记录键))
assert 世界.读取提案(身份, 回执["target_ref"])["content"] == {
"甲方": "existing-a",
"乙方": "existing-b",
"关系类型": "同船",
}
@pytest.mark.case_id(
"NC-w21-21c00d",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="目标结构绑定指向另一类型",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["真实S04类型不匹配时零写入,不能暗改来源类型"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_目标结构不能把来源类型暗改为另一类型__21c00d(目标方法环境):
装配, 身份, 源, 原值, 参数 = 目标方法环境
原值["target_schemas"]["B04:craft"]["schema_id"] = "character"
结果 = 导入快照(装配, 身份, [源], 映射配置(原值), **参数)
assert 结果["states"] == {"quarantined": 1}
assert 结果["errors"][0]["reason_code"] == "target_type_mismatch"
assert 装配.要求知识方法().列出方法(身份) == []
@pytest.mark.case_id(
"NC-w21-21c008",
environment="独立临时PG实例、随机目标库、真实S01/B02/B04与合成来源;不访问真实生产",
given="有源作者事实缺正文依据",
when="在独立临时PG实例的production角色随机库执行真实目标入口并核对",
then=["隔离且零目标事实,不降级为独立设定"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_缺正文依据不降级独立设定__21c008(目标方法环境):
装配, 身份, _, _, 参数 = 目标方法环境
源, 原值 = _本书输入(装配, 身份)
原值["body_bindings"] = []
结果 = 导入快照(装配, 身份, [源], 映射配置(原值), **参数)
assert 结果["states"] == {"quarantined": 1}
assert 结果["errors"][0]["reason_code"] == "target_source_missing"
assert 装配.要求故事世界().列出世界实体(身份, "migration-work") == []
def _核对保全且未发业务命令(环境, 源, 原值, 状态, 原因):
from muse.正式变更.接口 import 变更错误
装配, 身份, _, _, 参数 = 环境
映射 = 映射配置(原值)
一 = 导入快照(装配, 身份, [源], 映射, **参数)
assert 一 == 导入快照(装配, 身份, [源], 映射, **参数)
assert 一["states"] == {状态: 1} and 一["all_accounted"] is True
正式 = 装配.要求知识方法().正式
行 = 迁移台账(正式.数据库, 身份, 正式).读取(源.source.记录键)
assert 行 is not None and 行["original"] == 源.导出()
assert 行["source_hash"] == 源.源哈希 and 行["reason_code"] == 原因
assert 行["targets"] == [] and 行["receipt"] is None and 行["frozen_request"] is None
with pytest.raises(变更错误) as 拒绝:
正式.读取回执(身份, 迁移命令ID(身份.作者, 源.source.记录键))
assert 拒绝.value.错误码 == "RECEIPT_NOT_FOUND"
assert 装配.要求知识方法().列出方法(身份) == []
核对 = 核对迁移结果(装配, 身份, [源], 映射, **参数)
assert 核对["data_checks_passed"] is (状态 == "historical")
assert 核对["package_acceptance_claimed"] is False
return 行
@pytest.mark.case_id(
"TC-knowledge-missing-endpoint",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="K05明确关系的一个名称端点缺失或多义",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=["实际隔离并报告对应原因;原值完整、无迁移业务回执、无不完整关系"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("多义", [False, True])
def test_显式关系缺失或多义端点真实隔离__21d001(目标方法环境, 多义):
装配, 身份, _, _, _ = 目标方法环境
初, 原值 = _本书输入(装配, 身份)
数据 = 初.导出()
数据.pop("source_hash")
数据["record"].update(
draft_type="relation",
draft_payload={"甲方": "甲", "乙方": "乙", "type": "同船", "fields": {}},
)
源 = 旧快照.从载荷(数据)
基础 = {
"system": "pg",
"database": "synthetic-legacy",
"work_id": 7,
"use": "author",
"scope": "migration-work",
}
原值["objects"] = [{**基础, "refs": ["name:甲"], "target_ref": "existing-a"}]
if 多义:
原值["objects"] += [
{**基础, "refs": ["name:乙"], "target_ref": id_} for id_ in ("existing-b", "existing-c")
]
原值["schemas"]["character_relation"] = {
"type": "object",
"properties": {},
"additionalProperties": False,
}
_核对保全且未发业务命令(
目标方法环境, 源, 原值, "quarantined", "ambiguous_endpoint" if 多义 else "missing_endpoint"
)
assert 装配.要求故事世界().列出世界实体(身份, "migration-work") == []
@pytest.mark.case_id(
"TC-knowledge-conflicting-type-keys",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="型=craft与type=character冲突",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=["台账同时保留两值且无业务候选;重复导入不改结论"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_冲突类型双值留台账且不写候选__21d002(目标方法环境):
_, _, 初, 原值, _ = 目标方法环境
数据 = 初.导出()
数据.pop("source_hash")
数据["record"]["draft_payload"]["type"] = "character"
源 = 旧快照.从载荷(数据)
行 = _核对保全且未发业务命令(目标方法环境, 源, 原值, "quarantined", "type_conflict")
载荷 = 行["original"]["record"]["draft_payload"]
assert 载荷["型"] == "craft" and 载荷["type"] == "character"
@pytest.mark.case_id(
"TC-knowledge-zero-work-not-pattern",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="work_id=0普通地点卡",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=["隔离且无新作品或公共方法,不能猜成公共范式"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_零作品普通地点不能成为公共方法__21d003(目标方法环境):
装配, 身份, 初, 原值, _ = 目标方法环境
数据 = 初.导出()
数据.pop("source_hash")
数据["record"]["draft_payload"] = {"type": "location", "name": "旧城", "fields": {}}
原值["schemas"]["location"] = {
"type": "object",
"properties": {},
"additionalProperties": False,
}
源 = 旧快照.从载荷(数据)
_核对保全且未发业务命令(目标方法环境, 源, 原值, "quarantined", "zero_work_not_pattern")
assert 装配.要求作品().列出作品(身份) == []
@pytest.mark.case_id(
"TC-knowledge-cross-book-instances",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="公共方法卡含两本书的同号章实例,分别映射真实B03来源版本",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=["从真实B04提案读到两组来源ID/版本/章定位;原引文留台账,重放不复制或启用"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_跨书同章号分别进入真实方法证据__21d004(目标方法环境):
装配, 身份, 初, 原值, 参数 = 目标方法环境
with 装配.要求知识方法().数据库.连接() as 连:
乙 = 资料服务().导入(
连,
身份.作者,
导入请求(
kind="reference",
title="合成乙书",
origin="seed://legacy-second-book",
content="第十二章,合成乙举例。",
authorized_uses=("analysis",),
),
)
甲引用 = 原值["sources"][0]
甲引用.update(aliases=["甲书"], chapters={"12": "甲书:第12章"})
原值["sources"].append(
{
**甲引用,
"legacy_id": 10,
"work_id": 10,
"target_ref": 乙.source_id,
"aliases": ["乙书"],
"chapters": {"12": "乙书:第12章"},
}
)
数据 = 初.导出()
数据.pop("source_hash")
数据["record"]["draft_payload"]["实例"] = [
{"书": "甲书", "章": 12, "引文": "合成甲"},
{"书": "乙书", "章": 12, "引文": "合成乙"},
]
源 = 旧快照.从载荷(数据)
映射 = 映射配置(原值)
一 = 导入快照(装配, 身份, [源], 映射, **参数)
assert 一 == 导入快照(装配, 身份, [源], 映射, **参数)
assert 一["states"] == {"mapped": 1}
服务 = 装配.要求知识方法()
回执 = 服务.正式.读取回执(身份, 迁移命令ID(身份.作者, 源.source.记录键))
提案 = 服务.读取方法提案(身份, 回执["target_ref"])
assert [(e.source_id, e.revision, e.定位) for e in 提案["证据"]][-2:] == [
(甲引用["target_ref"], 1, "甲书:第12章"),
(乙.source_id, 1, "乙书:第12章"),
]
assert 提案["current_version"] == 0 and 提案["state"] != "enabled"
行 = 迁移台账(服务.数据库, 身份, 服务.正式).读取(源.source.记录键)
assert 行 is not None and 行["original"] == 源.导出()
assert len(服务.列出方法(身份)) == 1
assert 核对迁移结果(装配, 身份, [源], 映射, **参数)["data_checks_passed"] is True
@pytest.mark.case_id(
"TC-knowledge-no-resurrection",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="旧实体、别名、向量已删除,另含拒绝及来源撤回",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=["完整原行进入历史;无新迁移业务回执、候选、别名或活跃索引"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize(
"表,原行,原因,状态",
[
(
"muse_knowledge_entity",
{"work_id": 7, "entity_type": "character", "name": "旧角色", "deleted": True},
"inactive_history",
"historical",
),
# example_upgrade_alias 属于冻结旧DDL差集,分流前统一 unknown_table 隔离;
# 旧 status=deleted 也不能让未承接表转为成功(旧知识记录分流.md 差集登记)。
(
"example_upgrade_alias",
{"work_id": 7, "canonical_name": "甲", "alias": "旧别名", "deleted": True},
"unknown_table",
"quarantined",
),
(
"example_knowledge_embedding",
{"work_id": 7, "entity_id": 11, "embedding": [0.2, 0.3], "deleted": True},
"inactive_history",
"historical",
),
(
"muse_knowledge_draft",
{"work_id": 0, "status": "rejected", "draft_payload": {"型": "craft"}},
"inactive_history",
"historical",
),
(
"muse_knowledge_draft",
{"work_id": 0, "source_status": "withdrawn", "draft_payload": {"型": "craft"}},
"source_revoked",
"historical",
),
],
)
def test_删除别名向量拒绝与撤权仅保历史__21d005(目标方法环境, 表, 原行, 原因, 状态):
_, _, 初, 原值, _ = 目标方法环境
数据 = 初.导出()
数据.pop("source_hash")
数据["source"]["table"], 数据["record"] = 表, 原行
源 = 旧快照.从载荷(数据)
行 = _核对保全且未发业务命令(目标方法环境, 源, 原值, 状态, 原因)
assert 行["route"]["content"] == {} and not 行["route"]["target_ref"]
@pytest.mark.case_id(
"TC-knowledge-unknown-fields",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="动态字段陌生、类型错误或必填缺失",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=["具体原因隔离并保全部原动态值,目标不启用"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize(
"方式,原因",
[("未知", "unknown_fields"), ("类型错", "field_type_mismatch"), ("缺必填", "missing_required")],
)
def test_不可解释动态值原样留存且不启用__21d006(目标方法环境, 方式, 原因):
_, _, 初, 原值, _ = 目标方法环境
数据 = 初.导出()
数据.pop("source_hash")
字段 = 数据["record"]["draft_payload"]["字段"]
原值["schemas"]["craft"]["required"] = ["原理"]
if 方式 == "未知":
字段["陌生字段"] = {"原值": [1, 2]}
elif 方式 == "类型错":
字段["原理"] = 3
else:
del 字段["原理"]
源 = 旧快照.从载荷(数据)
行 = _核对保全且未发业务命令(目标方法环境, 源, 原值, "quarantined", 原因)
assert 行["original"]["record"]["draft_payload"]["字段"] == 字段
@pytest.mark.case_id(
"TC-knowledge-source-namespaces",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="PG草稿和SQLite卡主键相同、没有可信镜像关联",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=[
"两条来源身份和命令独立留痕;PG保真实未决候选,SQLite资格不明独立隔离,不合并或复制活跃卡"
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_PG与SQLite同号卡不合并或复制活跃对象__21d007(目标方法环境):
装配, 身份, PG源, 原值, 参数 = 目标方法环境
SQLite源 = 旧快照.从载荷(
{
"source": {**PG源.导出()["source"], "system": "sqlite", "table": "cards"},
"record": {
"id": PG源.source.id,
"kind": "craft",
"work_id": 0,
"payload_json": json.dumps(PG源.原行["draft_payload"], ensure_ascii=False),
},
}
)
原值["works"] += [{**r, "system": "sqlite"} for r in 原值["works"]]
映射 = 映射配置(原值)
快照 = [PG源, SQLite源]
结果 = 导入快照(装配, 身份, 快照, 映射, **参数)
assert 结果 == 导入快照(装配, 身份, 快照, 映射, **参数)
assert 结果["states"] == {"mapped": 1, "quarantined": 1}
assert 结果["distinct_source_count"] == 2
服务 = 装配.要求知识方法()
行 = 迁移台账(服务.数据库, 身份, 服务.正式).列出([s.source.记录键 for s in 快照])
assert len(行) == 2 and len({r["command_id"] for r in 行}) == 2
assert {r["source_key"]: r["original"] for r in 行} == {s.source.记录键: s.导出() for s in 快照}
assert len(服务.列出方法(身份)) == 1
assert next(r for r in 行 if r["source_key"] == SQLite源.source.记录键)["targets"] == []
@pytest.mark.case_id(
"TC-f6f4e45e6f52",
environment="独立临时PG实例内随机production库与合成旧输入;SQLite例使用私有临时副本",
given="合成旧SQLite副本为WAL,保留runs/events/reviews/revisions/cards五表及原主键",
when="用真实隔离PG公共迁移入口导入、重放并回查台账及领域对象",
then=["只读前后WAL和五表仍可查;五类原行完整落台账,四类历史/无资格卡隔离,不复造新运行或候选"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_旧账本WAL与五表按历史迁移__f6f4e4(目标方法环境, tmp_path):
import hashlib
import sqlite3
from 读取旧SQLite import 读取旧库
装配, 身份, _, _, 参数 = 目标方法环境
副本 = tmp_path / "旧账本副本.sqlite"
# 最小合成旧表保留原主键、关联键与待验证原值;不执行保全DDL建设新版库。
with sqlite3.connect(副本) as 连:
assert 连.execute("PRAGMA journal_mode=WAL").fetchone()[0].lower() == "wal"
连.executescript("""
CREATE TABLE runs(id TEXT PRIMARY KEY, input_json TEXT, output_text TEXT);
CREATE TABLE events(run_id TEXT, seq INTEGER, payload_json TEXT,
PRIMARY KEY(run_id,seq));
CREATE TABLE reviews(id INTEGER PRIMARY KEY, run_id TEXT, action TEXT, reviewer TEXT);
CREATE TABLE revisions(id INTEGER PRIMARY KEY, review_id INTEGER,
before_text TEXT, after_text TEXT);
CREATE TABLE cards(id TEXT PRIMARY KEY, kind TEXT, work_id INTEGER, payload_json TEXT);
INSERT INTO runs VALUES('old-run','{}','旧合成产物');
INSERT INTO events VALUES('old-run',1,'{"quote":"合成旧事件"}');
INSERT INTO reviews VALUES(1,'old-run','adopt','旧作者');
INSERT INTO revisions VALUES(1,1,'旧句','改句');
INSERT INTO cards VALUES('1','character',7,'{"type":"character","name":"合成旧卡"}');
""")
连.close()
哈希 = hashlib.sha256(副本.read_bytes()).hexdigest()
必表 = ("runs", "events", "reviews", "revisions", "cards")
快照 = 读取旧库(副本, 数据集="synthetic-wal-ledger", 表=必表, 快照版本="export-1")
assert {s.source.table for s in 快照} == set(必表)
assert hashlib.sha256(副本.read_bytes()).hexdigest() == 哈希
映射 = 映射配置({"works": []})
结果 = 导入快照(装配, 身份, 快照, 映射, **参数)
assert 结果 == 导入快照(装配, 身份, 快照, 映射, **参数)
assert 结果["states"] == {"historical": 4, "quarantined": 1}
服务 = 装配.要求知识方法()
行 = 迁移台账(服务.数据库, 身份, 服务.正式).列出([s.source.记录键 for s in 快照])
assert {r["source_key"]: r["original"] for r in 行} == {s.source.记录键: s.导出() for s in 快照}
assert all(r["receipt"] is None and r["targets"] == [] for r in 行)
assert 装配.任务运行 is not None and 装配.任务运行.列出任务(身份.作者) == []
assert 服务.列出方法(身份) == []
with sqlite3.connect(副本.as_uri() + "?mode=ro", uri=True) as 连:
assert 连.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
名称 = {r[0] for r in 连.execute("SELECT name FROM sqlite_master WHERE type='table'")}
assert set(必表) <= 名称
连.close()
def _正文文件输入(装配, 身份, tmp_path, *, archive=False):
import hashlib
from 读取旧文件 import 读取文件集
if not archive:
_本书输入(装配, 身份)
源根 = tmp_path / "文件源"
源根.mkdir()
内容 = b"\xff\x00archive" if archive else "\ufeff迁移候选\r\n😀e\u0301\n".encode()
(源根 / "章.txt").write_bytes(内容)
保全 = tmp_path / "文件保全"
源 = 读取文件集(
源根,
[{"relpath": "章.txt", "revision": "1", "sha256": hashlib.sha256(内容).hexdigest()}],
数据集="synthetic-target-file",
输出目录=保全,
)[0]
路由 = {
"source_key": 源.source.记录键,
"source_hash": 源.源哈希,
"route": "archive" if archive else "body_candidate",
}
值 = {"works": [], "file_routes": [路由]}
if not archive:
当前 = 装配.要求正文().读取正文(身份, "migration-chapter")
值["works"] = [
{
"system": "file",
"database": 源.source.database,
"legacy_id": 7,
"use": "author",
"target_ref": "migration-work",
"author_id": 身份.作者,
}
]
路由.update(
format="text_utf8",
legacy_work_id=7,
target={
"chapter_id": "migration-chapter",
"branch_id": "main",
"base_revision": 当前["revision"],
},
)
return 源, 值, 保全
@pytest.mark.case_id(
"NC-w21-21f001",
environment="合成文件与明确私有目录、独立临时PG实例随机production库",
given="文件导入后作者编辑候选,随后副本缺失",
when="按文件保全/正文合同执行真实受控入口并回查结果",
then=[
"真实B05作者候选、当前正文不变、原版本回放不覆盖后续编辑;副本缺失使双面对账失败但不阻日常B05"
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_正文文件只产作者候选且重放不覆盖后续编辑__21f001(目标方法环境, tmp_path):
from 读取旧文件 import 解码文件
装配, 身份, _, _, 参数 = 目标方法环境
源, 值, 保全 = _正文文件输入(装配, 身份, tmp_path)
参数 = {**参数, "文件回执": 保全 / "文件回执.json"}
映射 = 映射配置(值)
正文 = 装配.要求正文()
前 = 正文.读取正文(身份, "migration-chapter")
结果 = 导入快照(装配, 身份, [源], 映射, **参数)
assert 结果["states"] == {"mapped": 1} and 结果["file_checks"]["passed"] is True
assert 正文.读取正文(身份, "migration-chapter") == 前
回执 = 正文.正式.读取回执(身份, 迁移命令ID(身份.作者, 源.source.记录键))
候选 = 正文.读取候选(身份, 回执["target_ref"], 版本=1)
assert 候选["origin"] == "author" and 候选["visible_text"] == 解码文件(源).decode()
assert 候选["revision"] == 1 and 候选["decision"] == "undecided"
改稿 = 正文草稿((段落("author-paragraph", (文本节点("作者后续修改"),)),))
正文.编辑候选(身份, "file-author-edit", 候选["candidate_id"], 1, 候选["candidate_hash"], 改稿)
assert 导入快照(装配, 身份, [源], 映射, **参数) == 结果
assert 正文.读取候选(身份, 候选["candidate_id"])["visible_text"] == "作者后续修改"
assert 正文.读取候选(身份, 候选["candidate_id"])["revision"] == 2
行 = 迁移台账(正文.数据库, 身份, 正文.正式).读取(源.source.记录键)
assert (
行 is not None
and 行["targets"][0]["kind"] == "candidate"
and 行["targets"][0]["revision"] == 1
)
assert 核对迁移结果(装配, 身份, [源], 映射, **参数)["data_checks_passed"] is True
(保全 / "原始文件/章.txt").unlink()
核对 = 核对迁移结果(装配, 身份, [源], 映射, **参数)
assert 核对["pg_checks"]["passed"] is True and 核对["file_checks"]["passed"] is False
assert 核对["data_checks_passed"] is False
assert 检查切换条件(装配, 身份, [源], [源], 映射, **参数)["data_checks_passed"] is False
assert 正文.读取候选(身份, 候选["candidate_id"])["revision"] == 2
@pytest.mark.case_id(
"NC-w21-21f002",
environment="合成文件与明确私有目录、独立临时PG实例随机production库",
given="合法正文文件和目标章但未提供文件回执",
when="按文件保全/正文合同执行真实受控入口并回查结果",
then=["缺文件保全回执时零台账和业务候选写入"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_文件副本未核实即零目标写入__21f002(目标方法环境, tmp_path):
装配, 身份, _, _, 参数 = 目标方法环境
源, 值, _ = _正文文件输入(装配, 身份, tmp_path)
with pytest.raises(迁移错误, match="file_copy_unverified"):
导入快照(装配, 身份, [源], 映射配置(值), **参数)
正文 = 装配.要求正文()
assert 正文.列出候选(身份, "migration-chapter") == []
assert 迁移台账(正文.数据库, 身份, 正文.正式).读取(源.source.记录键) is None
@pytest.mark.case_id(
"NC-w21-21f003",
environment="合成文件与明确私有目录、独立临时PG实例随机production库",
given="正文路由存在陈旧基线、同作者错书、跨作者章或格式错误",
when="按文件保全/正文合同执行真实受控入口并回查结果",
then=["陈旧基线、同作者错书、跨作者与坏正文格式精确拒绝"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize(
"方式,原因",
[
("陈旧基线", "target_base_stale"),
("同作者错书", "target_scope_mismatch"),
("跨作者", "SCOPE_DENIED"),
("坏格式", "file_document_invalid"),
],
)
def test_正文文件不绕基线作用域或格式__21f003(目标方法环境, tmp_path, 方式, 原因):
装配, 身份, _, _, 参数 = 目标方法环境
源, 值, 保全 = _正文文件输入(装配, 身份, tmp_path)
路由 = 值["file_routes"][0]
if 方式 == "陈旧基线":
路由["target"]["base_revision"] = 0
elif 方式 == "坏格式":
路由["format"] = "document_v1"
else:
另一身份 = (
身份
if 方式 == "同作者错书"
else 调用身份("another-author", None, 用途.生产, 内容用途.规划)
)
作品 = 装配.要求作品()
结构 = 作品.新档案结构(另一身份, "another-work", "work_core", 1)
作品.保存档案(
另一身份, "file-another-work", 档案保存("another-work", 0, {"名称": "另一作品"}, 结构)
)
作品.添加章节(
另一身份,
"file-another-chapter",
"another-work",
"another-chapter",
"另一章",
预期目录版本=0,
)
路由["target"].update(chapter_id="another-chapter", base_revision=0)
结果 = 导入快照(装配, 身份, [源], 映射配置(值), **参数, 文件回执=保全 / "文件回执.json")
assert 结果["states"] == {"quarantined": 1}
assert 结果["errors"][0]["reason_code"] == 原因
assert 装配.要求正文().列出候选(身份, "migration-chapter") == []
@pytest.mark.case_id(
"NC-w21-21f004",
environment="合成文件与明确私有目录、独立临时PG实例随机production库",
given="真实S01候选已提交而台账保存中断,另有篡改回执/种类",
when="按文件保全/正文合同执行真实受控入口并回查结果",
then=["B05提交后台账中断按原回执恢复;错种类/候选哈希及伪目标kind拒绝"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_正文候选提交后中断取原回执且拒错种类__21f004(目标方法环境, tmp_path, monkeypatch):
from 转换业务对象 import 核对目标产物
装配, 身份, _, _, 参数 = 目标方法环境
源, 值, 保全 = _正文文件输入(装配, 身份, tmp_path)
参数 = {**参数, "文件回执": 保全 / "文件回执.json"}
映射 = 映射配置(值)
原保存 = 迁移台账.保存终态
def 中断(self, *args, **kwargs):
raise RuntimeError("模拟领域已提交后台账中断")
with monkeypatch.context() as m:
m.setattr(迁移台账, "保存终态", 中断)
with pytest.raises(RuntimeError, match="模拟领域"):
导入快照(装配, 身份, [源], 映射, **参数)
正文 = 装配.要求正文()
行 = 迁移台账(正文.数据库, 身份, 正文.正式).读取(源.source.记录键)
assert 行 is not None and 行["state"] == "reserved"
assert len(正文.列出候选(身份, "migration-chapter")) == 1
assert 导入快照(装配, 身份, [源], 映射, **参数)["states"] == {"mapped": 1}
assert len(正文.列出候选(身份, "migration-chapter")) == 1
回执 = 正文.正式.读取回执(身份, 行["command_id"])
错 = json.loads(json.dumps(回执))
错["results"][0]["proposal_id"] = 错["results"][0].pop("candidate_id")
with pytest.raises(迁移错误, match="target_receipt_invalid"):
核对目标产物(装配, 身份, 行["frozen_request"], 错)
错 = json.loads(json.dumps(回执))
错["results"][0]["candidate_hash"] = "0" * 64
with pytest.raises(迁移错误, match="target_mismatch"):
核对目标产物(装配, 身份, 行["frozen_request"], 错)
目标 = 核对目标产物(装配, 身份, 行["frozen_request"], 回执)
目标[0]["kind"] = "proposal"
with pytest.raises(迁移错误, match="target_evidence_mismatch"):
迁移台账(正文.数据库, 身份, 正文.正式).保存终态(
源.source.记录键, "mapped", 目标=目标, 回执=回执
)
assert 迁移台账.保存终态 is 原保存
@pytest.mark.case_id(
"NC-w21-21f005",
environment="合成文件与明确私有目录、独立临时PG实例随机production库",
given="明确archive用途的非UTF8原文件及完整保全回执",
when="按文件保全/正文合同执行真实受控入口并回查结果",
then=["二进制原文件入历史不造候选,原字节可恢复,文件面和PG面都核对"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_文件归档保全原字节而不造正文候选__21f005(目标方法环境, tmp_path):
from 读取旧文件 import 解码文件
装配, 身份, _, _, 参数 = 目标方法环境
源, 值, 保全 = _正文文件输入(装配, 身份, tmp_path, archive=True)
参数 = {**参数, "文件回执": 保全 / "文件回执.json"}
结果 = 导入快照(装配, 身份, [源], 映射配置(值), **参数)
assert 结果["states"] == {"historical": 1}
正式 = 装配.要求知识方法().正式
行 = 迁移台账(正式.数据库, 身份, 正式).读取(源.source.记录键)
assert 行 is not None and 行["original"] == 源.导出() and 行["targets"] == []
assert 解码文件(旧快照.从载荷(行["original"])) == b"\xff\x00archive"
assert 核对迁移结果(装配, 身份, [源], 映射配置(值), **参数)["data_checks_passed"] is True
@pytest.mark.case_id(
"NC-w21-21f006",
environment="合成文件与明确私有目录、独立临时PG实例随机production库",
given="两个并发同源正文导入,随后改变映射或源字节并提供对应新保全",
when="按文件保全/正文合同执行真实受控入口并回查结果",
then=["同文件并发只创建一个候选;正确新保全仍不能绕过源或映射漂移"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_正文文件并发只建一候选且源或映射漂移拒绝__21f006(目标方法环境, tmp_path):
import hashlib
from concurrent.futures import ThreadPoolExecutor
from 读取旧文件 import 读取文件集
装配, 身份, _, _, 参数 = 目标方法环境
源, 值, 保全 = _正文文件输入(装配, 身份, tmp_path)
参数 = {**参数, "文件回执": 保全 / "文件回执.json"}
映射 = 映射配置(值)
with ThreadPoolExecutor(max_workers=2) as 池:
任务 = [池.submit(导入快照, 装配, 身份, [源], 映射, **参数) for _ in range(2)]
assert all(t.result()["states"] == {"mapped": 1} for t in 任务)
正文 = 装配.要求正文()
assert len(正文.列出候选(身份, "migration-chapter")) == 1
错映射 = json.loads(json.dumps(值))
错映射["file_routes"][0]["target"]["branch_id"] = "other"
with pytest.raises(迁移错误, match="drift"):
导入快照(装配, 身份, [源], 映射配置(错映射), **参数)
(tmp_path / "文件源/章.txt").write_bytes(b"changed")
新 = 读取文件集(
tmp_path / "文件源",
[{"relpath": "章.txt", "revision": "1", "sha256": hashlib.sha256(b"changed").hexdigest()}],
数据集="synthetic-target-file",
输出目录=tmp_path / "新保全",
)[0]
值["file_routes"][0]["source_hash"] = 新.源哈希
参数["文件回执"] = tmp_path / "新保全/文件回执.json"
with pytest.raises(迁移错误, match="drift"):
导入快照(装配, 身份, [新], 映射配置(值), **参数)
assert len(正文.列出候选(身份, "migration-chapter")) == 1
行 = 迁移台账(正文.数据库, 身份, 正文.正式).读取(源.source.记录键)
assert 行 is not None and 行["original"] == 源.导出()
@pytest.mark.case_id(
"NC-w21-21f007",
environment="合成文件与明确私有目录、独立临时PG实例随机production库",
given="合法未保存空章,通过CLI先缺文件回执再提供完整回执",
when="按文件保全/正文合同执行真实受控入口并回查结果",
then=["CLI向合法空章创建未决候选,缺文件面先拒绝,当前正文仍为版本0"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_文件CLI向空章导入真实候选且要求文件面__21f007(目标方法环境, tmp_path, capsys):
from 入口 import main
装配, 身份, _, _, 参数 = 目标方法环境
_, 值, 保全 = _正文文件输入(装配, 身份, tmp_path)
装配.要求作品().添加章节(
身份, "file-empty-chapter", "migration-work", "empty-chapter", "空章", 预期目录版本=1
)
值["file_routes"][0]["target"].update(chapter_id="empty-chapter", base_revision=0)
配置 = tmp_path / "应用.toml"
配置.write_text(
'["数据库"]\n"取值方式"="受控存储"\n"位置"='
+ json.dumps(装配.配置.数据库.位置, ensure_ascii=False)
+ '\n["资源"]\n"发布身份"="synthetic"\n["运行"]\n"用途"="production"'
'\n[HTTP]\n"作者ID"="migration-author"\n"口令文件"="口令.txt"\n',
encoding="utf-8",
)
(tmp_path / "口令.txt").write_text("synthetic-only", encoding="utf-8")
映射文件 = tmp_path / "映射.json"
映射文件.write_text(json.dumps(值, ensure_ascii=False), encoding="utf-8")
命令 = [
"导入目标",
"--快照",
str(保全 / "快照.json"),
"--映射",
str(映射文件),
"--配置",
str(配置),
"--允许实例",
str(参数["允许实例"]),
"--管理引用",
参数["管理引用"].位置,
"--目标回执",
str(参数["目标回执"]),
"--输出",
str(tmp_path / "导入报告.json"),
]
assert main(命令) == 1
正文 = 装配.要求正文()
assert 正文.列出候选(身份, "empty-chapter") == []
命令 += ["--文件回执", str(保全 / "文件回执.json")]
assert main(命令) == 0
候选 = 正文.列出候选(身份, "empty-chapter")
assert len(候选) == 1 and 候选[0]["base_revision"] == 0 and 候选[0]["origin"] == "author"
assert 正文.读取正文(身份, "empty-chapter")["revision"] == 0
assert "迁移候选" not in capsys.readouterr().out
assert json.loads((tmp_path / "导入报告.json").read_text())["states"] == {"mapped": 1}
def _旧载荷输入(装配, 身份, 载荷=None, *, 关系=False):
初, 映射 = _本书输入(装配, 身份)
正文 = 装配.要求正文()
正文.保存人工(
身份,
"save-old-shape-body",
"migration-chapter",
1,
正文草稿((段落("p1", (文本节点("林深坐在二号观测位。"),)),)),
)
文 = 正文.读取正文(身份, "migration-chapter")
数据 = 初.导出()
数据.pop("source_hash")
数据["record"].update(
work_id=12,
source_id=17473,
draft_type="relation" if 关系 else "entity",
draft_payload=载荷
if 载荷 is not None
else {
"type": "character",
"name": "林深",
"brief": "机师",
"fields": {"身份": "驾驶员"},
"source": {"workId": 12, "chapter": 1, "chapterId": 17473},
"evidence": "林深坐在二号观测位。",
},
)
源 = 旧快照.从载荷(数据)
映射["works"][0]["legacy_id"] = 12
映射["sources"][0].update(
legacy_id=17473,
work_id=12,
revision=str(文["revision"]),
chapters={"1": "chapter:migration-chapter"},
)
依据 = 正文依据(
"migration-chapter",
"main",
文["revision"],
文["document_hash"],
(
定位字面引文(
TypeAdapter(正文草稿).validate_python(文["document"]), "林深坐在二号观测位。"
),
),
)
映射["body_bindings"][0].update(source_hash=源.源哈希, sources=[asdict(依据)])
if 关系:
映射["schemas"] = {
"character_relation": {
"type": "object",
"properties": {},
"additionalProperties": False,
}
}
映射["target_schemas"] = {
"B02:character_relation": {
"schema_id": "character_relation",
"base_version": 1,
"fields": {"甲方": "source", "乙方": "target", "关系类型": "relation_type"},
}
}
return 源, 映射
@pytest.mark.case_id(
"TC-a01de71b60f7",
environment="合成原样例与已授权独立临时production/muse_app随机PG库",
given="原英文卡:work12/旧章17473/章序1、林深/机师/身份驾驶员、原字面证据句",
when="纯分流与隔离PG公共迁移入口导入、重放及真实回查",
then=[
"真实B02未决提案类型/名称/动态值与真实章依据正确,旧章ID与描述/证据保台账;重放无正式实体"
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("版本类型", [int, str])
def test_旧英文实体卡保字段与章来源而只建未决提案__a01de7(目标方法环境, 版本类型):
装配, 身份, _, _, 参数 = 目标方法环境
源, 值 = _旧载荷输入(装配, 身份)
值["body_bindings"][0]["sources"][0]["revision"] = 版本类型(2)
映射 = 映射配置(值)
一 = 导入快照(装配, 身份, [源], 映射, **参数)
assert 一["states"] == {"mapped": 1}
assert 导入快照(装配, 身份, [源], 映射, **参数) == 一
世界 = 装配.要求故事世界()
回执 = 世界.正式.读取回执(身份, 迁移命令ID(身份.作者, 源.source.记录键))
提案 = 世界.读取提案(身份, 回执["target_ref"])
assert 提案["content"] == {"名称": "林深", "身份定位": "驾驶员"}
assert (
提案["sources"][0]["chapter_id"] == "migration-chapter"
and 提案["sources"][0]["revision"] == 2
)
assert 提案["sources"][0]["evidence"][0]["quote"] == "林深坐在二号观测位。"
行 = 迁移台账(世界.数据库, 身份, 世界.正式).读取(源.source.记录键)
assert 行 is not None and 行["original"] == 源.导出()
assert 行["route"]["type_id"] == "character" and 行["route"]["content"]["brief"] == "机师"
assert 行["route"]["content"]["source_hint"] == {
"workId": 12,
"chapter": 1,
"chapterId": 17473,
"evidence": "林深坐在二号观测位。",
}
assert 世界.列出世界实体(身份, "migration-work") == []
assert 核对迁移结果(装配, 身份, [源], 映射, **参数)["data_checks_passed"] is True
@pytest.mark.case_id(
"TC-a423fe5888ef",
environment="合成原样例与已授权独立临时production/muse_app随机PG库",
given="原关系type驾驶/source.name林深/target.name茧/description首次实战驾驶;明确合法人物与不兼容物品两类前置",
when="纯分流与隔离PG公共迁移入口导入、重放及真实回查",
then=[
"合法目标形成真实关系候选,真实端点名称/类型不变;物品端在B02以FACT_REFERENCE_INVALID拒绝且无正式关系"
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("乙方类型", ["character", "item"])
@pytest.mark.parametrize("形状", ["对象", "名称", "编号"])
def test_旧关系名称按明确身份承接并守实际类型__a423fe(目标方法环境, 乙方类型, 形状):
from muse.故事世界.接口 import 事实定位
from muse.正式变更.接口 import 作者动作, 变更错误
from 知识记录分流 import 分流知识记录
装配, 身份, _, _, 参数 = 目标方法环境
载荷 = {
"type": "驾驶",
"source": {"name": "林深"},
"target": {"name": "茧"},
"description": "首次实战驾驶",
}
if 形状 == "名称":
载荷.update(source="林深", target="茧")
if 形状 == "编号":
载荷.update(source=11, target=12)
源, 值 = _旧载荷输入(装配, 身份, 载荷, 关系=True)
世界 = 装配.要求故事世界()
for id_, 名, 类型 in (("existing-a", "林深", "character"), ("existing-b", "茧", 乙方类型)):
回 = 世界.保存实体提案(
身份,
"seed-" + id_,
work_id="migration-work",
object_id=id_,
expected_revision=0,
schema_id=类型,
schema_version=1,
content={"名称": 名},
independent=True,
)
提 = 世界.读取提案(身份, 回["target_ref"])
定位 = 事实定位(回["target_ref"], 提["revision"], 提["proposal_hash"], "world:" + id_, 0)
审 = 世界.打开事实审阅(身份, 定位)
世界.决定事实(
身份,
"confirm-" + id_,
定位,
作者动作.采纳,
审["review_id"],
审["review_hash"],
("object",),
)
# 类型来自明确合成前置,不从名字或“驾驶”一词推断,也不由迁移器改类型。
值["objects"] = [
{
"system": "pg",
"database": "synthetic-legacy",
"work_id": 12,
"use": "author",
"scope": "migration-work",
"refs": ["name:" + 名, "entity:" + str(号)],
"target_ref": id_,
}
for 名, 号, id_ in (("林深", 11, "existing-a"), ("茧", 12, "existing-b"))
]
映射 = 映射配置(值)
分流 = 分流知识记录(源, 映射)
assert 分流.disposition == "待转换"
assert (
分流.content["source"],
分流.content["target"],
分流.content["relation_type"],
分流.content["brief"],
) == ("existing-a", "existing-b", "驾驶", "首次实战驾驶")
一 = 导入快照(装配, 身份, [源], 映射, **参数)
assert 一 == 导入快照(装配, 身份, [源], 映射, **参数)
命令 = 迁移命令ID(身份.作者, 源.source.记录键)
if 乙方类型 == "character":
assert 一["states"] == {"mapped": 1}
回执 = 世界.正式.读取回执(身份, 命令)
assert 世界.读取提案(身份, 回执["target_ref"])["content"] == {
"甲方": "existing-a",
"乙方": "existing-b",
"关系类型": "驾驶",
}
assert 核对迁移结果(装配, 身份, [源], 映射, **参数)["data_checks_passed"] is True
else:
assert (
一["states"] == {"quarantined": 1}
and 一["errors"][0]["reason_code"] == "FACT_REFERENCE_INVALID"
)
with pytest.raises(变更错误) as 错:
世界.正式.读取回执(身份, 命令)
assert 错.value.错误码 == "RECEIPT_NOT_FOUND"
端点 = {
e["object_id"]: (e["type_id"], e["content"]["名称"])
for e in 世界.列出世界实体(身份, "migration-work")
}
assert 端点 == {"existing-a": ("character", "林深"), "existing-b": (乙方类型, "茧")}
assert 世界.人物关系视图(身份, "migration-work")["关系"] == []
行 = 迁移台账(世界.数据库, 身份, 世界.正式).读取(源.source.记录键)
assert 行 is not None and 行["original"] == 源.导出()
@pytest.mark.case_id(
"TC-aaeff3e706c5",
environment="合成原样例与已授权独立临时production/muse_app随机PG库",
given="原缺target关系样例保source=林深、type=驾驶、description=x",
when="纯分流与隔离PG公共迁移入口导入、重放及真实回查",
then=[
"不被描述键或源名称遮蔽;分流/台账精确missing_endpoint与relation.target,S01无回执且无候选/正式关系"
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_旧缺端样例不被描述键遮蔽且精确失败关闭__aaeff3(目标方法环境):
from 知识记录分流 import 分流知识记录
装配, 身份, _, _, _ = 目标方法环境
源, 值 = _旧载荷输入(
装配, 身份, {"type": "驾驶", "source": "林深", "description": "x"}, 关系=True
)
值["objects"] = [
{
"system": "pg",
"database": "synthetic-legacy",
"work_id": 12,
"use": "author",
"scope": "migration-work",
"refs": ["name:林深"],
"target_ref": "existing-a",
}
]
判定 = 分流知识记录(源, 映射配置(值))
assert 判定.reason_code == "missing_endpoint" and 判定.detail == "relation.target"
行 = _核对保全且未发业务命令(目标方法环境, 源, 值, "quarantined", "missing_endpoint")
assert 行["route"]["detail"] == "relation.target"
assert 装配.要求故事世界().列出世界实体(身份, "migration-work") == []
assert 装配.要求故事世界().人物关系视图(身份, "migration-work")["关系"] == []
@pytest.mark.case_id(
"NC-w21-211001",
environment="独立临时production/muse_app随机PG库",
given="原作品/旧章声明冲突、错误引文、同版本跨章同句、布尔版本或非数组依据",
when="按新旧载荷合同经真实分流或隔离目标入口执行并回查",
then=[
"明确source_scope_conflict/target_source_mismatch,保原台账但零S01/领域写,不留reserved假成功"
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize(
"改动,原因",
[
("作品声明", "source_scope_conflict"),
("旧章声明", "source_scope_conflict"),
("错误引文", "target_source_mismatch"),
("跨章同句", "target_source_mismatch"),
("布尔版本", "target_source_mismatch"),
("坏依据集合", "target_source_mismatch"),
],
)
def test_旧章声明和真实依据不能各自正确却互相矛盾__211001(目标方法环境, 改动, 原因):
装配, 身份, _, _, _ = 目标方法环境
源, 值 = _旧载荷输入(装配, 身份)
数据 = 源.导出()
数据.pop("source_hash")
if 改动 == "作品声明":
数据["record"]["draft_payload"]["source"]["workId"] = 99
elif 改动 == "旧章声明":
数据["record"]["draft_payload"]["source"]["chapterId"] = 99
elif 改动 == "错误引文":
数据["record"]["draft_payload"]["evidence"] = "字面不同的引文"
elif 改动 == "布尔版本":
值["body_bindings"][0]["sources"][0]["revision"] = True
elif 改动 == "坏依据集合":
值["body_bindings"][0]["sources"] = True
else:
装配.要求作品().添加章节(
身份, "other-chapter", "migration-work", "other-chapter", "第二章", 预期目录版本=1
)
正文 = 装配.要求正文()
草稿 = 正文草稿((段落("p1", (文本节点("林深坐在二号观测位。"),)),))
for 版 in (0, 1):
正文.保存人工(身份, "other-body-" + str(版), "other-chapter", 版, 草稿)
文 = 正文.读取正文(身份, "other-chapter")
assert 文["revision"] == 2
值["body_bindings"][0]["sources"][0] = asdict(
正文依据(
"other-chapter",
"main",
文["revision"],
文["document_hash"],
(定位字面引文(草稿, "林深坐在二号观测位。"),),
)
)
源 = 旧快照.从载荷(数据)
值["body_bindings"][0]["source_hash"] = 源.源哈希
行 = _核对保全且未发业务命令(目标方法环境, 源, 值, "quarantined", 原因)
assert (
行["reason_code"] == 原因 and 装配.要求故事世界().列出世界实体(身份, "migration-work") == []
)
def _预先确认世界(装配, 身份, id_, 类型, 内容, 依据, *, 家族="entity", 版本=0, 动作=None):
from muse.故事世界.接口 import 事实定位, 事实提案请求
from muse.正式变更.接口 import 作者动作
世界 = 装配.要求故事世界()
结构 = 世界.新事实结构(身份, "migration-work", id_, 类型, 1, 家族=家族)
回 = 世界.提出事实(
身份,
"seed-proposal-" + id_ + str(版本),
事实提案请求(
work_id="migration-work",
object_id=id_,
expected_revision=版本,
schema=结构,
content=内容,
sources=tuple(TypeAdapter(正文依据).validate_python(x) for x in 依据),
effective_from="migration-chapter",
instance_family=家族,
independent=False,
),
)
if 动作 is 作者动作.人工保存:
return 回
提 = 世界.读取提案(身份, 回["target_ref"])
定位 = 事实定位(提["proposal_id"], 提["revision"], 提["proposal_hash"], "world:" + id_, 版本)
审 = 世界.打开事实审阅(身份, 定位)
return 世界.决定事实(
身份,
"seed-adopt-" + id_ + str(版本),
定位,
作者动作.采纳 if 动作 is None else 动作,
审["review_id"],
审["review_hash"],
("object",),
)
def _已确认历史输入(装配, 身份, *, 关系=False):
初, 值 = _旧载荷输入(装配, 身份)
依据 = 值["body_bindings"][0]["sources"]
if 关系:
for id_, 名 in (("old-end-a", "林深"), ("old-end-b", "茧")):
_预先确认世界(装配, 身份, id_, "character", {"名称": 名}, 依据)
新ID = "linked-relation" if 关系 else "linked-entity"
类型 = "character_relation" if 关系 else "character"
内容 = (
{"甲方": "old-end-a", "乙方": "old-end-b", "关系类型": "驾驶"}
if 关系
else {"名称": "林深", "身份定位": "驾驶员"}
)
回执 = _预先确认世界(装配, 身份, 新ID, 类型, 内容, 依据, 家族="relation" if 关系 else "entity")
正式行 = {
"id": 91,
"work_id": 12,
"revision": 1,
"deleted": False,
"source_status": "active",
"source_action_policy": "allowed",
"description": "首次实战驾驶" if 关系 else "机师",
"attributes": None if 关系 else {"身份": "驾驶员"},
}
if 关系:
正式行.update(source_entity_id=11, target_entity_id=12, relation_type="驾驶")
else:
正式行.update(
entity_type="character",
normalized_name="林深",
status="active",
source_type=None,
source_id=None,
lineage_payload={"sourceRefs": [{"chapter": 1}]},
)
原目标 = 旧快照.从载荷(
{
"source": {
**初.导出()["source"],
"table": "muse_knowledge_relation" if 关系 else "muse_knowledge_entity",
"id": "91",
},
"record": 正式行,
}
)
草稿行 = {
**初.原行,
"id": 7,
"revision": 2,
"draft_type": "relation" if 关系 else "entity",
"draft_payload": {},
"status": "confirmed",
"confidence": 0.9,
"entity_id": None if 关系 else 91,
"target_object_id": 91 if 关系 else None,
}
草稿 = 旧快照.从载荷(
{"source": {**初.导出()["source"], "id": "7", "revision": "2"}, "record": 草稿行}
)
新确认 = {"command_id": 回执["command_id"], "receipt_id": 回执["receipt_id"], "revision": 1}
值["confirmations"] = [
{
"system": "pg",
"database": "synthetic-legacy",
"source_key": s.source.记录键,
"source_hash": s.源哈希,
"target_ref": 新ID,
"decision_ref": "old-decision:" + s.source.id,
"existing_confirmation": dict(新确认),
}
for s in (原目标, 草稿)
]
值["confirmations"][1]["target_source"] = {
"source_key": 原目标.source.记录键,
"source_hash": 原目标.源哈希,
}
域 = {
"system": "pg",
"database": "synthetic-legacy",
"work_id": 12,
"use": "author",
"scope": "migration-work",
}
值["objects"] = [
{**域, "refs": [("relation:" if 关系 else "entity:") + "91"], "target_ref": 新ID}
]
if 关系:
值["objects"] += [
{**域, "refs": ["entity:" + str(i)], "target_ref": id_}
for i, id_ in ((11, "old-end-a"), (12, "old-end-b"))
]
值["schemas"] = {类型: {"type": "object", "properties": {}, "additionalProperties": False}}
值["target_schemas"] = {
"B02:" + 类型: {
"schema_id": 类型,
"base_version": 1,
"fields": {"甲方": "source", "乙方": "target", "关系类型": "relation_type"},
}
}
return 原目标, 草稿, 值, 回执
def _业务与索引指纹(装配):
# 只在隔离测试库聚合指纹,不把SQL放入迁移产品代码,不读取旧embedding表。
表 = (
"muse_world_object",
"muse_world_version",
"muse_fact_proposal",
"muse_fact_proposal_version",
"muse_author_review",
"muse_candidate_decision",
"muse_change_command",
"muse_change_impact",
"muse_document_index",
"muse_method_index",
"muse_method_index_version",
)
with 装配.要求故事世界().数据库.连接(只读=True) as 连:
return {
名: 连.execute(
sql.SQL(
"SELECT count(*),md5(coalesce(string_agg(j,',' ORDER BY j),'')) "
"FROM (SELECT row_to_json(t)::text AS j FROM {} t) s"
).format(sql.Identifier(名))
).fetchone()
for 名 in 表
}
@pytest.mark.case_id(
"TC-knowledge-confirmed-idempotent",
environment="合成旧输入、独立临时production/muse_app随机PG目标",
given="实体/关系的旧正式源与已确认草稿、预存的真实S01采纳目标,两个输入顺序重复导入",
when="明确前置完成后才测迁移:导入/重放/点名/真实只读对账及全业务和B09指纹比较",
then=[
"两来源同目标/原版本/真实确认,旧声明保全;台账mapped不新写S01/世界/候选/B09,反复结果稳定",
"仅验证已有真实新目标及S01采纳证据时的关联,不证明旧decision_ref真实性,不等于从空库恢复旧批准;旧草稿缺当时目标版本不倒填时间线。",
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("关系", [False, True])
@pytest.mark.parametrize("草稿在前", [False, True])
def test_已确认双源只关联既存版本和真实决定不复制__212000(目标方法环境, 关系, 草稿在前):
from muse.正式变更.接口 import 变更错误
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, 原回 = _已确认历史输入(装配, 身份, 关系=关系)
输入 = [草稿, 正式源] if 草稿在前 else [正式源, 草稿]
前 = _业务与索引指纹(装配)
映射 = 映射配置(值)
一 = 导入快照(装配, 身份, 输入, 映射, **参数)
assert 一["states"] == {"mapped": 2}
assert 一 == 导入快照(装配, 身份, 输入, 映射, **参数)
assert len(一["history_links"]) == 2
assert {r["target_ref"] for r in 一["history_links"]} == {值["confirmations"][0]["target_ref"]}
assert {r["receipt_id"] for r in 一["history_links"]} == {原回["receipt_id"]}
正式 = 装配.要求故事世界().正式
账 = 迁移台账(正式.数据库, 身份, 正式)
for s in 输入:
行 = 账.读取(s.source.记录键)
assert 行 is not None and 行["original"] == s.导出() and 行["receipt"] == 原回
assert (
行["frozen_request"]["mode"] == "link_confirmed"
and 行["targets"][0]["kind"] == "world_object"
)
assert 行["frozen_request"]["payload"]["decision_ref"] == "old-decision:" + s.source.id
with pytest.raises(变更错误) as 错:
正式.读取回执(身份, 行["command_id"])
assert 错.value.错误码 == "RECEIPT_NOT_FOUND"
assert 核对迁移结果(装配, 身份, 输入, 映射, **参数)["data_checks_passed"] is True
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"TC-ef5133dabe23",
environment="合成旧输入、独立临时production/muse_app随机PG目标",
given="旧草稿7/work12/chapter17473/revision2/target_object_id91且payload={},明确正式旧关系源无source_type/source_id",
when="明确前置完成后才测迁移:导入/重放/点名/真实只读对账及全业务和B09指纹比较",
then=[
"单独点名重放回读旧7→91→明确新目标版/真实回执,不再创建/确认,不补原payload或缺列",
"仅验证已有真实新目标及S01采纳证据时的关联,不证明旧decision_ref真实性,不等于从空库恢复旧批准;旧草稿缺当时目标版本不倒填时间线。",
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_已确认空关系草稿点名重放保旧7到91__ef5133(目标方法环境):
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, 回 = _已确认历史输入(装配, 身份, 关系=True)
assert 草稿.原行["draft_payload"] == {} and 草稿.原行["target_object_id"] == 91
assert not {"source_type", "source_id"} & set(正式源.原行)
映射 = 映射配置(值)
导入快照(装配, 身份, [正式源, 草稿], 映射, **参数)
前 = _业务与索引指纹(装配)
一 = 导入快照(装配, 身份, [草稿], 映射, **参数)
assert 一["states"] == {"mapped": 1} and 一 == 导入快照(装配, 身份, [草稿], 映射, **参数)
h = 一["history_links"][0]
assert (
h["legacy_draft_id"],
h["legacy_target_id"],
h["target_ref"],
h["revision"],
h["receipt_id"],
) == ("7", "91", "linked-relation", 1, 回["receipt_id"])
assert h["kind"] == "linked_existing_confirmation"
assert 核对迁移结果(装配, 身份, [草稿], 映射, **参数)["data_checks_passed"] is True
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-212001",
environment="合成输入与独立临时production/muse_app随机PG库",
given="假命令/回执/版本、缺正式源/错源hash/错旧主键或pending伪装",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=[
"精确existing_confirmation_invalid/legacy_link_conflict/legacy_target_source_missing,原值保全、零业务变化"
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize(
"错误", ["假命令", "假回执", "错版本", "缺正式源", "错正式哈希", "错旧主键", "pending伪装"]
)
def test_历史关联不能只凭旧声明或伪指针转正__212001(目标方法环境, 错误):
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份, 关系=True)
if 错误 == "假命令":
for r in 值["confirmations"]:
r["existing_confirmation"]["command_id"] = "no-such-command"
elif 错误 == "假回执":
for r in 值["confirmations"]:
r["existing_confirmation"]["receipt_id"] = "fake-receipt"
elif 错误 == "错版本":
for r in 值["confirmations"]:
r["existing_confirmation"]["revision"] = 2
elif 错误 == "错正式哈希":
值["confirmations"][1]["target_source"]["source_hash"] = "0" * 64
elif 错误 in ("错旧主键", "pending伪装"):
d = 草稿.导出()
d.pop("source_hash")
if 错误 == "错旧主键":
d["record"]["target_object_id"] = 92
else:
d["record"]["status"] = "pending"
草稿 = 旧快照.从载荷(d)
值["confirmations"][1]["source_hash"] = 草稿.源哈希
输入 = [草稿] if 错误 == "缺正式源" else [草稿, 正式源]
前 = _业务与索引指纹(装配)
一 = 导入快照(装配, 身份, 输入, 映射配置(值), **参数)
assert 一["states"].get("quarantined", 0) >= 1
行 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式).读取(
草稿.source.记录键
)
assert (
行 is not None
and 行["state"] == "quarantined"
and 行["targets"] == []
and 行["receipt"] is None
)
assert 行["original"] == 草稿.导出() and _业务与索引指纹(装配) == 前
assert 行["reason_code"] == (
"legacy_target_source_missing"
if 错误 in ("假命令", "假回执", "错版本", "缺正式源")
else "legacy_link_conflict"
)
if 错误 in ("假命令", "假回执", "错版本"):
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
父 = 账.读取(正式源.source.记录键)
assert 父 is not None and 父["reason_code"] == "existing_confirmation_invalid"
@pytest.mark.case_id(
"NC-w21-212002",
environment="合成输入与独立临时production/muse_app随机PG库",
given="正式源或草稿保存终态前合成中断,正式源仍reserved时点名草稿",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["父处理中报legacy_dependency_pending并保reserved;恢复只补台账,无新S01/世界/索引"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("断点", ["正式", "草稿"])
def test_关联中断和父源处理中保持可重试且不写业务__212002(目标方法环境, monkeypatch, 断点):
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份, 关系=True)
原保存 = 迁移台账._保存关联
指定 = (正式源 if 断点 == "正式" else 草稿).source.记录键
def 中断(self, key, 目标, 回执):
if key == 指定:
raise RuntimeError("合成中断")
return 原保存(self, key, 目标, 回执)
monkeypatch.setattr(迁移台账, "_保存关联", 中断)
前 = _业务与索引指纹(装配)
映射 = 映射配置(值)
with pytest.raises(RuntimeError, match="合成中断"):
导入快照(装配, 身份, [草稿, 正式源], 映射, **参数)
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
行 = 账.读取(指定)
assert (
行 is not None
and 行["state"] == "reserved"
and 行["frozen_request"]["mode"] == "link_confirmed"
)
if 断点 == "正式":
with pytest.raises(迁移错误) as 错:
导入快照(装配, 身份, [草稿], 映射, **参数)
assert 错.value.code == "legacy_dependency_pending"
子 = 账.读取(草稿.source.记录键)
assert 子 is not None and 子["state"] == "reserved"
monkeypatch.setattr(迁移台账, "_保存关联", 原保存)
assert 导入快照(装配, 身份, [草稿, 正式源], 映射, **参数)["states"] == {"mapped": 2}
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-212003",
environment="合成输入与独立临时production/muse_app随机PG库",
given="双线程以相反输入顺序关联同一正式源和草稿",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["统一锁序、相同结果、同目标确认,无业务/索引变化"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_历史双源并发按统一锁序仅关联一次__212003(目标方法环境):
from concurrent.futures import ThreadPoolExecutor
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份, 关系=True)
前 = _业务与索引指纹(装配)
def 执行(反序):
return 导入快照(
装配, 身份, [草稿, 正式源] if 反序 else [正式源, 草稿], 映射配置(值), **参数
)
with ThreadPoolExecutor(max_workers=2) as 池:
结果 = list(池.map(执行, [False, True]))
assert 结果[0] == 结果[1] and 结果[0]["states"] == {"mapped": 2}
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-212004",
environment="合成输入与独立临时production/muse_app随机PG库",
given="目标世界或正文后来编辑,旧源同版改成撤回及新版本撤回",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["原历史版/回执不变,同版drift拒写,新撤回只历史留存,不倒改或复活"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("修改", ["世界", "正文"])
def test_后续编辑与新源撤回不倒改已关联历史__212004(目标方法环境, 修改):
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, 原回 = _已确认历史输入(装配, 身份)
映射 = 映射配置(值)
一 = 导入快照(装配, 身份, [正式源, 草稿], 映射, **参数)
if 修改 == "世界":
_预先确认世界(
装配,
身份,
"linked-entity",
"character",
{"名称": "林深", "身份定位": "新的身份"},
值["body_bindings"][0]["sources"],
版本=1,
)
else:
装配.要求正文().保存人工(
身份,
"change-history-source",
"migration-chapter",
2,
正文草稿((段落("p1", (文本节点("正文内容已经改变。"),)),)),
)
前 = _业务与索引指纹(装配)
assert 导入快照(装配, 身份, [正式源, 草稿], 映射, **参数) == 一
assert {r["revision"] for r in 一["history_links"]} == {1}
d = 草稿.导出()
d.pop("source_hash")
d["record"]["source_status"] = "withdrawn"
同版 = 旧快照.从载荷(d)
with pytest.raises(迁移错误) as 错:
导入快照(装配, 身份, [同版], 映射, **参数)
assert 错.value.code == "drift"
d["source"]["revision"] = "3"
d["record"]["revision"] = 3
新版 = 旧快照.从载荷(d)
assert 导入快照(装配, 身份, [新版], 映射, **参数)["states"] == {"historical": 1}
assert (
导入快照(装配, 身份, [草稿], 映射, **参数)["history_links"][0]["receipt_id"]
== 原回["receipt_id"]
)
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-212005",
environment="合成输入与独立临时production/muse_app随机PG库",
given="链接冻结降create/未知mode、缺世界核验能力、伪proposal目标、pending原行伪link路由",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["冻结严格拒切换,存储必须核实际世界及原行,不凭传入JSON/state跳验证"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_冻结模式原行与世界读取能力不能靠自报绕过__212005(目标方法环境):
from copy import deepcopy
from 知识记录分流 import 分流结果
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份)
导入快照(装配, 身份, [正式源, 草稿], 映射配置(值), **参数)
世界 = 装配.要求故事世界()
账 = 迁移台账(世界.数据库, 身份, 世界.正式, 世界)
行 = 账.读取(草稿.source.记录键)
assert 行 is not None
前 = _业务与索引指纹(装配)
for 模式 in ("create", "unknown", [], {"mode": "link_confirmed"}):
冻结 = deepcopy(行["frozen_request"])
冻结["mode"] = 模式
with pytest.raises(迁移错误) as 错:
账.冻结请求(草稿.source.记录键, 冻结)
assert 错.value.code == "target_request_invalid"
缺世界 = 迁移台账(世界.数据库, 身份, 世界.正式)
with pytest.raises(迁移错误) as 错:
缺世界.保存终态(草稿.source.记录键, "mapped", 目标=行["targets"], 回执=行["receipt"])
assert 错.value.code == "target_evidence_missing"
坏目标 = deepcopy(行["targets"])
坏目标[0]["kind"] = "proposal"
with pytest.raises(迁移错误) as 错:
账.保存终态(草稿.source.记录键, "mapped", 目标=坏目标, 回执=行["receipt"])
assert 错.value.code == "target_evidence_mismatch"
d = 草稿.导出()
d.pop("source_hash")
d["source"]["id"] = "88"
d["record"].update(id=88, status="pending")
待审 = 旧快照.从载荷(d)
伪路由 = {
**deepcopy(行["route"]),
"source_key": 待审.source.记录键,
"source_hash": 待审.源哈希,
"original": 待审.导出(),
}
账.占用(待审, 行["mapping_hash"], 分流结果(**伪路由))
冻结 = deepcopy(行["frozen_request"])
冻结["payload"].update(source_key=待审.source.记录键, source_hash=待审.源哈希)
with pytest.raises(迁移错误) as 错:
账.冻结请求(待审.source.记录键, 冻结)
assert 错.value.code == "legacy_link_conflict"
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-212006",
environment="合成输入与独立临时production/muse_app随机PG库",
given="同一实际只读事务连接中读取S01回执和B02历史,工厂另开连接被禁止",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["两个公开于接口使用调用方连接,保持INTRANS,不关闭/提交/回滚"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_同事务只读入口不另开连接且保持事务归调用方__212006(目标方法环境, monkeypatch):
装配, 身份, _, _, _ = 目标方法环境
_, _, _, 回 = _已确认历史输入(装配, 身份)
世界 = 装配.要求故事世界()
with 世界.数据库.连接(只读=True) as 连, 连.transaction():
def 禁新连接(*args, **kwargs):
raise AssertionError("只读于入口不能另开连接")
with monkeypatch.context() as m:
m.setattr(数据库工厂, "连接", 禁新连接)
assert 世界.正式.读取回执于(连, 身份, 回["command_id"]) == 回
历史 = 世界.读取实体轨迹于(连, 身份, "migration-work", "linked-entity")
assert (
历史["current_revision"] == 1
and 历史["版本"][0]["content_hash"] == 回["results"][0]["content_hash"]
)
assert not 连.closed and 连.info.transaction_status.name == "INTRANS"
@pytest.mark.case_id(
"NC-w21-212007",
environment="合成输入与独立临时production/muse_app随机PG库",
given="旧正式内容、S04类型或明确另一作品与预存目标不一致",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["existing_target_mismatch/target_type_mismatch,根与草稿均拒,原业务/索引不变"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("差异", ["内容", "结构", "作用域"])
def test_旧正式值与真实目标不符必须精确拒绝__212007(目标方法环境, 差异):
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份)
if 差异 == "内容":
d = 正式源.导出()
d.pop("source_hash")
d["record"]["attributes"] = {"身份": "与已确认值不同"}
正式源 = 旧快照.从载荷(d)
值["confirmations"][0]["source_hash"] = 正式源.源哈希
值["confirmations"][1]["target_source"]["source_hash"] = 正式源.源哈希
elif 差异 == "结构":
值["target_schemas"]["B02:character"]["schema_id"] = "item"
else:
作品 = 装配.要求作品()
结构 = 作品.新档案结构(身份, "other-work", "work_core", 1)
作品.保存档案(身份, "other-work", 档案保存("other-work", 0, {"名称": "明确另一作品"}, 结构))
值["works"][0]["target_ref"] = "other-work"
for o in 值["objects"]:
o["scope"] = "other-work"
前 = _业务与索引指纹(装配)
一 = 导入快照(装配, 身份, [草稿, 正式源], 映射配置(值), **参数)
assert 一["states"] == {"quarantined": 2}
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
行 = 账.读取(正式源.source.记录键)
assert 行 is not None
assert 行["reason_code"] == (
"target_type_mismatch" if 差异 == "结构" else "existing_target_mismatch"
)
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-212008",
environment="合成输入与独立临时production/muse_app随机PG库",
given="真实同作者manual_save/reject/defer回执指向候选或对象",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["不得冒充adopt,existing_confirmation_invalid,不能因回执存在便关联"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("动作名", ["人工保存", "拒绝", "暂缓"])
def test_真实非采纳回执不能冒充已确认版本__212008(目标方法环境, 动作名):
from muse.正式变更.接口 import 作者动作
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份)
回 = _预先确认世界(
装配,
身份,
"linked-entity",
"character",
{"名称": "林深", "身份定位": "驾驶员"},
值["body_bindings"][0]["sources"],
版本=1,
动作=getattr(作者动作, 动作名),
)
for r in 值["confirmations"]:
r["existing_confirmation"] = {
"command_id": 回["command_id"],
"receipt_id": 回["receipt_id"],
"revision": 1,
}
前 = _业务与索引指纹(装配)
assert 导入快照(装配, 身份, [草稿, 正式源], 映射配置(值), **参数)["states"] == {
"quarantined": 2
}
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
行 = 账.读取(正式源.source.记录键)
assert 行 is not None and 行["reason_code"] == "existing_confirmation_invalid"
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-212009",
environment="合成输入与独立临时production/muse_app随机PG库",
given="record.id与source.id不一致,或类型别名值为数组",
when="在真实隔离目标通过公共用例/迁移台账验证并回查原数据和业务指纹",
then=["原主键在旧快照门禁source_invalid拒绝;坏别名unknown_type隔离,无未捕获TypeError"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("差异", ["原主键", "别名坏类型"])
def test_原主键与类型映射不一致不能关联__212009(目标方法环境, 差异):
装配, 身份, _, _, 参数 = 目标方法环境
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份)
前 = _业务与索引指纹(装配)
if 差异 == "原主键":
d = 正式源.导出()
d.pop("source_hash")
d["record"]["id"] = 92
with pytest.raises(迁移错误) as 错:
旧快照.从载荷(d)
assert 错.value.code == "source_invalid"
else:
值["type_aliases"] = {"character": ["character"]}
结果 = 导入快照(装配, 身份, [正式源, 草稿], 映射配置(值), **参数)
assert 结果["states"] == {"quarantined": 2}
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
父 = 账.读取(正式源.source.记录键)
assert 父 is not None and 父["reason_code"] == "unknown_type"
assert _业务与索引指纹(装配) == 前
def _来源回放输入(装配, 身份, *, 坏来源=False):
from copy import deepcopy
正式源, 草稿, 值, _ = _已确认历史输入(装配, 身份)
d = 正式源.导出()
d.pop("source_hash")
d["record"]["lineage_payload"] = {"sourceRefs": [{"chapter": 1}], "lastDraftId": 10}
正式源 = 旧快照.从载荷(d)
根指针 = {"source_key": 正式源.source.记录键, "source_hash": 正式源.源哈希}
值["confirmations"][0]["source_hash"] = 正式源.源哈希
原子声明 = deepcopy(值["confirmations"][1])
值["confirmations"] = 值["confirmations"][:1]
子 = []
for id_, 章 in ((11, 1), (12, 2)):
d = 草稿.导出()
d.pop("source_hash")
d["source"]["id"] = str(id_)
d["record"].update(
id=id_,
source_id=17472 + 章,
draft_payload={"source": "非结构化提示" if 坏来源 and 章 == 2 else {"chapter": 章}},
)
s = 旧快照.从载荷(d)
子.append(s)
值["confirmations"].append(
{
**deepcopy(原子声明),
"source_key": s.source.记录键,
"source_hash": s.源哈希,
"target_source": dict(根指针),
}
)
请求 = {
"base": 根指针,
"steps": [{"source_key": s.source.记录键, "source_hash": s.源哈希} for s in 子],
}
return [正式源, *子], 映射配置(值), 请求
@pytest.mark.case_id(
"TC-b671f4f0c215",
environment="合成旧源及独立临时production/muse_app随机PG库",
given="初始sourceRefs=[{chapter:1}]/lastDraftId10,明确步骤草稿11引用chapter1、草稿12引用chapter2;所有原源在PG留存",
when="经真实核对用例和CLI读取PG旧原行逐步重放、复验且核无业务/索引变化",
then=[
"第一步sourceRefs不重复;第二步追加chapter2;最终lastDraftId12,逐步结果互不串改;原值不变",
"按调用方明确顺序,从真实PG已关联旧源试算追加结果,不认证旧时序,不重写当前B02来源或生成Canonical;旧真实批准/时序未证内容仍仅历史声明。",
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_旧来源链按明确请求追加去重且不改正式数据__b671f4(目标方法环境, tmp_path, capsys):
装配, 身份, _, _, 参数 = 目标方法环境
输入, 映射, 请求 = _来源回放输入(装配, 身份)
assert 导入快照(装配, 身份, 输入, 映射, **参数)["states"] == {"mapped": 3}
前 = _业务与索引指纹(装配)
结果 = 核对迁移结果(装配, 身份, 输入, 映射, 来源回放=请求, **参数)
回放 = 结果["lineage_replay"]
assert 结果["data_checks_passed"] is True and 回放["passed"] is True
assert 回放["initial"] == {"sourceRefs": [{"chapter": 1}], "lastDraftId": 10}
assert 回放["steps"][0]["lineage"]["sourceRefs"] == [{"chapter": 1}]
assert 回放["steps"][0]["lineage"]["lastDraftId"] == 11
assert 回放["final"]["sourceRefs"] == [{"chapter": 1}, {"chapter": 2}]
assert (
回放["final"]["lastDraftId"] == 12 and 回放["final"]["lastSourceType"] == "chapter_extract"
)
assert (
回放["order_basis"] == "caller_selected_order"
and 回放["historical_order_verified"] is False
and 回放["canonical_written"] is False
)
assert 结果 == 核对迁移结果(装配, 身份, 输入, 映射, 来源回放=请求, **参数)
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
for s in 输入:
行 = 账.读取(s.source.记录键)
assert 行 is not None and 行["original"] == s.导出()
from 入口 import main
配置 = tmp_path / "应用.toml"
配置.write_text(
'["数据库"]\n"取值方式"="受控存储"\n"位置"='
+ json.dumps(装配.配置.数据库.位置, ensure_ascii=False)
+ '\n["资源"]\n"发布身份"="synthetic"\n["运行"]\n"用途"="production"'
'\n[HTTP]\n"作者ID"="migration-author"\n"口令文件"="口令.txt"\n',
encoding="utf-8",
)
(tmp_path / "口令.txt").write_text("synthetic-only", encoding="utf-8")
for 名, 值 in (("快照", [s.导出() for s in 输入]), ("映射", 映射.值), ("回放", 请求)):
(tmp_path / (名 + ".json")).write_text(json.dumps(值, ensure_ascii=False), encoding="utf-8")
命令 = [
"核对目标",
"--快照",
str(tmp_path / "快照.json"),
"--映射",
str(tmp_path / "映射.json"),
"--配置",
str(配置),
"--允许实例",
str(参数["允许实例"]),
"--管理引用",
参数["管理引用"].位置,
"--目标回执",
str(参数["目标回执"]),
"--来源回放",
str(tmp_path / "回放.json"),
"--输出",
str(tmp_path / "报告.json"),
]
assert main(命令 + ["--输出", str(tmp_path / "回放.json")]) == 1
assert json.loads((tmp_path / "回放.json").read_text()) == 请求
assert main(命令) == 0
assert json.loads((tmp_path / "报告.json").read_text())["lineage_replay"] == 回放
assert "林深" not in capsys.readouterr().out
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-213001",
environment="合成旧源与隔离PG,只读试算",
given="引用错hash/错根、非对象来源或步骤未留存",
when="由真实核对入口处理明确查询请求",
then=["来源回放失败且final=None、总核对失败;原记录及业务/索引不变"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("错误", ["错hash", "错根", "非对象引用", "缺步骤源"])
def test_回放必须核真实留存来源且错误不清空原值__213001(目标方法环境, 错误):
装配, 身份, _, _, 参数 = 目标方法环境
输入, 映射, 请求 = _来源回放输入(装配, 身份, 坏来源=错误 == "非对象引用")
导入快照(装配, 身份, 输入, 映射, **参数)
if 错误 == "错hash":
请求["steps"][0]["source_hash"] = "0" * 64
elif 错误 == "错根":
请求["base"] = dict(请求["steps"][0])
elif 错误 == "缺步骤源":
请求["steps"][1]["source_key"] = "not-stored"
前 = _业务与索引指纹(装配)
结果 = 核对迁移结果(装配, 身份, 输入, 映射, 来源回放=请求, **参数)
assert 结果["data_checks_passed"] is False and 结果["lineage_replay"]["passed"] is False
assert 结果["lineage_replay"]["issues"][0]["reason_code"] == "legacy_lineage_invalid"
assert 结果["lineage_replay"]["final"] is None and _业务与索引指纹(装配) == 前
def _参考分析输入(快照, 配置, 身份):
配置 = json.loads(json.dumps(配置))
数据 = 快照.导出()
数据.pop("source_hash")
数据["record"].update(
work_id=8,
draft_payload={
"type": "character",
"name": "林深",
"brief": "机师",
"fields": {"身份": "驾驶员"},
},
)
源 = 旧快照.从载荷(数据)
配置["works"][0].update(
legacy_id=8, use="reference", target_ref=配置["sources"][0]["target_ref"]
)
配置["schemas"] = {
"character": {
"type": "object",
"properties": {"身份": {"type": "string"}},
"additionalProperties": False,
}
}
配置["target_schemas"] = {
"B03:character": {
"schema_id": "character",
"base_version": 1,
"fields": {"名称": "name", "身份定位": "fields.身份"},
}
}
return 源, 映射配置(配置)
@pytest.mark.case_id(
"TC-knowledge-reference-entity",
environment="合成旧源及隔离production/muse_app随机PG库",
given="参考来源中的character原卡、真实来源版本及显式S04映射",
when="真实隔离PG导入、重放与逐项对账",
then=[
"B03不可变分析保规范内容/描述/旧源声明;S01只有未决保存回执",
"无研究任务、作者世界、批准或B09索引写入;相同输入重放不复制",
],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
def test_参考人物真实进入B03分析且不造任务或世界事实__214000(目标方法环境):
装配, 身份, 快照, 配置, 参数 = 目标方法环境
源, 映射 = _参考分析输入(快照, 配置, 身份)
前 = _业务与索引指纹(装配)
assert 装配.任务运行.列出任务(身份.作者) == []
结果 = 导入快照(装配, 身份, [源], 映射, **参数)
assert 结果["states"] == {"mapped": 1}
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
行 = 账.读取(源.source.记录键)
目标 = 行["targets"][0]
assert 目标["kind"] == "analysis" and 目标["owner"] == "B03" and 目标["revision"] == 1
with 装配.要求数据库().连接(只读=True) as 连:
分析 = 装配.要求资料().读取分析(连, 身份.作者, 目标["id"])
assert (
分析["origin"] == "legacy_import"
and 分析["source_id"] == 映射.值["works"][0]["target_ref"]
)
assert 分析["payload"]["request"]["content"] == {"名称": "林深", "身份定位": "驾驶员"}
assert 分析["payload"]["request"]["brief"] == "机师"
assert (
分析["payload"]["status"] == "proposed" and 分析["payload"]["model_verified"] is False
)
assert 分析["payload"]["canonical_written"] is False
assert 行["receipt"]["action"] == "manual_save" and 行["receipt"]["author_review_id"] is None
assert 行["original"] == 源.导出()
assert 导入快照(装配, 身份, [源], 映射, **参数) == 结果
assert 核对迁移结果(装配, 身份, [源], 映射, **参数)["data_checks_passed"] is True
后 = _业务与索引指纹(装配)
assert 后.pop("muse_change_command")[0] == 前.pop("muse_change_command")[0] + 1
assert 后 == 前 and 装配.任务运行.列出任务(身份.作者) == []
@pytest.mark.case_id(
"NC-w21-214001",
environment="隔离PG及合成提供方,不访问真实资料或外部模型",
given="固定来源与合法参考卡,分别改变版本/哈希/作者/类型族/字段/来源身份",
when="调用真实领域/CLI/迁移入口",
then=["拒绝坏依据,零S01/分析/任务/世界新写"],
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
)
@pytest.mark.parametrize(
"坏处", ["错版本", "错哈希", "错作者", "错类型族", "未知字段", "坏来源身份"]
)
def test_参考分析拒绝坏依据且零领域和任务副作用__214001(目标方法环境, 坏处):
from dataclasses import replace
from muse.资料研究.接口 import 参考分析请求
from 知识记录分流 import 分流知识记录
from 转换业务对象 import 准备目标请求
装配, 身份, 快照, 配置, _ = 目标方法环境
源, 映射 = _参考分析输入(快照, 配置, 身份)
冻结 = 准备目标请求(装配, 身份, 源, 分流知识记录(源, 映射), 映射)
请求 = TypeAdapter(参考分析请求).validate_python(冻结["payload"])
前 = _业务与索引指纹(装配)
with pytest.raises(Muse错误):
if 坏处 == "错版本":
请求 = replace(请求, source_revision=99)
elif 坏处 == "错哈希":
请求 = replace(请求, source_content_hash="a" * 64)
elif 坏处 == "错作者":
身份 = replace(身份, 作者="foreign-author")
elif 坏处 == "错类型族":
请求 = replace(请求, type_id="craft", schema_id="craft")
elif 坏处 == "未知字段":
请求 = replace(请求, content={"名称": "甲", "未定义": "不能藏值"})
else:
请求 = replace(请求, source_id="not-a-source-id")
装配.要求资料().提出参考分析(身份, "bad-analysis", 请求)
with 装配.要求数据库().连接(只读=True) as 连:
assert 装配.要求资料().列出分析(连, "migration-author", 冻结["scope"]) == []
assert _业务与索引指纹(装配) == 前
assert 装配.任务运行.列出任务("migration-author") == []
@pytest.mark.case_id(
"NC-w21-214002",
environment="隔离PG及合成提供方,不访问真实资料或外部模型",
given="同源并发及同ID不同命令/标题",
when="调用真实领域/CLI/迁移入口",
then=["并发只建一个分析;异请求不覆盖原值或另换命令绕过"],
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
)
def test_参考分析并发幂等及异命令不覆盖__214002(目标方法环境):
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from muse.资料研究.接口 import 参考分析请求
装配, 身份, 快照, 配置, 参数 = 目标方法环境
源, 映射 = _参考分析输入(快照, 配置, 身份)
with ThreadPoolExecutor(max_workers=2) as 池:
结果 = list(池.map(lambda _: 导入快照(装配, 身份, [源], 映射, **参数), range(2)))
assert all(r["states"] == {"mapped": 1} for r in 结果)
账 = 迁移台账(装配.要求故事世界().数据库, 身份, 装配.要求故事世界().正式)
行 = 账.读取(源.source.记录键)
请求 = TypeAdapter(参考分析请求).validate_python(行["frozen_request"]["payload"])
前 = _业务与索引指纹(装配)
with pytest.raises(Muse错误):
装配.要求资料().提出参考分析(
身份, "new-command-for-same-id", replace(请求, title="不能覆盖")
)
assert _业务与索引指纹(装配) == 前
with 装配.要求数据库().连接(只读=True) as 连:
assert len(装配.要求资料().列出分析(连, 身份.作者, 请求.source_id)) == 1
@pytest.mark.case_id(
"NC-w21-214003",
environment="隔离PG及合成提供方,不访问真实资料或外部模型",
given="真实参考人物/势力未决卡与显式引用",
when="调用真实领域/CLI/迁移入口",
then=["目标类型和声明状态均须满足;未决卡不冒充confirmed"],
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
)
def test_参考引用不能跨类型或把未决卡当确认对象__214003(目标方法环境):
from dataclasses import replace
from muse.资料研究.接口 import 参考分析请求
from 知识记录分流 import 分流知识记录
from 转换业务对象 import 准备目标请求
装配, 身份, 快照, 配置, _ = 目标方法环境
源, 映射 = _参考分析输入(快照, 配置, 身份)
冻结 = 准备目标请求(装配, 身份, 源, 分流知识记录(源, 映射), 映射)
原 = TypeAdapter(参考分析请求).validate_python(冻结["payload"])
for 型 in ("character", "faction"):
请求 = replace(
原,
analysis_id="reference-" + 型,
type_id=型,
schema_id=型,
title="引用目标",
content={"名称": "引用目标"},
)
装配.要求资料().提出参考分析(身份, "create-reference-" + 型, 请求)
前 = _业务与索引指纹(装配)
for 型 in ("character", "faction"):
with pytest.raises(Muse错误, match="引用类型、状态或来源范围不一致"):
请求 = replace(原, content={"名称": "甲", "阵营": "reference-" + 型})
装配.要求资料().提出参考分析(身份, "bad-reference-" + 型, 请求)
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-w21-214004",
environment="隔离PG及合成提供方,不访问真实资料或外部模型",
given="S01回执事务失败或业务提交后台账中断",
when="调用真实领域/CLI/迁移入口",
then=["事务失败无孤儿,业务成功原回执可恢复mapped,不重建结果"],
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
)
@pytest.mark.parametrize("断点", ["回执事务", "迁移终态"])
def test_参考分析中断按原命令恢复且事务失败不留孤儿__214004(目标方法环境, monkeypatch, 断点):
from muse.正式变更.存储 import 变更存储
装配, 身份, 快照, 配置, 参数 = 目标方法环境
源, 映射 = _参考分析输入(快照, 配置, 身份)
def 中断(*args, **kwargs):
raise RuntimeError("合成中断")
with monkeypatch.context() as m:
if 断点 == "回执事务":
m.setattr(变更存储, "保存回执", 中断)
else:
m.setattr(迁移台账, "保存终态", 中断)
with pytest.raises(RuntimeError, match="合成中断"):
导入快照(装配, 身份, [源], 映射, **参数)
with 装配.要求数据库().连接(只读=True) as 连:
初 = 装配.要求资料().列出分析(连, 身份.作者, 映射.值["works"][0]["target_ref"])
assert len(初) == (0 if 断点 == "回执事务" else 1)
assert 导入快照(装配, 身份, [源], 映射, **参数)["states"] == {"mapped": 1}
with 装配.要求数据库().连接(只读=True) as 连:
后 = 装配.要求资料().列出分析(连, 身份.作者, 映射.值["works"][0]["target_ref"])
assert len(后) == 1
if 初:
assert 后 == 初
assert 核对迁移结果(装配, 身份, [源], 映射, **参数)["data_checks_passed"] is True
assert 装配.任务运行.列出任务(身份.作者) == []
@pytest.mark.case_id(
"NC-w21-214005",
environment="隔离PG及合成提供方,不访问真实资料或外部模型",
given="既存分析之后收紧S04显示/导出字段策略",
when="调用真实领域/CLI/迁移入口",
then=["呈现遵当前策略,原内容/哈希不改;拒aiContext伪授消费资格"],
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
)
def test_参考分析按用途投影且不改留存哈希__214005(目标方法环境):
from muse.元数据.接口 import 启用命令, 字段限制, 定义哈希, 策略快照
from muse.编排.研究拆书 import 查询拆书状态
装配, 身份, 快照, 配置, 参数 = 目标方法环境
源, 映射 = _参考分析输入(快照, 配置, 身份)
导入快照(装配, 身份, [源], 映射, **参数)
with 装配.要求数据库().连接(只读=True) as 连:
初 = 装配.要求资料().列出分析(连, 身份.作者, 映射.值["works"][0]["target_ref"])[0]
维护 = 数据库引用("受控存储", str(Path(装配.配置.数据库.位置).with_name("维护连接.txt")))
with 数据库工厂(维护, 用途.维护).连接() as 连, 连.transaction():
元数据 = 元数据服务(连)
版 = 元数据.当前策略("character").version
新 = 策略快照(
"character", 版 + 1, (字段限制("character.身份定位", ("uiVisible", "exportable")),)
)
元数据.更新策略(
新,
启用命令(
"hide-analysis-field",
定义哈希(asdict(新)),
版,
"synthetic-review",
"type:character",
),
)
视图 = 查询拆书状态(装配, 身份.作者, 初["source_id"])
assert 视图["台账"] == [] and 视图["对照"] is None
assert 视图["分析"][0]["payload"]["request"]["content"] == {"名称": "林深"}
with 装配.要求数据库().连接(只读=True) as 连:
assert 装配.要求资料().读取分析(连, 身份.作者, 初["analysis_id"]) == 初
with pytest.raises(Muse错误, match="不授予AI消费"):
装配.要求资料().读取分析(连, 身份.作者, 初["analysis_id"], 字段用途="aiContext")
assert 核对迁移结果(装配, 身份, [源], 映射, **参数)["data_checks_passed"] is True
@pytest.mark.case_id(
"NC-w21-213002",
environment="合成旧源与隔离PG,只读试算",
given="空步骤、超过500步或请求未知键",
when="由真实核对入口处理明确查询请求",
then=["在读取额外来源前拒绝,checked_sources=0,legacy_lineage_invalid,不写任何数据"],
contract="docs/系统架构/新版设计/数据模型/旧知识记录分流.md",
)
@pytest.mark.parametrize("坏形状", ["空步骤", "超限", "未知键"])
def test_回放请求有界且非法形状不读取额外来源__213002(目标方法环境, 坏形状):
装配, 身份, _, _, 参数 = 目标方法环境
输入, 映射, 请求 = _来源回放输入(装配, 身份)
导入快照(装配, 身份, 输入, 映射, **参数)
if 坏形状 == "空步骤":
请求["steps"] = []
elif 坏形状 == "超限":
请求["steps"] = [请求["steps"][0]] * 501
else:
请求["unknown"] = True
前 = _业务与索引指纹(装配)
结果 = 核对迁移结果(装配, 身份, 输入, 映射, 来源回放=请求, **参数)
assert 结果["data_checks_passed"] is False
assert 结果["lineage_replay"]["checked_sources"] == 0
assert 结果["lineage_replay"]["issues"] == [{"reason_code": "legacy_lineage_invalid"}]
assert _业务与索引指纹(装配) == 前
@pytest.mark.case_id(
"NC-mig-36b063",
environment="隔离PostgreSQL外部环境与临时目录",
given="隔离真实PG库与固定PG作品/章/block源",
when="经公开口导入并重放",
then=["全部入账且幂等,零自动确认,作者/参考两用途各自正确"],
contract="docs/系统架构/新版设计/迁移与验收.md",
)
@pytest.mark.parametrize("use", ["author", "reference"], ids=["author", "reference"])
@pytest.mark.parametrize("分片", [False, True], ids=["False", "True"])
def test_PG作品章block真实公开导入与幂等重放_零自动确认__8f17e3(
目标方法环境, monkeypatch, use, 分片
):
"""真实PG用例;合成文本不是数据库替身,需要父代理显式外部环境运行。"""
from test_PG作品正文迁移 import 合成PG组
from muse.正文写作.接口 import 可见文本
from PG作品映射 import PG目标ID
装配, 身份, _, _, 参数 = 目标方法环境
rows, value = 合成PG组(use=use)
mapping = 映射配置(value)
calls = []
正式 = 装配.要求作品().正式
submit = 正式.提交变更
def 记录提交(身份, 命令):
calls.append((命令.入口, 命令.动作.value))
assert 命令.动作.value == "manual_save"
assert 命令.审阅ID is None
return submit(身份, 命令)
monkeypatch.setattr(正式, "提交变更", 记录提交)
if 分片:
first = 导入快照(装配, 身份, list(reversed(rows[:4])), mapping, **参数)
assert first["states"] == ({"mapped": 4} if use == "author" else {"reserved": 4})
partial = 导入快照(装配, 身份, [rows[4]], mapping, **参数)
assert partial["states"] == {"reserved": 1}
导入快照(装配, 身份, rows[5:], mapping, **参数)
result = 导入快照(装配, 身份, list(reversed(rows)), mapping, **参数)
else:
result = 导入快照(装配, 身份, list(reversed(rows)), mapping, **参数)
assert result["states"] == {"mapped": len(rows)}
ledger = 迁移台账(装配.要求数据库(), 身份, 正式, 装配=装配)
records = [ledger.读取(s.source.记录键) for s in rows]
assert all(r["original"] == s.导出() for r, s in zip(records, rows, strict=True))
assert records[1]["original"]["record"]["status"] == "confirmed"
assert (
records[1]["original"]["source"]["database"] != records[0]["original"]["source"]["database"]
)
if use == "author":
assert [c[0] for c in calls] == [
"B01.save_profile",
*("B01.change_directory" for _ in range(3)),
"B05.save_candidate",
"B05.save_candidate",
]
work_id = PG目标ID(rows[0], 身份.作者, "work")
directory = 装配.要求作品().读取目录(身份, work_id)
assert [c["chapter_id"] for c in directory["chapters"]] == [
PG目标ID(c, 身份.作者, "chapter") for c in rows[1:4]
]
assert records[4]["receipt"] == records[5]["receipt"]
assert records[4]["targets"] == records[5]["targets"]
target = records[4]["targets"][0]
candidate = 装配.要求正文().读取候选(身份, target["id"], 版本=1)
assert (
可见文本(TypeAdapter(正文草稿).validate_python(candidate["document"]))
== rows[4].原行["content_text"] + "\n" + rows[5].原行["content_text"]
)
assert len(records[4]["frozen_request"]["payload"]["draft"]["block_map"]) == 2
for chapter in rows[1:4]:
assert (
装配.要求正文().读取正文(身份, PG目标ID(chapter, 身份.作者, "chapter"))["revision"]
== 0
)
# 修改当前档案后仍核原历史版本,而不是拿当前值冒充导入值。
profile = 装配.要求作品().读取表单(身份, work_id)
schema = 装配.要求作品().新档案结构(身份, work_id, "work_core", 1)
装配.要求作品().保存档案(
身份,
"pg-later-author-edit",
档案保存(work_id, profile["revision"], {"名称": "后续编辑"}, schema),
)
assert 装配.要求作品().读取档案版本依据(身份, work_id, 1)["revision"] == 1
else:
assert calls == []
assert all(r["receipt"] == records[0]["receipt"] for r in records)
assert all(
r["frozen_request"]["payload"] == {"member_of": value["pg_works"][0]["work"]}
for r in records[1:]
)
assert 装配.要求作品().列出作品(身份) == []
with 装配.要求数据库().连接(只读=True) as 连:
sources = 装配.要求资料().列出来源(连)
source = next(
s for s in sources if str(s["source_id"]) == records[0]["receipt"]["source_id"]
)
assert source["authorized_uses"] == []
# 参考导入零授权用途,原文受当前用途门禁保护而不回读;
# 存储哈希与不可变回执仍须与冻结请求逐项一致。
version = 装配.要求资料().读取版本回执(连, str(source["source_id"]), 1)
assert version["import_result"] == records[0]["receipt"]["import_result"]
assert (
version["content_hash"]
== hashlib.sha256(
records[0]["frozen_request"]["payload"]["request"]["content"].encode()
).hexdigest()
)
before_calls = len(calls)
assert 导入快照(装配, 身份, rows, mapping, **参数)["states"] == result["states"]
assert len(calls) == before_calls
assert [ledger.读取(s.source.记录键)["receipt"] for s in rows] == [
r["receipt"] for r in records
]
assert 核对迁移结果(装配, 身份, rows, mapping, **参数)["data_checks_passed"]