# -*- coding: utf-8 -*- """M11b-2d-A3 契约守卫用例：函数式夹具契约 + 真源返回键齐全性。 来源（响应 QC #4「探针要么删要么升级」） ========================================== 本文件由一次性调试探针 ``modules/world_sync/.m11b2d_probe.py``（3250B，随引擎代提交 ``13386a4`` 入库）改造而来——探针用 ``print`` 输出、无断言、不可重复判定。探针本体已 由 ``git rm`` 从工作区与 git 索引移除，删除提交为 ``81fdb86``（chore(M11b-2d-A3): 移除一次性调试探针），本文件即升级结果。 守卫什么（三件事） ================== 1. **旧类式符号不再存在**：``RuntimeStateEventWriter`` / ``configure_writer`` / ``build_plan`` / 类式 ``write_event_with_state`` 不得再出现在夹具层 ``tests/_m11b2_compat`` 上（它们曾构成第二条夹具路径，把测试钉在已删除的调用形态上， 实测 22 failed）。 2. **新函数式符号可用**：夹具只保留一条路径——``install/uninstall/apply/read/ trace_kinds/update_statements/insert_statements`` + ``RuntimeEngine``，且真源本体 ``apply_runtime_event`` / ``read_states`` 经夹具直连（不是副本）。 3. **真源返回键齐全**：``apply`` 的返回值键集合**逐键断言**，不做任何增补（夹具 ``apply`` 只是 ``asyncio.run(apply_runtime_event(...))`` 的同步入口）。分支覆盖 成功写 / 幂等命中 / 乐观锁冲突 / 参数门禁 / 快照 / 提交后广播。 权威契约定义（本文档不搬运说明） ================================ * 夹具层契约：``tests/_m11b2_compat.py`` 顶部 docstring（唯一权威实现说明）； * 登记文档（索引式，含真源逐参数签名与返回键语义）： ``docs/work-log-2026-09-22-m11b2d-fixture-contract.md``； * 真源本体：``modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py`` （``apply_runtime_event`` / ``read_states``）。 本文件只做机械断言，断言值一律**以真源实现为准**（如广播通道名取 ``broadcast.channel_of(world_id, tenant_id)`` 的真实产物，不写理想化字面量）。 边界（遵守 QC #3 允许路径与本任务禁改清单） ========================================== * 不改业务实现、不改 ``pbl_runtime_ext/models/*.json``、不改 ``load_path.py``； * 不改 ``tests/_m11b2_compat.py`` 任何既有符号（只读引用）； * 不改 ``test_m11b2_single_tx.py`` / ``test_m11b2_wiring.py``（并行任务独占）。 created_at 说明（与 ``test_m11b2_single_tx.py`` 同一口径） ========================================================= 真源事件表两条唯一索引（``uk_event_uid`` = tenant_id+event_uid+created_at、 ``uk_tenant_idem`` = tenant_id+idem_key+created_at）都含分区键 created_at，而真源不填 idem_key（DEFAULT ''），故**同秒两条事件会撞 DB 唯一索引**。本文件用 :func:`stamp` 给每条事件分配互不相同的 ``occurred_at``（真源把它同时写进 created_at/updated_at/ occurred_at），与并行文件的做法一致，属测试侧用法修正，不改真源与夹具语义。 夹具的 ``_default_clause`` 把 models 里 ``default: ""``（空串）当「无默认值」跳过， 于是 ``event_code`` 等「NOT NULL DEFAULT ''」列在 sqlite 侧成为无默认值的 NOT NULL 列， 真源 INSERT（按语义不填这些列）会撞约束。本文件与 ``test_m11b2_single_tx.py`` 同口径： **仅在构造期**临时补正空串默认值渲染，建完表立即还原，不改 compat 既有定义与行为语义。 """ import inspect import os import sys import threading import time import unittest HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) sys.path.insert(0, os.path.dirname(HERE)) import _m11b2_compat as compat # noqa: E402 （导入时已把 modules/pbl_runtime_ext 加入 sys.path） # 真源本体（被测对象，不是副本） apply_runtime_event = compat.apply_runtime_event # pbl_runtime_ext.tx_write 真源本体 read_states = compat.read_states RtxError = compat.RtxError EVENT_TABLE = compat.EVENT_TABLE STATE_TABLE = compat.STATE_TABLE SNAPSHOT_TABLE = compat.SNAPSHOT_TABLE ERR_TENANT_MISSING = compat.ERR_TENANT_MISSING ERR_BAD_PARAM = compat.ERR_BAD_PARAM ERR_STATE_CONFLICT = compat.ERR_STATE_CONFLICT # 广播通道名的真源实现（pbl_runtime_ext/broadcast.py:22 channel_of，前缀 CHANNEL_PREFIX="pbl.world"） from pbl_runtime_ext import broadcast as rtx_broadcast # noqa: E402 TENANT = 7 WORLD = 1 SESSION = 11 # 真源 tx_write.apply_runtime_event 成功分支的返回键（源码 tx_write.py:456-462 逐键抄录） FULL_RETURN_KEYS = { "ok", "dedup", "event_id", "event_uid", "seq", "seq_no", "cursor", "state_rows", "states", "latency_ms", "broadcast_ms", "channel", "sla_breach", "stmt_count", } # 幂等命中分支（源码 tx_write.py:383-389）：无 states/seq_no/broadcast_ms/sla_breach DEDUP_RETURN_KEYS = { "ok", "dedup", "event_id", "event_uid", "seq", "cursor", "state_rows", "latency_ms", "channel", "stmt_count", } # 真源 apply_runtime_event 形参（源码 tx_write.py:344-348），全部默认 None TRUE_SOURCE_PARAMS = [ "tenant_id", "world_id", "session_id", "event_type", "payload", "state_updates", "client_event_id", "idem_key", "snapshot", "base_version", "actor_id", "causation_id", "occurred_at", "env", ] # 夹具层禁止复活的旧类式符号 DEAD_CLASS_STYLE_SYMBOLS = [ "RuntimeStateEventWriter", "configure_writer", "build_plan", "write_event_with_state", ] # 夹具层必须存在的函数式入口 LIVE_FIXTURE_SYMBOLS = [ "RuntimeEngine", "install", "uninstall", "apply", "read", "trace_kinds", "update_statements", "insert_statements", "apply_runtime_event", "read_states", "build_row", "assert_append_only", "RtxError", "percentile", "EVENT_TABLE", "STATE_TABLE", "SNAPSHOT_TABLE", ] _ORIGIN_DEFAULT_CLAUSE = compat._default_clause _BASE = time.mktime(time.strptime("2026-06-01 00:00:00", "%Y-%m-%d %H:%M:%S")) _stamp_lock = threading.Lock() _stamp_n = [0] def stamp(): """互不相同的 occurred_at（=真源写入的 created_at），避开含分区列的唯一索引互撞。""" with _stamp_lock: _stamp_n[0] += 1 n = _stamp_n[0] return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(_BASE + n)) def expected_channel(world_id=WORLD, tenant_id=TENANT): """通道名以真源实现为准：pbl.world..（租户隔离在通道层生效）。""" return rtx_broadcast.channel_of(world_id, tenant_id) def _default_clause_with_empty(fd): dv = fd.get("default") if isinstance(dv, str) and dv.strip() == "": return " DEFAULT ''" return _ORIGIN_DEFAULT_CLAUSE(fd) def make_engine(path=":memory:"): """真 sqlite 事务引擎（表结构仍由夹具从 pbl_runtime_ext/models/*.json 生成）。""" compat._default_clause = _default_clause_with_empty try: return compat.RuntimeEngine(path=path) finally: compat._default_clause = _ORIGIN_DEFAULT_CLAUSE def su(entity="e1", state=None): return {"entity_id": entity, "state": {"hp": 2} if state is None else state} class FixtureSurfaceCase(unittest.TestCase): """夹具面：旧符号已死、新符号可直连真源。""" def test_dead_class_style_symbols_absent(self): for name in DEAD_CLASS_STYLE_SYMBOLS: self.assertFalse( hasattr(compat, name), "夹具层不得复活旧类式符号 %s（第二条夹具路径 = M11b-2c 22 failed 根因）" % name) def test_live_function_style_symbols_present(self): for name in LIVE_FIXTURE_SYMBOLS: self.assertTrue(hasattr(compat, name), "夹具缺少函数式符号 %s" % name) self.assertTrue(callable(compat.install)) self.assertTrue(callable(compat.uninstall)) self.assertTrue(callable(compat.apply)) self.assertTrue(callable(compat.read)) def test_fixture_apply_is_thin_wrapper_on_true_source(self): """夹具 apply/read 不得自带实现——必须是真源的同步入口（零返回键增补）。""" self.assertIs(compat.apply_runtime_event, apply_runtime_event) src = inspect.getsource(compat.apply) self.assertIn("apply_runtime_event", src) self.assertNotIn("update(", src) # 不补键 self.assertNotIn("setdefault", src) self.assertNotIn("_augment", src) def test_true_source_signature(self): """真源逐参数签名钉死：形参名/顺序/默认值（与登记文档同源）。""" spec = inspect.signature(apply_runtime_event) self.assertEqual(list(spec.parameters), TRUE_SOURCE_PARAMS) for name in TRUE_SOURCE_PARAMS: self.assertIsNone(spec.parameters[name].default, "形参 %s 默认值应为 None" % name) def test_read_states_signature(self): spec = inspect.signature(read_states) self.assertEqual(list(spec.parameters), ["tenant_id", "session_id", "entity_ids", "world_id", "env"]) class ReturnKeysCase(unittest.TestCase): """真源返回键齐全性（走夹具函数式路径，断言真值，不接受 print）。""" def setUp(self): self.eng = make_engine() compat.install(self.eng) self.addCleanup(compat.uninstall) def _apply(self, **kw): base = {"tenant_id": TENANT, "world_id": WORLD, "session_id": SESSION, "event_type": "unit.move", "payload": {"dx": 1}} base.update(kw) if "occurred_at" not in kw: base["occurred_at"] = stamp() return compat.apply(**base) def test_success_return_keys_exact(self): r = self._apply(state_updates=[su()], client_event_id="ev-1", actor_id="u1") self.assertEqual(set(r.keys()), FULL_RETURN_KEYS) self.assertIs(r["ok"], True) self.assertIs(r["dedup"], False) self.assertIsNotNone(r["event_id"]) self.assertEqual(r["seq"], 1) self.assertEqual(r["seq_no"], r["seq"]) # 对外别名一致 self.assertEqual(r["state_rows"], 1) # 通道名以真源实现为准：pbl.world.. self.assertEqual(r["channel"], expected_channel()) self.assertEqual(r["channel"], "pbl.world.7.1") self.assertIsInstance(r["stmt_count"], int) self.assertGreaterEqual(r["stmt_count"], 2) # 事件 INSERT + 状态 UPSERT self.assertIsInstance(r["latency_ms"], (int, float)) self.assertGreaterEqual(r["latency_ms"], 0) self.assertIsInstance(r["states"], list) self.assertEqual(set(r["states"][0].keys()), {"entity_id", "state_version", "action", "state_key", "state_value", "seq"}) self.assertEqual(r["states"][0]["action"], "insert") self.assertEqual(r["states"][0]["state_version"], 1) def test_dedup_return_keys_exact_and_no_second_write(self): ts = stamp() r1 = self._apply(state_updates=[su()], client_event_id="ev-1", occurred_at=ts) # 幂等命中：同 client_event_id → 同 event_uid，走「已提交事务 + 不重复写」分支 r2 = self._apply(state_updates=[su(state={"hp": 9})], client_event_id="ev-1", occurred_at=ts) self.assertEqual(set(r2.keys()), DEDUP_RETURN_KEYS) self.assertIs(r2["dedup"], True) self.assertEqual(r2["event_id"], r1["event_id"]) # 复用既有事件，不新写 self.assertEqual(r2["seq"], r1["seq"]) self.assertEqual(r2["state_rows"], 0) # 真源语义：幂等命中**仍然计时**（tx_write.py:442 在返回前回填 latency_ms=ms）， # 故断言「是数值且 >= 0」，不断言 None（旧断言与真源实现不符，QC #4）。 self.assertIsInstance(r2["latency_ms"], (int, float)) self.assertGreaterEqual(r2["latency_ms"], 0) self.assertEqual(r2["channel"], expected_channel()) self.assertEqual(self.eng.count(EVENT_TABLE), 1) self.assertEqual(self.eng.count(STATE_TABLE), 1) self.assertEqual(len(self.eng.broadcasts), 1) # 幂等命中不重复广播 def test_read_states_rows_shape(self): self._apply(state_updates=[su()], client_event_id="ev-1") rows = compat.read(tenant_id=TENANT, session_id=SESSION) self.assertIsInstance(rows, list) self.assertEqual(len(rows), 1) row = rows[0] self.assertEqual(row["entity_id"], "e1") self.assertEqual(row["state"], {"hp": 2}) self.assertEqual(row["state_version"], 1) self.assertIn("checksum", row) self.assertIn("updated_at", row) def test_snapshot_writes_snapshot_table(self): r = self._apply(state_updates=[su(state={"hp": 5})], client_event_id="ev-4", snapshot={"reason": "join", "hp": 5}) self.assertEqual(set(r.keys()), FULL_RETURN_KEYS) self.assertEqual(self.eng.count(SNAPSHOT_TABLE), 1) def test_broadcast_strictly_after_commit(self): self._apply(state_updates=[su()], client_event_id="ev-1") kinds = compat.trace_kinds(self.eng) self.assertEqual(kinds[0], "BEGIN") self.assertIn("COMMIT", kinds) self.assertIn("BROADCAST", kinds) self.assertGreater(kinds.index("BROADCAST"), kinds.index("COMMIT"), "广播必须在事务提交之后（M11b-3 契约）") self.assertEqual(self.eng.broadcasts[0]["channel"], expected_channel()) self.assertEqual(self.eng.broadcasts[0]["event"]["event_type"], "unit.move") def test_event_table_is_append_only(self): """事件表只允许 INSERT：UPDATE 语句里不得出现事件表名。""" self._apply(state_updates=[su()], client_event_id="ev-1") self._apply(state_updates=[su(state={"hp": 3})], client_event_id="ev-2") for sql, _p in compat.update_statements(self.eng): self.assertNotIn(EVENT_TABLE.upper(), sql.upper(), "事件表出现 UPDATE 语句，违反 append-only：%s" % sql) inserts = compat.insert_statements(self.eng, EVENT_TABLE) self.assertEqual(len(inserts), 2) for sql, _p in inserts: self.assertIn(EVENT_TABLE.upper(), sql.upper()) class GateErrorCase(unittest.TestCase): """参数门禁与乐观锁冲突：必须抛 RtxError 且 code 精确、事务回滚不留脏行。""" def setUp(self): self.eng = make_engine() compat.install(self.eng) self.addCleanup(compat.uninstall) def _expect(self, code, **kw): base = {"tenant_id": TENANT, "world_id": WORLD, "session_id": SESSION, "event_type": "x", "payload": {}} base.update(kw) with self.assertRaises(RtxError) as ctx: compat.apply(**base) self.assertEqual(ctx.exception.code, code) return ctx.exception def test_missing_tenant_is_fail_closed(self): self._expect(ERR_TENANT_MISSING, tenant_id=None) self.assertEqual(self.eng.count(EVENT_TABLE), 0) def test_missing_required_params(self): for key in ("world_id", "session_id", "event_type"): self._expect(ERR_BAD_PARAM, **{key: None}) self.assertEqual(self.eng.count(EVENT_TABLE), 0) def test_bad_state_updates(self): self._expect(ERR_BAD_PARAM, state_updates="notalist") self._expect(ERR_BAD_PARAM, state_updates=[{"state": {"a": 1}}]) # 缺 entity_id self._expect(ERR_BAD_PARAM, state_updates=[{"entity_id": "e1"}]) # 缺 state self.assertEqual(self.eng.count(STATE_TABLE), 0) def test_state_conflict_rolls_back(self): compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION, event_type="unit.move", payload={}, state_updates=[su()], client_event_id="ev-1", occurred_at=stamp()) before_evt = self.eng.count(EVENT_TABLE) before_state = self.eng.count(STATE_TABLE) self._expect(ERR_STATE_CONFLICT, event_type="x", state_updates=[su(state={"hp": 0})], client_event_id="ev-3", base_version=999, occurred_at=stamp()) kinds = compat.trace_kinds(self.eng) self.assertEqual(kinds[-1], "ROLLBACK", "冲突分支必须回滚，不得 COMMIT") self.assertEqual(self.eng.count(EVENT_TABLE), before_evt) self.assertEqual(self.eng.count(STATE_TABLE), before_state) class RegistrationDocCase(unittest.TestCase): """登记文档守卫（响应 QC #2：交付件正文引用的文档必须真实存在，不留悬空引用）。 文档只做索引式登记，权威实现仍在 ``tests/_m11b2_compat.py`` docstring 与真源源码； 本用例钉的是「登记与真源同源」：形参名、返回键、错误码、废弃符号四类清单 必须在文档正文中逐条出现——任一处漏登记即红。 """ DOC = os.path.join(os.path.dirname(HERE), "docs", "work-log-2026-09-22-m11b2d-fixture-contract.md") def _text(self): self.assertTrue(os.path.isfile(self.DOC), "登记文档缺失（交付件引用悬空）：%s" % self.DOC) with open(self.DOC, encoding="utf-8") as f: return f.read() def test_doc_exists_and_registers_contract(self): text = self._text() for p in TRUE_SOURCE_PARAMS: self.assertIn("`" + p + "`", text, "登记文档缺少真源形参 " + p) for key in sorted(FULL_RETURN_KEYS): self.assertIn("`" + key + "`", text, "登记文档缺少返回键 " + key) for code in ("PBL-TENANT-0001", "PBL-PARAM-0001", "PBL-STATE-CONFLICT", "PBL-RTX-0001-APPEND", "PBL-BROAD-0001", "PBL-RTX-0002", "PBL-RTX-0003", "PBL-RTX-0004", "PBL-RTX-0006"): self.assertIn(code, text, "登记文档缺少错误码 " + code + " 的含义说明") for dead in DEAD_CLASS_STYLE_SYMBOLS: self.assertIn(dead, text, "登记文档缺少废弃符号 " + dead + " 与替代说明") def test_doc_live_fixture_symbols_registered(self): """夹具层公开符号（现行唯一路径）必须逐个登记在文档表格里。""" text = self._text() for name in ("RuntimeEngine", "install", "uninstall", "apply", "read", "trace_kinds", "update_statements", "insert_statements"): self.assertIn("`" + name + "`", text, "登记文档缺少夹具符号 " + name + "（§1.1 清单）") if __name__ == "__main__": unittest.main(verbosity=2)