muse-agent-example/tests/迁移/test_作品资产迁移.py

"""旧声音和节奏通过真实owner写未决候选，核对原回执与来源绑定。"""

from dataclasses import asdict

import pytest
import test_隔离目标台账 as 既有

from muse.审校修订.接口 import 声音样张引用
from 导入新库 import 导入快照
from 建立映射 import 旧快照, 映射配置
from 映射台账 import 迁移台账
from 核对结果 import 核对迁移结果

pytestmark = pytest.mark.数据库
迁移目标 = 既有.迁移目标
目标方法环境 = 既有.目标方法环境


def 准备资产(环境, type_id):
    app, author, _, _, params = 环境
    original, mapping = 既有._本书输入(app, author)
    value = original.导出()
    value.pop("source_hash")
    value["record"]["draft_payload"] = {
        "type": type_id,
        "name": "旧作品画像",
        "fields": {"概述": "短句推动开局"},
    }
    source = 旧快照.从载荷(value)
    owner, field = ("B06", "句式画像") if type_id == "style" else ("B01", "黄金三章画像")
    mapping["schemas"] = {
        type_id: {
            "type": "object",
            "properties": {"概述": {"type": "string"}},
            "additionalProperties": False,
        }
    }
    mapping["target_schemas"] = {
        owner + ":" + type_id: {
            "schema_id": type_id,
            "base_version": 1,
            "fields": {field: "fields.概述"},
        }
    }
    document = app.要求正文().读取正文(author, "migration-chapter")
    sample = 声音样张引用("document", "migration-chapter", 1, document["document_hash"], "main")
    mapping["asset_bindings"] = [
        {
            "source_key": source.source.记录键,
            "source_hash": source.源哈希,
            "sources": [asdict(sample)],
            "parameters": {
                "expected_voice_revision" if type_id == "style" else "expected_revision": 0,
            },
        }
    ]
    return app, author, source, mapping, params


@pytest.mark.parametrize("type_id", ["style", "pacing"])
def test_旧作品画像进入正确域未决候选且原字段来源可回查__29e001(目标方法环境, type_id):
    app, author, source, mapping, params = 准备资产(目标方法环境, type_id)
    result = 导入快照(app, author, [source], 映射配置(mapping), **params)
    assert result["states"] == {"mapped": 1}, result
    ledger = 迁移台账(app.要求数据库(), author, app.要求知识方法().正式)
    row = ledger.读取(source.source.记录键)
    assert row["original"] == source.导出()
    target = row["targets"][0]
    if type_id == "style":
        proposal = app.要求审校().读取声音提案(author, target["id"])
        assert proposal["request"]["narrator"] == {"句式画像": "短句推动开局"}
        assert proposal["request"]["samples"] == mapping["asset_bindings"][0]["sources"]
        assert app.要求审校().当前声音(author, "migration-work") is None
        assert target["owner"] == "B06" and target["kind"] == "voice_proposal"
    else:
        proposal = app.要求规划().读取候选(author, target["id"])
        assert proposal["content"] == {"黄金三章画像": "短句推动开局"}
        assert proposal["decision"] == "undecided"
        assert app.要求规划().读取表单(author, "migration-work", "pacing", 1)["revision"] == 0
        assert target["owner"] == "B01" and target["kind"] == "plan_candidate"
    assert 导入快照(app, author, [source], 映射配置(mapping), **params) == result
    assert 核对迁移结果(app, author, [source], 映射配置(mapping), **params)["data_checks_passed"]


@pytest.mark.parametrize("bad", ["missing", "hash", "different_source", "unknown_parameter"])
def test_资产缺绑定或冒用来源时留原隔离不造批准__29e002(目标方法环境, bad):
    app, author, source, mapping, params = 准备资产(目标方法环境, "style")
    binding = mapping["asset_bindings"][0]
    if bad == "missing":
        mapping["asset_bindings"] = []
    elif bad == "hash":
        binding["sources"][0]["source_hash"] = "0" * 64
    elif bad == "different_source":
        mapping["sources"][0]["target_ref"] = "another-source"
    else:
        binding["parameters"]["auto_confirm"] = True
    result = 导入快照(app, author, [source], 映射配置(mapping), **params)
    assert result["states"] == {"quarantined": 1}, result
    row = 迁移台账(app.要求数据库(), author, app.要求知识方法().正式).读取(source.source.记录键)
    assert row["original"] == source.导出() and not row["targets"]
    assert app.要求审校().当前声音(author, "migration-work") is None


@pytest.mark.parametrize("type_id", ["style", "pacing"])
def test_资产owner已提交后台账中断只补原回执__29e003(目标方法环境, monkeypatch, type_id):
    app, author, source, mapping, params = 准备资产(目标方法环境, type_id)
    original = 迁移台账.保存终态

    def interrupt(self, key, state, **values):
        if state == "mapped":
            raise RuntimeError("synthetic stop after owner commit")
        return original(self, key, state, **values)

    monkeypatch.setattr(迁移台账, "保存终态", interrupt)
    with pytest.raises(RuntimeError, match="after owner commit"):
        导入快照(app, author, [source], 映射配置(mapping), **params)
    ledger = 迁移台账(app.要求数据库(), author, app.要求知识方法().正式)
    before = ledger.读取(source.source.记录键)
    receipt = app.要求知识方法().正式.读取回执(author, before["command_id"])
    monkeypatch.setattr(迁移台账, "保存终态", original)
    assert 导入快照(app, author, [source], 映射配置(mapping), **params)["states"] == {"mapped": 1}
    assert ledger.读取(source.source.记录键)["receipt"] == receipt
    # 实际S01动作仅为人工候选保存；没有迁移出来的作者批准。
    assert receipt["action"] == "manual_save"


def test_旧声音确认声明不冒充新版批准__29e004(目标方法环境):
    app, author, source, mapping, params = 准备资产(目标方法环境, "style")
    value = source.导出()
    value.pop("source_hash")
    value["record"]["status"] = "confirmed"
    source = 旧快照.从载荷(value)
    result = 导入快照(app, author, [source], 映射配置(mapping), **params)
    assert result["states"] == {"historical": 1}
    row = 迁移台账(app.要求数据库(), author, app.要求知识方法().正式).读取(source.source.记录键)
    assert row["receipt"] is None and row["targets"] == []
    assert row["reason_code"] == "missing_confirmation"
    assert app.要求审校().当前声音(author, "migration-work") is None