286 lines
13 KiB
Python
286 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""M11b-2d-A3 契约守卫用例：函数式夹具契约 + 真源返回键齐全性。
|
||
|
||
来源
|
||
====
|
||
本文件由一次性调试探针 ``modules/world_sync/.m11b2d_probe.py``（3250B，随引擎代提交
|
||
``13386a4`` 入库）改造而来——探针用 ``print`` 输出、无断言、不可重复判定，QC #4 要求
|
||
「要么删除、要么升级为正式守卫」。本文件即升级结果，探针本体已 ``git rm``。
|
||
|
||
守卫什么（三件事）
|
||
==================
|
||
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(...))`` 的同步入口）。分支覆盖
|
||
成功写 / 幂等命中 / 乐观锁冲突 / 参数门禁 / 快照 / 提交后广播。
|
||
|
||
边界（遵守 QC #3 允许路径与本任务禁改清单）
|
||
==========================================
|
||
* 不改业务实现、不改 ``pbl_runtime_ext/models/*.json``、不改 ``load_path.py``；
|
||
* 不改 ``tests/_m11b2_compat.py`` 任何既有符号（只读引用）；
|
||
* 不改 ``test_m11b2_single_tx.py`` / ``test_m11b2_wiring.py``（并行任务独占）；
|
||
* 权威契约定义在 ``tests/_m11b2_compat.py`` 顶部 docstring 与
|
||
``docs/work-log/``… 不，登记文档为 ``skill/SKILL.md`` 指向的
|
||
``docs/work-log/m11b2d-fixture-contract.md``，本文件只做机械断言，不重复搬运说明。
|
||
|
||
夹具的 ``_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 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
|
||
|
||
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
|
||
|
||
# 真源 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/broadcast_ms/sla_breach/seq_no
|
||
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
|
||
|
||
|
||
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):
|
||
"""真源逐参数签名钉死：形参名/顺序/默认值（文档 m11b2d-fixture-contract.md 同源）。"""
|
||
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": 7, "world_id": 1, "session_id": 11,
|
||
"event_type": "unit.move", "payload": {"dx": 1}}
|
||
base.update(kw)
|
||
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)
|
||
self.assertEqual(r["channel"], "pbl:rtx:w1:t7")
|
||
self.assertIsInstance(r["stmt_count"], int)
|
||
self.assertGreaterEqual(r["stmt_count"], 2) # 事件 INSERT + 状态 UPSERT
|
||
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):
|
||
r1 = self._apply(state_updates=[su()], client_event_id="ev-1")
|
||
r2 = self._apply(state_updates=[su(state={"hp": 9})], client_event_id="ev-1")
|
||
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)
|
||
self.assertIsNone(r2["latency_ms"])
|
||
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=7, session_id=11)
|
||
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"], "pbl:rtx:w1:t7")
|
||
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": 7, "world_id": 1, "session_id": 11,
|
||
"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=7, world_id=1, session_id=11, event_type="unit.move",
|
||
payload={}, state_updates=[su()], client_event_id="ev-1")
|
||
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)
|
||
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)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|