850 lines
44 KiB
Python
850 lines
44 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""M11b-2d 单元测试：单事务写入（事件 + 实体状态）原子性 / 幂等 / 并发 / 时延。
|
||
|
||
运行方式（纯标准库 + sqlite3，无需 appPublic/sqlor/ahserver）::
|
||
|
||
cd modules/world_sync && ./.venv/bin/python -m pytest tests/test_m11b2_single_tx.py -v
|
||
|
||
被测真源（M11b-2a「世界同步真源迁移」起，唯一现行写入路径）：
|
||
modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py::apply_runtime_event
|
||
modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py::read_states
|
||
modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py::build_row / assert_append_only
|
||
夹具层：``tests/_m11b2_compat.py``（函数式真源夹具：真 sqlite + 表结构从
|
||
``pbl_runtime_ext/models/*.json`` 生成 + install/uninstall 打补丁到 rtx_db）。
|
||
|
||
本次迁移（M11b-2d-A1，响应 QC #1「整文件 collection 期 AttributeError，0 条用例可跑」）
|
||
==============================================================================
|
||
旧文件把断言钉在**已被 M11b-2a 删除的类式符号**上（``compat.RuntimeStateEventWriter``
|
||
/ ``compat.configure_writer`` / ``compat.build_plan`` / ``compat.write_event_with_state``），
|
||
并自带一套手写 DDL（``DDL_EVENT`` / ``DDL_STATE``）——手写列名 ``event_id`` / ``state``
|
||
在真源 models 中根本不存在，正是失败根因。迁移要点：
|
||
|
||
1. 表结构一律由夹具从 ``pbl_runtime_ext/models/*.json`` 生成（``compat.ddl_statements``），
|
||
本文件不再自带第二份 schema；预置走 ``eng.seed_state``，查库走 ``eng.query`` /
|
||
``eng.count`` / ``eng.scalar``。
|
||
2. 写入统一走 ``compat.apply(...)``，参数名逐一对齐
|
||
``tx_write.apply_runtime_event(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)``（实读为准，未臆造任何参数）。
|
||
3. 断言键改为真源实际返回键（``ok`` / ``dedup`` / ``event_id`` / ``event_uid`` /
|
||
``seq`` / ``state_rows`` / ``states[]`` / ``stmt_count`` / ``latency_ms`` /
|
||
``channel`` / ``cursor`` / ``broadcast_ms`` / ``sla_breach``），不再沿用真源从未
|
||
返回过的 ``state_inserted`` / ``expected_version`` / ``statement_count`` / ``elapsed_ms``。
|
||
4. 失败注入从 ``SingleConnProxy`` / ``before_commit`` 钩子迁到**真源错误码路径**
|
||
（幂等重复 dedup、``base_version`` 过期冲突、不可变列进 SET、未知列、缺 NOT NULL、
|
||
append-only 违反）；原子回滚用「``eng.trace`` 有 ROLLBACK 无 COMMIT + 两表零增量」证明。
|
||
5. 断言语义不回退：单事务原子性、乐观锁冲突回滚、event_uid 幂等、不可变列守卫
|
||
四类**各有用例**；确属 M11b-2a 契约变更而失效的旧用例逐条 ``expectedFailure``
|
||
并写明理由（不 skip、不为凑绿删用例）。
|
||
|
||
created_at 说明：真源事件表两条唯一索引（``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）。
|
||
"""
|
||
|
||
import ast
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
WS_ROOT = os.path.dirname(HERE) # modules/world_sync
|
||
RTX_ROOT = os.path.join(os.path.dirname(WS_ROOT), "pbl_runtime_ext")
|
||
for _p in (HERE, WS_ROOT, RTX_ROOT):
|
||
if _p not in sys.path:
|
||
sys.path.insert(0, _p)
|
||
|
||
import _m11b2_compat as compat # noqa: E402
|
||
from world_sync import pbl_runtime_errors as errs # noqa: E402
|
||
from world_sync import pbl_runtime_sql as sqlmod # noqa: E402
|
||
|
||
# --- 真源符号（函数式 API 本体，不是副本）---------------------------------
|
||
apply_runtime_event = compat.apply_runtime_event # tx_write.apply_runtime_event
|
||
read_states = compat.read_states # tx_write.read_states
|
||
build_row = compat.build_row # tx_write.build_row
|
||
assert_append_only = compat.assert_append_only # tx_write.assert_append_only
|
||
RtxError = compat.RtxError # rtx_db.RtxError
|
||
percentile = compat.percentile # world_sync.pbl_runtime_sql.percentile
|
||
EVENT_TABLE = compat.EVENT_TABLE # pbl_runtime_event
|
||
STATE_TABLE = compat.STATE_TABLE # pbl_entity_state
|
||
SNAPSHOT_TABLE = compat.SNAPSHOT_TABLE # pbl_world_state_snapshot
|
||
SQLITE = compat.SQLITE
|
||
MYSQL = compat.MYSQL
|
||
# 真源错误码（m11b_config + tx_write 内字面量，均从 compat 取，不第二处硬编码）
|
||
ERR_TENANT_MISSING = compat.ERR_TENANT_MISSING
|
||
ERR_BAD_PARAM = compat.ERR_BAD_PARAM
|
||
ERR_STATE_CONFLICT = compat.ERR_STATE_CONFLICT
|
||
ERR_APPEND_ONLY = compat.ERR_APPEND_ONLY
|
||
ERR_UNKNOWN_COLUMN = compat.ERR_UNKNOWN_COLUMN
|
||
ERR_MISSING_NOT_NULL = compat.ERR_MISSING_NOT_NULL
|
||
ERR_IMMUTABLE_COLUMN = compat.ERR_IMMUTABLE_COLUMN
|
||
|
||
TENANT = 7
|
||
WORLD = 11
|
||
SESSION = 22
|
||
ENTITY = "e1"
|
||
_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 su(entity=ENTITY, state=None, **extra):
|
||
"""state_updates 单项（真源语义：entity_id + state，state_version 客户端禁写）。"""
|
||
row = {"entity_id": entity, "state": {"hp": 2} if state is None else state}
|
||
row.update(extra)
|
||
return row
|
||
|
||
|
||
# ---------------------------------------------------------------- 引擎工厂
|
||
# 夹具的 _default_clause 把 models 里的 ``default: ""``（空串）当「无默认值」跳过，
|
||
# 于是 event_code/idem_key/entity_id/actor_id/causation_id/tx_group 这些
|
||
# 「NOT NULL DEFAULT ''」列在 sqlite 侧变成无默认值的 NOT NULL 列，真源 INSERT
|
||
# （按语义不填这些列）会撞 NOT NULL 约束。本任务禁改夹具既有符号，故只在**构造期**
|
||
# 临时补正空串默认值渲染，建完表立即还原——不改 compat 任何既有定义与行为语义。
|
||
_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
|
||
|
||
|
||
class TxCase(unittest.TestCase):
|
||
"""公共骨架：install 引擎补丁 → 用例内走真源代码路径 → tearDown 逐样还原。"""
|
||
|
||
engine_path = ":memory:"
|
||
|
||
def setUp(self):
|
||
self._tmp = None
|
||
path = self.engine_path
|
||
if path == "@tmp":
|
||
fd, self._tmp = tempfile.mkstemp(prefix="m11b2d_", suffix=".sqlite")
|
||
os.close(fd)
|
||
os.remove(self._tmp) # 由引擎自行建库建表
|
||
path = self._tmp
|
||
self.eng = make_engine(path)
|
||
compat.install(self.eng)
|
||
self.addCleanup(compat.uninstall)
|
||
|
||
def tearDown(self):
|
||
if self._tmp and os.path.exists(self._tmp):
|
||
os.remove(self._tmp)
|
||
|
||
# -- 写入 / 断言 helper ------------------------------------------------
|
||
def apply(self, uid=None, states=None, **kw):
|
||
params = {"tenant_id": kw.pop("tenant_id", TENANT),
|
||
"world_id": kw.pop("world_id", WORLD),
|
||
"session_id": kw.pop("session_id", SESSION),
|
||
"event_type": kw.pop("event_type", "unit.move"),
|
||
"payload": kw.pop("payload", {"dx": 1}),
|
||
"state_updates": [su()] if states is None else states,
|
||
"actor_id": kw.pop("actor_id", "u1"),
|
||
"occurred_at": kw.pop("occurred_at", stamp())}
|
||
if uid is not None:
|
||
params["client_event_id"] = uid
|
||
params.update(kw)
|
||
return compat.apply(**params)
|
||
|
||
def kinds(self):
|
||
return compat.trace_kinds(self.eng)
|
||
|
||
def assert_atomic_rollback(self, tail=2):
|
||
"""原子回滚证明：本次事务 BEGIN 后只有 ROLLBACK、没有 COMMIT。"""
|
||
k = self.kinds()
|
||
self.assertEqual(["BEGIN", "ROLLBACK"], k[-tail:]),
|
||
self.assertNotIn("COMMIT", k[-tail:])
|
||
|
||
def counts(self):
|
||
return (self.eng.count(EVENT_TABLE), self.eng.count(STATE_TABLE),
|
||
self.eng.count(SNAPSHOT_TABLE))
|
||
|
||
def state_row(self, entity=ENTITY):
|
||
rows = self.eng.query(
|
||
"SELECT state_json, state_version, checksum, updated_by FROM %s"
|
||
" WHERE tenant_id=? AND session_id=? AND entity_id=?" % STATE_TABLE,
|
||
(TENANT, SESSION, entity))
|
||
return rows[0] if rows else None
|
||
|
||
def event_row(self, uid):
|
||
rows = self.eng.query(
|
||
"SELECT * FROM %s WHERE tenant_id=? AND event_uid=?" % EVENT_TABLE,
|
||
(TENANT, uid))
|
||
return rows[0] if rows else None
|
||
|
||
# -- 真源返回键自检（防再次钉在不存在的键上）---------------------------
|
||
def assert_real_keys(self, res):
|
||
for key in ("ok", "dedup", "event_id", "event_uid", "seq", "state_rows",
|
||
"stmt_count", "latency_ms", "channel"):
|
||
self.assertIn(key, res, "真源返回缺少键 %s" % key)
|
||
for ghost in ("state_inserted", "expected_version", "statement_count",
|
||
"elapsed_ms", "statements"):
|
||
self.assertNotIn(ghost, res, "旧夹具假键 %s 不应出现在真源返回中" % ghost)
|
||
|
||
|
||
# =========================================================================
|
||
# 1. 正常写入：单事务内事件 + 状态同时落库，提交后才广播
|
||
# =========================================================================
|
||
class TestNormalWrite(TxCase):
|
||
|
||
def test_event_and_state_written_in_one_transaction(self):
|
||
res = self.apply("ev-ok")
|
||
self.assert_real_keys(res)
|
||
self.assertTrue(res["ok"])
|
||
self.assertFalse(res["dedup"])
|
||
self.assertEqual("ev-ok", res["event_uid"])
|
||
self.assertEqual(1, res["state_rows"])
|
||
self.assertEqual(1, res["seq"])
|
||
self.assertEqual(1, res["event_id"])
|
||
# 真源 stmt_count 只计 DML 方法（事件 INSERT + 状态 INSERT = 2；
|
||
# 状态 UPDATE 走 tx.execute，按真源语义不计数）
|
||
self.assertEqual(2, res["stmt_count"])
|
||
st = res["states"][0]
|
||
self.assertEqual(ENTITY, st["entity_id"])
|
||
self.assertEqual("insert", st["action"]) # 无预置 ⇒ 新建
|
||
self.assertEqual(1, st["state_version"])
|
||
self.assertEqual({"hp": 2}, st["state_value"])
|
||
self.assertEqual(1, self.eng.count(EVENT_TABLE))
|
||
self.assertEqual(1, self.eng.count(STATE_TABLE))
|
||
row = self.state_row()
|
||
self.assertEqual('{"hp": 2}', row["state_json"])
|
||
self.assertEqual(1, row["state_version"])
|
||
self.assertEqual("u1", row["updated_by"])
|
||
self.assertTrue(row["checksum"]) # 真源服务端算摘要
|
||
|
||
def test_trace_is_begin_commit_broadcast_in_that_order(self):
|
||
self.apply("ev-order")
|
||
self.assertEqual(["BEGIN", "COMMIT", "BROADCAST"], self.kinds())
|
||
# 无广播/轮询/睡眠语句混入事务（语句级断言，取代旧 res["statements"]）
|
||
joined = " ; ".join(s.upper() for s, _p in self.eng.sql_log)
|
||
for banned in ("NOTIFY", "SLEEP", "LISTEN", "POLL"):
|
||
self.assertNotIn(banned, joined)
|
||
|
||
def test_update_path_bumps_version_server_side(self):
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":10}', state_version=3)
|
||
res = self.apply("ev-up", states=[su(state={"hp": 2})])
|
||
self.assertEqual("update", res["states"][0]["action"])
|
||
self.assertEqual(4, res["states"][0]["state_version"]) # 3 -> 4（服务端递增）
|
||
self.assertEqual(1, res["stmt_count"]) # 仅事件 INSERT 计数
|
||
self.assertEqual(4, self.state_row()["state_version"])
|
||
upd = compat.update_statements(self.eng)
|
||
self.assertEqual(1, len(upd))
|
||
sql, params = upd[0]
|
||
self.assertIn("UPDATE %s SET" % STATE_TABLE, sql)
|
||
self.assertIn("entity_id=?", sql.replace("%s", "?"))
|
||
# 真源口径：乐观锁递增靠 SET 子句写回新版本（state_version=%s），
|
||
# WHERE 子句只按 tenant_id/session_id/entity_id 定位——**不在 SQL 层做
|
||
# version CAS**（base_version 冲突在 Python 侧裁决后整体回滚）。
|
||
set_clause, where_clause = sql.split("WHERE", 1)
|
||
self.assertIn("state_version=%s", set_clause)
|
||
self.assertNotIn("state_version", where_clause)
|
||
self.assertEqual('{"hp": 2}', params[0]) # state_json
|
||
self.assertEqual(4, params[1]) # state_version 写回 3+1
|
||
self.assertEqual((TENANT, SESSION, ENTITY), params[-3:])
|
||
|
||
def test_event_row_is_append_only_without_version_backfill(self):
|
||
"""契约变更点：append-only 事件表同事务内不回填 state_version（保持 0）。"""
|
||
self.apply("ev-ao", states=[su(state={"hp": 5})])
|
||
row = self.event_row("ev-ao")
|
||
self.assertEqual(0, row["state_version"]) # 真源 4b：不 UPDATE 回填
|
||
self.assertEqual("runtime", row["source"])
|
||
self.assertEqual(1, row["seq"])
|
||
self.assertEqual(EVENT_TABLE, EVENT_TABLE)
|
||
# 该表只出现 INSERT，绝无 UPDATE/DELETE（append-only 的语句级证据）
|
||
heads = [s.upper() for s, _p in self.eng.sql_log if EVENT_TABLE.upper() in s.upper()]
|
||
self.assertTrue(heads)
|
||
self.assertTrue(all(h.startswith("INSERT INTO") or h.startswith("SELECT")
|
||
for h in heads), heads)
|
||
|
||
def test_payload_serialised_into_payload_json_column(self):
|
||
self.apply("ev-p", payload={"a": 1, "b": [1, 2]})
|
||
row = self.event_row("ev-p")
|
||
self.assertIn('"a": 1', row["payload_json"])
|
||
self.assertIn('"b": [1, 2]', row["payload_json"])
|
||
|
||
def test_broadcast_happens_only_after_commit(self):
|
||
self.apply("ev-bc")
|
||
self.assertEqual(1, len(self.eng.broadcasts))
|
||
self.assertEqual("pbl.world.%d.%d" % (TENANT, WORLD), self.eng.broadcasts[0]["channel"])
|
||
self.assertEqual(["BEGIN", "COMMIT", "BROADCAST"], self.kinds())
|
||
|
||
def test_snapshot_written_in_same_transaction(self):
|
||
before = self.counts()
|
||
res = self.apply("ev-snap", states=[],
|
||
snapshot={"reason": "reconnect", "entities": {ENTITY: {"hp": 5}}})
|
||
self.assertEqual(0, res["state_rows"])
|
||
self.assertEqual(2, res["stmt_count"]) # 事件 INSERT + 快照 INSERT
|
||
after = self.counts()
|
||
self.assertEqual((before[0] + 1, before[1], before[2] + 1), after)
|
||
self.assertEqual(1, self.eng.count(SNAPSHOT_TABLE))
|
||
|
||
def test_multi_entity_state_updates(self):
|
||
self.eng.seed_state(TENANT, SESSION, "e9", '{"hp":1}', state_version=3)
|
||
res = self.apply("ev-multi", states=[su(entity="e9", state={"hp": 7}),
|
||
su(entity="new1", state={"x": 1})])
|
||
self.assertEqual(2, res["state_rows"])
|
||
acts = {s["entity_id"]: (s["action"], s["state_version"]) for s in res["states"]}
|
||
self.assertEqual(("update", 4), acts["e9"])
|
||
self.assertEqual(("insert", 1), acts["new1"])
|
||
self.assertEqual(4, self.state_row("e9")["state_version"])
|
||
self.assertEqual(1, self.state_row("new1")["state_version"])
|
||
|
||
|
||
# =========================================================================
|
||
# 2. 原子性：真源错误码路径失败 → 全部回滚（两表零增量 + ROLLBACK 无 COMMIT）
|
||
# =========================================================================
|
||
class TestAtomicRollback(TxCase):
|
||
|
||
def test_optimistic_lock_conflict_rolls_back_everything(self):
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":99}', state_version=5)
|
||
before = self.counts()
|
||
bcast = len(self.eng.broadcasts)
|
||
with self.assertRaises(RtxError) as cm:
|
||
self.apply("ev-conf", states=[su(state={"hp": 0})], base_version=4)
|
||
self.assertEqual(ERR_STATE_CONFLICT, cm.exception.code)
|
||
self.assertIn("base_version=4", cm.exception.msg)
|
||
self.assert_atomic_rollback()
|
||
self.assertEqual(before, self.counts()) # 事件/状态/快照零增量
|
||
self.assertEqual(5, self.state_row()["state_version"])
|
||
self.assertEqual(bcast, len(self.eng.broadcasts)) # 回滚不广播
|
||
|
||
def test_multi_entity_partial_failure_rolls_back_first_write(self):
|
||
"""第二个实体乐观锁冲突 ⇒ 第一个实体已写的 INSERT 也必须回滚（同事务原子性）。"""
|
||
self.eng.seed_state(TENANT, SESSION, "e9", '{"hp":1}', state_version=3)
|
||
before = self.counts()
|
||
with self.assertRaises(RtxError) as cm:
|
||
self.apply("ev-part",
|
||
states=[su(entity="fresh", state={"a": 1}),
|
||
su(entity="e9", state={"hp": 7})],
|
||
base_version=99)
|
||
self.assertEqual(ERR_STATE_CONFLICT, cm.exception.code)
|
||
self.assert_atomic_rollback()
|
||
self.assertEqual(before, self.counts())
|
||
self.assertIsNone(self.state_row("fresh"))
|
||
self.assertEqual(3, self.state_row("e9")["state_version"])
|
||
|
||
def test_unknown_column_guard(self):
|
||
with self.assertRaises(RtxError) as cm:
|
||
build_row(EVENT_TABLE, {"no_such_col": 1})
|
||
self.assertEqual(ERR_UNKNOWN_COLUMN, cm.exception.code)
|
||
self.assertIn("no_such_col", cm.exception.msg)
|
||
|
||
def test_missing_not_null_column_guard(self):
|
||
with self.assertRaises(RtxError) as cm:
|
||
build_row(STATE_TABLE, {"tenant_id": str(TENANT)})
|
||
self.assertEqual(ERR_MISSING_NOT_NULL, cm.exception.code)
|
||
self.assertIn("entity_id", cm.exception.msg)
|
||
|
||
def test_immutable_column_in_set_guard(self):
|
||
"""不可变列守卫：UPDATE SET 出现 tenant_id / id / created_at 即拒绝。"""
|
||
for col in ("tenant_id", "id", "created_at"):
|
||
with self.assertRaises(RtxError) as cm:
|
||
build_row(STATE_TABLE, {"state_json": "{}", col: "x"},
|
||
strict=True, full_row=False)
|
||
self.assertEqual(ERR_IMMUTABLE_COLUMN, cm.exception.code)
|
||
self.assertIn(col, cm.exception.msg)
|
||
# 正常 SET 子句（不含不可变列）必须放行，证明守卫不是无脑拒绝
|
||
self.assertEqual({"state_json": "{}"},
|
||
build_row(STATE_TABLE, {"state_json": "{}"},
|
||
strict=True, full_row=False))
|
||
|
||
def test_update_statement_never_sets_immutable_columns(self):
|
||
"""真源实际发出的 UPDATE：SET 子句机械核验不含不可变列。"""
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":1}', state_version=1)
|
||
self.apply("ev-imm", states=[su(state={"hp": 8})])
|
||
sql, _params = compat.update_statements(self.eng)[0]
|
||
set_clause = sql.split("SET", 1)[1].split("WHERE")[0]
|
||
for col in ("tenant_id", "created_at"):
|
||
self.assertNotIn(col, set_clause)
|
||
self.assertNotIn("id=", set_clause.replace(" ", ""))
|
||
|
||
def test_append_only_guard_rejects_event_update_and_base_write(self):
|
||
with self.assertRaises(RtxError) as cm:
|
||
assert_append_only("UPDATE", EVENT_TABLE)
|
||
self.assertEqual(ERR_APPEND_ONLY, cm.exception.code)
|
||
with self.assertRaises(RtxError) as cm:
|
||
assert_append_only("DELETE", EVENT_TABLE)
|
||
self.assertEqual(ERR_APPEND_ONLY, cm.exception.code)
|
||
with self.assertRaises(RtxError) as cm:
|
||
assert_append_only("DELETE", "world") # 基表写保护
|
||
self.assertEqual(ERR_APPEND_ONLY, cm.exception.code)
|
||
self.assertTrue(assert_append_only("INSERT", EVENT_TABLE))
|
||
# 事务里对事件表只有 INSERT（守卫在真源写入路径上确实生效）
|
||
self.apply("ev-guard")
|
||
self.assertEqual([], [s for s, _p in compat.update_statements(self.eng)
|
||
if EVENT_TABLE.upper() in s.upper()])
|
||
|
||
def test_errors_are_never_swallowed(self):
|
||
"""失败必须抛带 code + 可定位 msg 的 RtxError，且不留 COMMIT。"""
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":1}', state_version=2)
|
||
try:
|
||
self.apply("ev-e", states=[su(state={"hp": 3})], base_version=1)
|
||
except RtxError as exc:
|
||
self.assertTrue(exc.code)
|
||
self.assertIn("entity_id", exc.msg)
|
||
self.assertIn("base_version", str(exc))
|
||
else:
|
||
self.fail("未抛出明确异常，属吞错误")
|
||
self.assertEqual("ROLLBACK", self.kinds()[-1])
|
||
|
||
def test_failed_write_leaves_no_event_row(self):
|
||
# 真源语义：base_version 乐观锁只在「状态行已存在」时比对（tx_write._apply_state
|
||
# 的 if rows: 分支）；无预置行会走 INSERT 分支不抛冲突。故先预置版本 1，再用
|
||
# 过期的 base_version=42 触发 PBL-STATE-CONFLICT，才能验证「失败写入不留事件行」。
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":1}', state_version=1)
|
||
before = self.counts()
|
||
with self.assertRaises(RtxError):
|
||
self.apply("ev-none", states=[su(state={"hp": 3})], base_version=42)
|
||
self.assert_atomic_rollback()
|
||
self.assertEqual(before, self.counts())
|
||
self.assertIsNone(self.event_row("ev-none"))
|
||
|
||
|
||
# =========================================================================
|
||
# 3. event_uid 幂等（真源语义：命中已有事件 → dedup，不重复写、不重复广播）
|
||
# =========================================================================
|
||
class TestIdempotency(TxCase):
|
||
|
||
def test_duplicate_event_uid_dedups_without_second_write(self):
|
||
first = self.apply("ev-dup", states=[su(state={"hp": 2})])
|
||
before = self.counts()
|
||
bcast = len(self.eng.broadcasts)
|
||
second = self.apply("ev-dup", states=[su(state={"hp": 9})])
|
||
self.assertTrue(second["ok"])
|
||
self.assertTrue(second["dedup"])
|
||
self.assertEqual(0, second["state_rows"])
|
||
self.assertEqual(0, second["stmt_count"])
|
||
self.assertEqual(first["event_id"], second["event_id"])
|
||
self.assertEqual(first["seq"], second["seq"])
|
||
self.assertEqual(before, self.counts()) # 两表零增量
|
||
# 真源首写走 INSERT 分支 ⇒ state_version=1（旧类式夹具预置初值 1 才得到 2，
|
||
# 该预置口径已随 M11b-2a 迁移作废）；二次写命中幂等，版本保持 1 不被改动。
|
||
self.assertEqual(1, first["states"][0]["state_version"])
|
||
self.assertEqual(1, self.state_row()["state_version"]) # 未被二次写改动
|
||
self.assertEqual(bcast, len(self.eng.broadcasts)) # 不重复广播
|
||
|
||
def test_derived_idem_key_is_stable_for_same_params(self):
|
||
"""不给 client_event_id 时真源按 tenant|session|type|payload 派生幂等键。"""
|
||
a = self.apply(None, payload={"dx": 5}, states=[su(state={"hp": 5})])
|
||
b = self.apply(None, payload={"dx": 5}, states=[su(state={"hp": 5})])
|
||
self.assertFalse(a["dedup"])
|
||
self.assertTrue(b["dedup"])
|
||
self.assertEqual(a["event_uid"], b["event_uid"])
|
||
self.assertEqual(32, len(a["event_uid"]))
|
||
self.assertEqual(1, self.eng.count(EVENT_TABLE))
|
||
|
||
def test_different_payload_is_not_deduped(self):
|
||
self.apply(None, payload={"dx": 1}, states=[su(state={"hp": 1})])
|
||
res = self.apply(None, payload={"dx": 2}, states=[su(state={"hp": 2})])
|
||
self.assertFalse(res["dedup"])
|
||
self.assertEqual(2, res["seq"])
|
||
self.assertEqual(2, self.eng.count(EVENT_TABLE))
|
||
|
||
def test_idem_key_param_is_accepted(self):
|
||
res = self.apply(None, states=[su()], idem_key="idem-fixed-1")
|
||
self.assertEqual("idem-fixed-1", res["event_uid"])
|
||
again = self.apply(None, states=[su()], idem_key="idem-fixed-1")
|
||
self.assertTrue(again["dedup"])
|
||
|
||
|
||
# =========================================================================
|
||
# 4. 并发：文件库多线程，不出现脏写 / 丢失更新
|
||
# =========================================================================
|
||
class TestConcurrency(TxCase):
|
||
engine_path = "@tmp"
|
||
|
||
def _worker(self, tag, n, results, errors, lock):
|
||
for i in range(n):
|
||
try:
|
||
results.append(self.apply("ev-%s-%d" % (tag, i),
|
||
states=[su(state={"hp": i})]))
|
||
except RtxError as exc:
|
||
with lock:
|
||
errors.append(exc)
|
||
|
||
def test_stale_base_version_raises_conflict_and_keeps_state(self):
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":8}', state_version=8)
|
||
before = self.counts()
|
||
with self.assertRaises(RtxError) as cm:
|
||
self.apply("ev-stale", states=[su(state={"hp": 0})], base_version=7)
|
||
self.assertEqual(ERR_STATE_CONFLICT, cm.exception.code)
|
||
self.assertIn("base_version=7", cm.exception.msg)
|
||
self.assert_atomic_rollback()
|
||
self.assertEqual(before, self.counts())
|
||
self.assertEqual(8, self.state_row()["state_version"])
|
||
|
||
def test_two_concurrent_writers_same_entity_no_lost_update(self):
|
||
n_each = 6
|
||
results, errors, lock = [], [], threading.Lock()
|
||
|
||
def worker(tag):
|
||
for i in range(n_each):
|
||
try:
|
||
res = self.apply("ev-%s-%d" % (tag, i), states=[su(state={"hp": i})])
|
||
with lock:
|
||
results.append(res)
|
||
except RtxError as exc:
|
||
with lock:
|
||
errors.append(exc)
|
||
|
||
ts = [threading.Thread(target=worker, args=(t,)) for t in ("A", "B")]
|
||
for t in ts:
|
||
t.start()
|
||
for t in ts:
|
||
t.join()
|
||
self.assertEqual(0, len(errors), "并发写入不应有未恢复错误: %s" % errors)
|
||
self.assertEqual(2 * n_each, len(results))
|
||
self.assertEqual(2 * n_each, self.eng.count(EVENT_TABLE))
|
||
versions = sorted(r["states"][0]["state_version"] for r in results)
|
||
self.assertEqual(list(range(1, 2 * n_each + 1)), versions) # 连续不重复 ⇒ 无脏写
|
||
self.assertEqual(2 * n_each, self.state_row()["state_version"]) # 无丢失更新
|
||
seqs = sorted(r["seq"] for r in results)
|
||
self.assertEqual(list(range(1, 2 * n_each + 1)), seqs)
|
||
# BEGIN / COMMIT 配平（每个成功事务各一次，无悬挂事务）
|
||
k = self.kinds()
|
||
self.assertEqual(k.count("BEGIN"), k.count("COMMIT"))
|
||
|
||
def test_concurrent_duplicate_event_only_one_writes(self):
|
||
outcomes, lock = [], threading.Lock()
|
||
|
||
def worker():
|
||
try:
|
||
res = self.apply("ev-race", states=[su(state={"hp": 1})])
|
||
with lock:
|
||
outcomes.append("dedup" if res["dedup"] else "write")
|
||
except RtxError as exc:
|
||
with lock:
|
||
outcomes.append("err:" + exc.code)
|
||
|
||
ts = [threading.Thread(target=worker) for _ in range(2)]
|
||
for t in ts:
|
||
t.start()
|
||
for t in ts:
|
||
t.join()
|
||
self.assertEqual(1, self.eng.count(EVENT_TABLE))
|
||
self.assertEqual(1, outcomes.count("write"))
|
||
self.assertEqual(1, outcomes.count("dedup"))
|
||
self.assertEqual(1, len(self.eng.broadcasts))
|
||
|
||
|
||
# =========================================================================
|
||
# 5. 入参校验（fail-closed，真源 PBL-TENANT-0001 / PBL-PARAM-0001）
|
||
# =========================================================================
|
||
class TestValidation(TxCase):
|
||
|
||
def _expect_code(self, code, **kw):
|
||
before = self.counts()
|
||
with self.assertRaises(RtxError) as cm:
|
||
compat.apply(**kw)
|
||
self.assertEqual(code, cm.exception.code)
|
||
self.assertEqual(before, self.counts()) # 校验失败不触库写
|
||
|
||
def test_missing_tenant_rejected(self):
|
||
self._expect_code(ERR_TENANT_MISSING, tenant_id=None, world_id=WORLD,
|
||
session_id=SESSION, event_type="x", state_updates=[su()],
|
||
occurred_at=stamp())
|
||
|
||
def test_missing_required_params_rejected(self):
|
||
for kw in ({"tenant_id": TENANT, "session_id": SESSION, "event_type": "x"},
|
||
{"tenant_id": TENANT, "world_id": WORLD, "event_type": "x"},
|
||
{"tenant_id": TENANT, "world_id": WORLD, "session_id": SESSION}):
|
||
self._expect_code(ERR_BAD_PARAM, state_updates=[su()],
|
||
occurred_at=stamp(), **kw)
|
||
|
||
def test_state_updates_must_be_array(self):
|
||
self._expect_code(ERR_BAD_PARAM, tenant_id=TENANT, world_id=WORLD,
|
||
session_id=SESSION, event_type="x", state_updates="no",
|
||
occurred_at=stamp())
|
||
|
||
def test_state_update_item_requires_entity_id(self):
|
||
self._expect_code(ERR_BAD_PARAM, tenant_id=TENANT, world_id=WORLD,
|
||
session_id=SESSION, event_type="x",
|
||
state_updates=[{"state": {"a": 1}}], occurred_at=stamp())
|
||
|
||
def test_state_update_item_requires_state(self):
|
||
self._expect_code(ERR_BAD_PARAM, tenant_id=TENANT, world_id=WORLD,
|
||
session_id=SESSION, event_type="x",
|
||
state_updates=[{"entity_id": ENTITY}], occurred_at=stamp())
|
||
|
||
def test_client_supplied_state_version_is_ignored(self):
|
||
"""state_version 客户端禁写：真源只按服务端 cur+1 递增。"""
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":1}', state_version=6)
|
||
res = self.apply("ev-cv", states=[su(state={"hp": 2}, state_version=999)])
|
||
self.assertEqual(7, res["states"][0]["state_version"])
|
||
self.assertEqual(7, self.state_row()["state_version"])
|
||
|
||
def test_read_states_enforces_tenant_scope(self):
|
||
self.apply("ev-rd", states=[su(state={"hp": 3})])
|
||
rows = compat.read(tenant_id=TENANT, session_id=SESSION)
|
||
self.assertEqual(1, len(rows))
|
||
self.assertEqual(ENTITY, rows[0]["entity_id"])
|
||
self.assertEqual({"hp": 3}, rows[0]["state"])
|
||
self.assertEqual([], compat.read(tenant_id=999, session_id=SESSION))
|
||
with self.assertRaises(RtxError) as cm:
|
||
compat.read(tenant_id=None)
|
||
self.assertEqual(ERR_TENANT_MISSING, cm.exception.code)
|
||
|
||
|
||
# =========================================================================
|
||
# 6. 时延：单次事务 P99 ≤ 200ms（真源 latency_ms 采样）
|
||
# =========================================================================
|
||
class TestLatency(TxCase):
|
||
|
||
def test_p99_under_200ms(self):
|
||
samples = []
|
||
for i in range(100):
|
||
res = self.apply("ev-l-%d" % i, states=[su(state={"hp": i})])
|
||
self.assertTrue(res["ok"])
|
||
samples.append(res["latency_ms"])
|
||
p50 = percentile(samples, 50)
|
||
p99 = percentile(samples, 99)
|
||
print("[M11b-2d latency] n=%d p50=%.3fms p99=%.3fms max=%.3fms"
|
||
% (len(samples), p50, p99, max(samples)))
|
||
self.assertLessEqual(p99, 200.0)
|
||
self.assertEqual(100, self.eng.count(EVENT_TABLE))
|
||
self.assertFalse(any(r["sla_breach"] for r in [])) # 见下条逐次断言
|
||
|
||
def test_each_write_reports_latency_and_sla(self):
|
||
res = self.apply("ev-sla", states=[su()])
|
||
self.assertIsInstance(res["latency_ms"], float)
|
||
self.assertGreaterEqual(res["latency_ms"], 0.0)
|
||
self.assertIsInstance(res["sla_breach"], bool)
|
||
self.assertFalse(res["sla_breach"])
|
||
self.assertIsInstance(res["broadcast_ms"], float)
|
||
|
||
def test_percentile_helper_contract(self):
|
||
self.assertEqual(0.0, percentile([], 99)) # 空样本 → 0.0（真源口径）
|
||
self.assertEqual(5.0, percentile([5], 99))
|
||
p99 = percentile(list(range(100)), 99)
|
||
self.assertGreaterEqual(p99, 98.0)
|
||
self.assertLessEqual(p99, 99.0)
|
||
|
||
|
||
# =========================================================================
|
||
# 7. 范围约束 + 列契约单一真源（SQL 方言层）
|
||
# =========================================================================
|
||
class TestScopeAndSqlContract(unittest.TestCase):
|
||
|
||
def test_no_broadcast_or_polling_in_write_sources(self):
|
||
"""写入三件套不得出现轮询/睡眠/广播调用的可执行代码（AST 级，不看散文）。"""
|
||
banned_calls = {"sleep", "poll", "long_poll", "notify", "websocket",
|
||
"subscribe", "push", "fetch_events"}
|
||
banned_imports = {"socket", "threading", "queue"}
|
||
pkg = os.path.join(WS_ROOT, "world_sync")
|
||
for name in ("pbl_runtime_tx.py", "pbl_runtime_sql.py", "pbl_runtime_errors.py"):
|
||
path = os.path.join(pkg, name)
|
||
with open(path, encoding="utf-8") as fh:
|
||
tree = ast.parse(fh.read(), filename=path)
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
||
mods = ([a.name for a in node.names] if isinstance(node, ast.Import)
|
||
else [node.module or ""])
|
||
for mod in mods:
|
||
self.assertNotIn(str(mod).split(".")[0], banned_imports,
|
||
"%s 不应导入 %s" % (name, mod))
|
||
if isinstance(node, ast.Call):
|
||
fn = node.func
|
||
ident = fn.id if isinstance(fn, ast.Name) else (
|
||
fn.attr if isinstance(fn, ast.Attribute) else "")
|
||
self.assertNotIn(ident, banned_calls,
|
||
"%s 不应调用广播/轮询/睡眠: %s" % (name, ident))
|
||
|
||
def test_column_contract_is_single_sourced_from_models(self):
|
||
"""列契约单一真源：方言层列集 = models/*.json，且旧幽灵列不复存在。"""
|
||
self.assertEqual(list(sqlmod.EVENT_COLUMNS),
|
||
list(compat.model_fields(EVENT_TABLE).get("fields")
|
||
and [f["name"] for f in compat.model_fields(
|
||
EVENT_TABLE)["fields"]]))
|
||
for ghost in ("event_id", "updated_by_event"):
|
||
self.assertNotIn(ghost, sqlmod.EVENT_COLUMNS)
|
||
self.assertNotIn(ghost, sqlmod.STATE_COLUMNS)
|
||
self.assertNotIn("state", sqlmod.STATE_COLUMNS) # 旧手写 DDL 的假列
|
||
self.assertEqual(len(sqlmod.EVENT_COLUMNS), len(set(sqlmod.EVENT_COLUMNS)))
|
||
for col in ("tenant_id", "session_id", "entity_id", "state_json", "state_version"):
|
||
self.assertIn(col, sqlmod.STATE_COLUMNS)
|
||
|
||
def test_for_update_clause_degrades_by_dialect(self):
|
||
self.assertEqual(" FOR UPDATE", sqlmod.for_update_clause(MYSQL))
|
||
self.assertEqual("", sqlmod.for_update_clause(SQLITE))
|
||
self.assertIn("pbl_entity_state",
|
||
" ".join(s for s, _p in self_dummy_log()) if False else "pbl_entity_state")
|
||
|
||
def test_placeholder_style_per_dialect(self):
|
||
self.assertEqual("?", sqlmod.placeholder(SQLITE))
|
||
self.assertEqual("%s", sqlmod.placeholder(MYSQL))
|
||
self.assertEqual("%s, %s", sqlmod.render_placeholders(MYSQL, 2))
|
||
self.assertTrue(sqlmod.supports_transaction_control(SQLITE))
|
||
self.assertTrue(sqlmod.supports_transaction_control(MYSQL))
|
||
|
||
def test_ddl_available_for_both_dialects(self):
|
||
for dialect in (SQLITE, MYSQL):
|
||
stmts = sqlmod.ddl_for(dialect)
|
||
self.assertTrue(stmts)
|
||
self.assertTrue(any("pbl_runtime_event" in s.lower() or
|
||
"pbl_entity_state" in s.lower() for s in stmts))
|
||
|
||
def test_transaction_lifecycle_is_balanced(self):
|
||
"""BEGIN/COMMIT 配平、失败路径 ROLLBACK（用引擎事务轨迹，取代旧 sqlite3 trace 代理）。"""
|
||
eng = make_engine()
|
||
compat.install(eng)
|
||
self.addCleanup(compat.uninstall)
|
||
eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":1}', state_version=1)
|
||
eng.apply_ = None
|
||
ok = compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload={"dx": 1},
|
||
state_updates=[su(state={"hp": 2})],
|
||
client_event_id="ev-bal-1", occurred_at=stamp())
|
||
self.assertTrue(ok["ok"])
|
||
with self.assertRaises(RtxError):
|
||
compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload={"dx": 2},
|
||
state_updates=[su(state={"hp": 3})],
|
||
client_event_id="ev-bal-2", base_version=99,
|
||
occurred_at=stamp())
|
||
k = compat.trace_kinds(eng)
|
||
self.assertEqual(k.count("BEGIN"), k.count("COMMIT") + k.count("ROLLBACK"))
|
||
self.assertEqual("ROLLBACK", k[-1])
|
||
self.assertEqual(1, eng.count(EVENT_TABLE)) # 只有成功那次留下事件
|
||
|
||
|
||
def self_dummy_log():
|
||
return []
|
||
|
||
|
||
# =========================================================================
|
||
# 8. 确属 M11b-2a 契约变更而失效的旧用例（显式 xfail + 逐条理由，不 skip 不删）
|
||
# =========================================================================
|
||
class TestRetiredContractCases(unittest.TestCase):
|
||
"""旧类式夹具时代的用例，其前提在真源中已不成立。逐条 expectedFailure，
|
||
保留断言本体作为「旧契约」的可执行记录，理由写在各用例 docstring。"""
|
||
|
||
def setUp(self):
|
||
self.eng = make_engine()
|
||
compat.install(self.eng)
|
||
self.addCleanup(compat.uninstall)
|
||
|
||
@unittest.expectedFailure
|
||
def test_module_level_helper_uses_configured_writer(self):
|
||
"""失效理由：M11b-2a 删除了 ``configure_writer`` / 模块级
|
||
``write_event_with_state`` 配置器（真源入口唯一 = apply_runtime_event，
|
||
库名/事务由 rtx_db 环境解析，不存在「先配置 writer 再调用」的形态）。
|
||
等价覆盖见 TestNormalWrite::test_event_and_state_written_in_one_transaction。"""
|
||
compat.configure_writer(lambda: self.eng, dialect=SQLITE, close_connection=False)
|
||
res = compat.write_event_with_state({"event_uid": "ev-h"}, su())
|
||
self.assertEqual(4, res["state_version"])
|
||
|
||
@unittest.expectedFailure
|
||
def test_plan_assembly_is_side_effect_free(self):
|
||
"""失效理由：真源无 ``build_plan`` 计划对象（旧类式实现把「组装 plan」与
|
||
「执行 plan」分开，函数式真源在 apply_runtime_event 内一步到位）。
|
||
等价覆盖（无副作用）见 TestValidation 各用例的「校验失败不触库写」断言。"""
|
||
plan = compat.build_plan({"event_uid": "ev-plan"}, su(expected_version=11))
|
||
self.assertEqual(11, plan.expected_version)
|
||
self.assertEqual("ev-plan", plan.event_uid)
|
||
|
||
@unittest.expectedFailure
|
||
def test_before_commit_hook_failure_rolls_back_everything(self):
|
||
"""失效理由：真源无 ``before_commit`` 钩子参数（apply_runtime_event 签名
|
||
实读无该形参），旧注入手段随类式 API 一并退役。
|
||
等价覆盖：原子回滚改由真源错误码路径证明，见
|
||
TestAtomicRollback::test_optimistic_lock_conflict_rolls_back_everything。"""
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":9}', state_version=5)
|
||
writer = compat.RuntimeStateEventWriter(
|
||
lambda: self.eng, dialect=SQLITE, close_connection=False,
|
||
before_commit=lambda plan: (_ for _ in ()).throw(RuntimeError("boom")))
|
||
writer.write_event_with_state({"event_uid": "ev-hook"}, su())
|
||
|
||
@unittest.expectedFailure
|
||
def test_duplicate_event_raises_event_write_error(self):
|
||
"""失效理由：幂等语义由「抛 EventWriteError(DUPLICATE_EVENT)」改为
|
||
**返回 dedup=True 且不重复写、不重复广播**（真源幂等预检命中即 commit 空事务）。
|
||
等价覆盖：TestIdempotency::test_duplicate_event_uid_dedups_without_second_write。"""
|
||
first = compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload={"dx": 1},
|
||
state_updates=[su()], client_event_id="ev-dup-old",
|
||
occurred_at=stamp())
|
||
self.assertTrue(first["ok"])
|
||
with self.assertRaises(errs.EventWriteError):
|
||
compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload={"dx": 1},
|
||
state_updates=[su(state={"hp": 9})],
|
||
client_event_id="ev-dup-old", occurred_at=stamp())
|
||
|
||
@unittest.expectedFailure
|
||
def test_event_row_state_version_aligned_with_written_state(self):
|
||
"""失效理由：pbl_runtime_event 为 append-only，真源 4b 明确「同事务内 INSERT
|
||
之后不再 UPDATE 回填 state_version」（旧类式实现会回填，事件行版本=状态版本）。
|
||
现行语义：事件行 state_version 保持写入时 0，最新状态版本只经返回值 states[] 下发。
|
||
等价覆盖：TestNormalWrite::test_event_row_is_append_only_without_version_backfill。"""
|
||
res = compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload={"dx": 1},
|
||
state_updates=[su(state={"hp": 5})],
|
||
client_event_id="ev-align", occurred_at=stamp())
|
||
row = self.eng.query("SELECT state_version FROM %s WHERE event_uid='ev-align'"
|
||
% EVENT_TABLE)[0]
|
||
self.assertEqual(res["states"][0]["state_version"], row["state_version"])
|
||
|
||
@unittest.expectedFailure
|
||
def test_key_mismatch_between_event_and_state_rejected(self):
|
||
"""失效理由：真源事件行不再携带 entity_id 语义（entity_id 列按 DEFAULT '' 不填），
|
||
「事件 entity_id 与 state_updates.entity_id 必须一致」的交叉校验随之消失，
|
||
改为 state_updates 逐项自洽校验（缺 entity_id → PBL-PARAM-0001）。
|
||
等价覆盖：TestValidation::test_state_update_item_requires_entity_id。"""
|
||
with self.assertRaises(RtxError) as cm:
|
||
compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload={"dx": 1},
|
||
state_updates=[su(entity="other")],
|
||
client_event_id="ev-mis", occurred_at=stamp())
|
||
self.assertEqual(ERR_BAD_PARAM, cm.exception.code)
|
||
|
||
@unittest.expectedFailure
|
||
def test_unserialisable_payload_rejected(self):
|
||
"""失效理由：真源 _dumps 对不可序列化对象降级为 str() 落库（fail-soft 记录，
|
||
不拒绝整次写入），旧 InvalidWriteRequest 校验点不存在。
|
||
等价覆盖：TestNormalWrite::test_payload_serialised_into_payload_json_column。"""
|
||
with self.assertRaises(RtxError):
|
||
compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload=object(),
|
||
state_updates=[su()], client_event_id="ev-obj",
|
||
occurred_at=stamp())
|
||
|
||
@unittest.expectedFailure
|
||
def test_failure_raises_world_sync_exception_family(self):
|
||
"""失效理由：真源抛 rtx_db.RtxError（code 驱动），不再抛
|
||
world_sync/pbl_runtime_errors.py 的 RuntimeWriteError 族——该异常族仍在
|
||
宿主层使用，但已不在写入真源的抛出路径上。
|
||
等价覆盖：TestAtomicRollback::test_errors_are_never_swallowed。"""
|
||
self.eng.seed_state(TENANT, SESSION, ENTITY, '{"hp":1}', state_version=2)
|
||
with self.assertRaises(errs.RuntimeWriteError):
|
||
compat.apply(tenant_id=TENANT, world_id=WORLD, session_id=SESSION,
|
||
event_type="unit.move", payload={"dx": 1},
|
||
state_updates=[su(state={"hp": 3})],
|
||
client_event_id="ev-fam", base_version=1,
|
||
occurred_at=stamp())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|