"""绑定目标的无歧义编解码与旧命令兼容边界。""" import json from dataclasses import replace from types import SimpleNamespace import pytest from muse.正式变更.接口 import 作者动作, 变更命令, 固定哈希 from muse.知识方法.方法版本 import 绑定对象依赖 from muse.知识方法.模型 import 方法错误, 绑定命令 from muse.知识方法.绑定身份 import 编码绑定身份, 解析绑定身份 @pytest.mark.case_id("NC-O09-BINDING-ROUNDTRIP") @pytest.mark.parametrize( "work,target", [ ("simple", "work"), ("synthetic:demo", "work:synthetic:demo"), ("作品:甲:乙", "https://example.invalid/a:b?x=1&y=%3A#z"), ("work:long", "x:" + "界" * 10000), ("work", '["target:with,quotes"]'), ], ) def test_长度中文URL和序列化都原样往返(work, target): key = 编码绑定身份(work, target) assert 解析绑定身份(json.loads(json.dumps(key))) == (work, target) calls = [] conn = SimpleNamespace( execute=lambda sql, params: calls.append(params) or SimpleNamespace(fetchone=lambda: (4,)) ) assert 绑定对象依赖().锁定当前(conn, key) == "4" assert calls == [(work, target)] @pytest.mark.case_id("NC-O09-BINDING-LEGACY") def test_旧简单命令哈希不变而歧义旧目标不猜拆(): 请求 = 绑定命令("work", "work", "target", "method", "version") 旧 = 变更命令( "old-command", "B04.bind_method", "binding:work:target", 作者动作.人工保存, 0, 请求 ) assert 固定哈希(replace(旧, 目标=编码绑定身份("work", "target"))) == 固定哈希(旧) 歧义 = "binding:synthetic:demo:work:synthetic:demo" with pytest.raises(方法错误) as 错误: 解析绑定身份(歧义) assert 错误.value.错误码 == "SOURCE_STALE" assert 编码绑定身份("synthetic:demo", "work:synthetic:demo") != 歧义 @pytest.mark.case_id("NC-O09-BINDING-COLLISION") def test_不同作品与目标组合不能碰撞(): a = 编码绑定身份("w:a", "target") b = 编码绑定身份("w", "a:target") assert a != b assert 解析绑定身份(a) == ("w:a", "target") assert 解析绑定身份(b) == ("w", "a:target") with pytest.raises(方法错误): 解析绑定身份('binding-v2:["w","a:target","extra"]') @pytest.mark.case_id("NC-O09-BINDING-SCOPE") def test_绑定查看先由作品owner拒绝跨作者再读取方法(monkeypatch): import muse.作品规划.接口 as 作品 import muse.知识方法.方法版本 as 方法 from muse.作品规划.模型 import 作品错误 checked = [] def scope(conn, author, work): checked.append((author, work)) raise 作品错误("SCOPE_DENIED", "作品不属于该作者") monkeypatch.setattr(作品, "读取作品范围", scope) monkeypatch.setattr(方法, "方法存储", lambda *a: pytest.fail("跨作用域后不应读方法")) 请求 = 绑定命令("other:work", "work", "https://example.invalid/target", "m", "v") with pytest.raises(作品错误) as 错误: 方法.方法绑定入口().查看( None, SimpleNamespace(作者="actor"), SimpleNamespace(业务请求=请求) ) assert 错误.value.错误码 == "SCOPE_DENIED" assert checked == [("actor", "other:work")] @pytest.mark.case_id("NC-O09-BINDING-OLD-RECEIPT") def test_旧命令回执不被新目标静默重写(): from muse.共享.调用身份 import 用途 from muse.正式变更.存储 import 变更存储 from muse.正式变更.接口 import 变更错误 请求 = 绑定命令("synthetic:demo", "work", "work:synthetic:demo", "m", "v") 旧 = 变更命令( "old", "B04.bind_method", "binding:synthetic:demo:work:synthetic:demo", 作者动作.人工保存, 0, 请求, ) 新 = replace(旧, 目标=编码绑定身份(请求.work_id, 请求.target_ref)) receipt = {"command_id": "old", "target_ref": 旧.目标} class Conn: def cursor(self, **kwargs): return self def execute(self, sql, params): assert not sql.lstrip().startswith("UPDATE ") return self def fetchone(self): return {"request_hash": 固定哈希(旧), "receipt": receipt} store = 变更存储(Conn()) actor = SimpleNamespace(作者="author", 用途=用途.生产) assert store.占用命令(actor, 旧) is receipt with pytest.raises(变更错误) as 错误: store.占用命令(actor, 新) assert 错误.value.错误码 == "COMMAND_PAYLOAD_MISMATCH" assert receipt["target_ref"] == 旧.目标