muse-agent-example/tests/单元/test_无变化保存.py

"""无变化保存仍经过owner校验，不制造业务历史。"""

from dataclasses import asdict, dataclass
from types import SimpleNamespace

import pytest

from muse.作品规划 import 作品档案 as 模块
from muse.作品规划.模型 import 作品错误


@dataclass
class 绑定:
    revision: int = 1


@pytest.fixture
def 档案环境(monkeypatch):
    状态 = {
        "content": {"名称": "A"},
        "current_revision": 1,
        "author_id": "author",
        "schema_binding": asdict(绑定()),
    }
    事件 = []

    class 存储:
        def __init__(self, 连):
            pass

        def 读取(self, work_id):
            return dict(状态)

        def 写档案(self, work_id, author, expected, value, binding, hash_):
            事件.append("write")
            状态.update(content=value, current_revision=expected + 1)
            return expected + 1

    class 元数据:
        def __init__(self, 连):
            pass

        def 读取绑定(self, target):
            return 绑定()

        def 保护提交(self, *args, **kwargs):
            事件.append("protect")
            if 状态.get("revoked"):
                raise 作品错误("PROJECTION_STALE", "用途已撤回")
            return None, None

    def 校验(_结构, _投影, old, changes):
        事件.append("validate")
        value = {**old, **{c.路径[0]: c.值 for c in changes}}
        return SimpleNamespace(引用需求=(), 值=value)

    monkeypatch.setattr(模块, "作品存储", 存储)
    monkeypatch.setattr(模块, "元数据服务", 元数据)
    monkeypatch.setattr(模块, "读取档案结构", lambda *a: (None, None))
    monkeypatch.setattr(模块, "校验字段变更", 校验)

    def 保存(content, revision=1):
        请求 = SimpleNamespace(
            work_id="w",
            target_ref="work:w",
            content=content,
            changes=(),
            expected_revision=revision,
            schema=SimpleNamespace(effective_schema_hash="schema", projection_version="projection"),
        )
        return 模块.档案参与者().提交(None, 模块.档案写入("author", 请求))

    return 状态, 事件, 保存


@pytest.mark.case_id("NC-O09-NOCHANGE")
def test_相同和部分同值保存不写业务版本(档案环境):
    状态, 事件, 保存 = 档案环境
    for content in ({"名称": "A"}, {}):
        结果 = 保存(content)
        assert 结果["no_change"] and not 结果["did_change"] and 结果["revision"] == 1
    assert 事件 == ["protect", "validate", "protect", "validate"]
    assert 状态["current_revision"] == 1


@pytest.mark.case_id("NC-O09-ABA")
def test_真实变化回到旧值仍保留新版本(档案环境):
    _, 事件, 保存 = 档案环境
    assert 保存({"名称": "B"})["revision"] == 2
    assert 保存({"名称": "A"}, 2)["revision"] == 3
    assert 事件.count("write") == 2


@pytest.mark.case_id("NC-O09-STALE")
def test_相同值不绕过版本和撤权(档案环境):
    状态, 事件, 保存 = 档案环境
    with pytest.raises(作品错误, match="其他保存"):
        保存({"名称": "A"}, 0)
    状态["revoked"] = True
    with pytest.raises(作品错误, match="撤回"):
        保存({"名称": "A"})
    assert "write" not in 事件


@pytest.mark.case_id("NC-O09-IMPACTS")
def test_参与者无变化只消除自己的影响(monkeypatch):
    from contextlib import nullcontext

    import muse.正式变更.接口 as 正式模块
    from muse.共享.标识 import 任务标识
    from muse.共享.调用身份 import 内容用途, 用途, 调用身份
    from muse.正式变更.接口 import (
        作者动作,
        依赖引用,
        修改影响,
        参与操作,
        参与者目录,
        变更命令,
        审阅内容,
        提交计划,
        正式变更服务,
    )

    @dataclass(frozen=True)
    class 请求:
        value: str = "same"

    class 入口:
        身份 = "synthetic"
        请求类型 = 请求

        def 查看(self, *a):
            return 审阅内容(
                "target",
                1,
                None,
                None,
                None,
                "hash",
                None,
                None,
                "source",
                ("save",),
                (依赖引用("target", "target", "1"),),
            )

        def 准备(self, *a):
            return 提交计划(
                (参与操作("unchanged", 请求()), 参与操作("changed", 请求())),
                (
                    修改影响("a", "save", "1", "2", "unchanged"),
                    修改影响("b", "save", "1", "2", "changed"),
                ),
            )

    class 参与者:
        请求类型 = 请求

        def __init__(self, id_, change):
            self.身份, self.change = id_, change

        def 提交(self, *a):
            return {"did_change": self.change, "no_change": not self.change}

    class 存储:
        def __init__(self, 连):
            pass

        def 占用命令(self, *a):
            return None

        def 保存影响(self, receipt, impact):
            return impact.对象

        def 保存回执(self, identity, command, receipt):
            return receipt

    monkeypatch.setattr(正式模块, "变更存储", 存储)
    monkeypatch.setattr(正式模块, "锁定依赖", lambda *a: None)
    连 = SimpleNamespace(transaction=lambda: nullcontext())
    库 = SimpleNamespace(用途=用途.生产, 连接=lambda: nullcontext(连))
    目录 = 参与者目录()
    目录.登记入口(入口())
    目录.登记参与者(参与者("unchanged", False))
    目录.登记参与者(参与者("changed", True))
    身份 = 调用身份("author", 任务标识("unit"), 用途.生产, 内容用途.规划)
    回执 = 正式变更服务(库, 目录).提交变更(
        身份, 变更命令("command", "synthetic", "target", 作者动作.人工保存, 1, 请求())
    )
    assert 回执["impact_ids"] == ["b"]
    assert 回执["did_change"] and not 回执["no_change"]
    assert len(回执["results"]) == 2


@pytest.mark.case_id("NC-O09-DECISIONS")
def test_批量决定单次查询限定作者与成对版本(monkeypatch):
    import muse.正式变更.接口 as 模块

    calls = []

    class 存储:
        def __init__(self, conn):
            pass

        def 查询(self, sql, params):
            calls.append((sql, params))
            return SimpleNamespace(
                fetchall=lambda: [
                    {"candidate_id": "a", "candidate_revision": 2, "decision": "adopt"}
                ]
            )

    monkeypatch.setattr(模块, "变更存储", 存储)
    assert 模块.读取作者决定集(None, "author", []) == {}
    assert calls == []
    assert 模块.读取作者决定集(None, "author", [("a", 2), ("b", 3), ("a", 2)]) == {
        ("a", 2): "adopt",
        ("b", 3): "undecided",
    }
    assert len(calls) == 1
    assert "author_id=%s" in calls[0][0] and "unnest" in calls[0][0]
    assert calls[0][1] == ("author", ["a", "b"], [2, 3])