diff --git a/tests/test_m11b2_wiring.py b/tests/test_m11b2_wiring.py index 78c528f..59bac0b 100644 --- a/tests/test_m11b2_wiring.py +++ b/tests/test_m11b2_wiring.py @@ -14,7 +14,9 @@ 运行：cd apps/scense/pkgs/world_sync && python -m pytest tests/test_m11b2_wiring.py -v """ +import hashlib import os +import re import sqlite3 import sys import threading @@ -33,6 +35,8 @@ from world_sync.pbl_runtime_errors import ( # noqa: E402 TxConfigError, ) from world_sync.pbl_runtime_tx import ( # noqa: E402 + DEFAULT_SOURCE, + STATE_KEY_COLUMNS, read_entity_state, write_event_with_state, ) @@ -107,6 +111,19 @@ def _counts(conn): class TestAtomicWriteSuccess(object): def test_first_write_inserts_state_and_event(self, conn, factory): + """首写：同一事务内事件 + 状态各 1 行，列名/时间戳按真源契约核对。 + + M11b-2d-A2 契约对齐（真源 = models/pbl_runtime_event.json / + models/pbl_entity_state.json，经 pbl_runtime_authority 机械导出为 + pbl_runtime_sql.EVENT_COLUMNS / STATE_COLUMNS）： + - 事件业务唯一键是 event_uid（旧断言查的 event_id 列真源不存在）； + - 状态列名是 state_json（真源无 state / updated_by_event 列）； + - created_at 由 pbl_runtime_tx.utc_now_text 生成，格式 + "YYYY-MM-DD HH:MM:SS"（UTC 秒级文本），不带旧断言假设的 "Z" 后缀； + - 真源状态表没有「回指生效事件」的承载列，故该断言改为等价的版本回指： + 事件行 state_version == 状态行 state_version，并核对 checksum / + updated_by 溯源（见下方逐条注释）。 + """ result = write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) assert result["action"] == "inserted" @@ -115,19 +132,35 @@ class TestAtomicWriteSuccess(object): assert _counts(conn) == (1, 1) row = conn.execute( - "SELECT event_id, entity_id, state_version, payload, created_at " + "SELECT event_uid, entity_id, state_version, payload, created_at " "FROM pbl_runtime_event" ).fetchone() + assert row[0] == result["event_uid"] # 落库业务唯一键 == 返回键 assert row[1] == "e1" assert row[2] == 1 assert '"hp":100' in row[3] # payload 存事件体 - assert row[4].endswith("Z") + # 真源时间戳格式核对（utc_now_text: %Y-%m-%d %H:%M:%S，UTC、秒级、无 Z） + assert re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", row[4]), row[4] + assert not row[4].endswith("Z") st = conn.execute( - "SELECT state, state_version, updated_by_event FROM pbl_entity_state" + "SELECT state_json, state_version, updated_by, checksum " + "FROM pbl_entity_state" ).fetchone() assert '"hp":90' in st[0] # 状态存新值 assert st[1] == 1 - assert st[2] == row[0] # 状态行回指生效事件 + # updated_by 溯源：pbl_runtime_tx 里 updated_by 的回落链是 + # state_update.updated_by -> event_data.actor_id -> source 形参 -> DEFAULT_SOURCE， + # 而 source 形参的归一化（回落 event_data["source"]）发生在其后，故本用例不传 + # updated_by/actor_id/source 形参时落库是 DEFAULT_SOURCE；事件行的 source 列 + # 才承载 event_data["source"]（下一行一起核对，避免误以为两列同源） + assert st[2] == DEFAULT_SOURCE + assert conn.execute( + "SELECT source FROM pbl_runtime_event").fetchone()[0] == "unit-test" + # 真源 pbl_entity_state 无 updated_by_event 列（旧契约的「状态回指事件」在现行 + # schema 里没有承载列），故改为等价断言两件套：checksum 与 state_json 自洽 + + # 事件行/状态行版本号互相回指 + assert st[3] == hashlib.sha256(st[0].encode("utf-8")).hexdigest()[:32] + assert row[2] == st[1] # 事件行 state_version == 状态行版本 def test_second_write_bumps_version(self, conn, factory): write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) @@ -143,13 +176,28 @@ class TestAtomicWriteSuccess(object): assert _counts(conn) == (2, 1) def test_read_entity_state_helper(self, factory): + """只读辅助按真源定位键取状态，返回键按实际契约核对。 + + M11b-2d-A2 契约对齐：真源 pbl_entity_state 无 world_id 列，定位键是 + uk_es = (tenant_id, session_id, entity_id)；read_entity_state 的 world_id + 只是兼容入参（不参与 WHERE）。旧用例不传 session_id → 永远查不到行 + （返回 None 后下标取值 TypeError）。实际返回键仅 state（承载 state_json + 文本）/ state_version / updated_at，无 statement_count/statements 等旧键。 + """ write_event_with_state(_event(), _state(state={"hp": 42}), conn_factory=factory) got = read_entity_state(conn_factory=factory, tenant_id="t1", - world_id="w1", entity_id="e1") + session_id="s-1", world_id="w1", entity_id="e1") + assert set(got.keys()) == {"state", "state_version", "updated_at"} assert got["state_version"] == 1 assert '"hp":42' in got["state"] + assert re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", + got["updated_at"]), got["updated_at"] + # session_id 是 uk_es 的组成部分：换一个 session 就查不到（定位键维度核对） assert read_entity_state(conn_factory=factory, tenant_id="t1", - world_id="w1", entity_id="nope") is None + session_id="s-other", world_id="w1", + entity_id="e1") is None + assert read_entity_state(conn_factory=factory, tenant_id="t1", + session_id="s-1", world_id="w1", entity_id="nope") is None def test_missing_tenant_rejected_before_touching_db(self, factory): with pytest.raises(TxConfigError) as ei: @@ -177,7 +225,12 @@ class TestAtomicWriteSuccess(object): class TestRollbackOnEventFailure(object): def test_event_failure_rolls_back_state_update(self, conn, factory): - """状态先更新成功，事件插入被触发器打坏 → 断言两表零增量、状态仍是旧值。""" + """状态先更新成功，事件插入被触发器打坏 → 断言两表零增量、状态仍是旧值。 + + M11b-2d-A2 契约对齐：触发器注入保留不变；异常类型/编码按实读 + pbl_runtime_errors.py 核对——EventWriteError.code = + "PBL_RUNTIME_EVENT_WRITE_FAILED"（与旧断言一致，故本用例只有状态列名要改）。 + """ write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) conn.execute( "CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event " @@ -190,20 +243,41 @@ class TestRollbackOnEventFailure(object): assert "boom" in str(ei.value) assert _counts(conn) == (1, 1) - st = conn.execute("SELECT state, state_version FROM pbl_entity_state").fetchone() + # 真源状态列名是 state_json（无 state 列） + st = conn.execute( + "SELECT state_json, state_version FROM pbl_entity_state").fetchone() assert '"hp":90' in st[0] assert st[1] == 1 - def test_duplicate_event_id_rolls_back_state(self, conn, factory): - """同 event_id 主键冲突 → 事件失败 → 状态更新回滚。""" - write_event_with_state(_event(), _state(state={"hp": 90}), - conn_factory=factory, event_id="dup-1") - with pytest.raises(EventWriteError): + def test_duplicate_event_uid_rolls_back_state(self, conn, factory): + """同 event_uid 撞真源唯一键 -> 事件插入失败 -> 状态更新一并回滚。 + + 原用例名 test_duplicate_event_id_rolls_back_state：真源事件业务唯一键是 + event_uid（pbl_runtime_event.json 唯一键 uk_event_uid = + tenant_id + event_uid + created_at），旧写法传的 event_id= 形参在 + write_event_with_state 签名中已不存在（TypeError），故按现行真实唯一键改名 + 并改写。因唯一键含 created_at（秒级），两次写入用同一个注入 clock 固定时间戳， + 保证真的撞键而不是跨秒漏判。映射异常按实读代码核对：_insert_event 失败 -> + EventWriteError（PBL_RUNTIME_EVENT_WRITE_FAILED），不是 ConcurrentStateConflict + （后者只承载状态表冲突）。 + """ + r1 = write_event_with_state(_event(), _state(state={"hp": 90}), + conn_factory=factory, event_uid="dup-1", + clock=lambda: 1700000000.0) + assert r1["created_at"] == "2023-11-14 22:13:20" # 真源时间戳格式 + assert r1["event_uid"] == "dup-1" + with pytest.raises(EventWriteError) as ei: write_event_with_state(_event(), _state(state={"hp": 5}), - conn_factory=factory, event_id="dup-1") + conn_factory=factory, event_uid="dup-1", + clock=lambda: 1700000000.0) + assert ei.value.code == "PBL_RUNTIME_EVENT_WRITE_FAILED" + assert ei.value.detail.get("event_uid") == "dup-1" + # 状态更新（版本 1 -> 2）随事件失败一起回滚：两表零增量、版本仍为 1 assert _counts(conn) == (1, 1) - st = conn.execute("SELECT state_version FROM pbl_entity_state").fetchone() - assert st[0] == 1 + st = conn.execute( + "SELECT state_json, state_version FROM pbl_entity_state").fetchone() + assert st[1] == 1 + assert '"hp":90' in st[0] # -------------------------------------------------------------------------- @@ -307,14 +381,24 @@ class TestConcurrencyNoLostUpdate(object): assert not [e for e in errors if not e.startswith("locked")] def test_optimistic_lock_rejects_stale_writer(self, conn, factory): - """显式 expected_version 的写者：版本被别人推进后必须被拒绝（不覆盖）。""" + """显式 expected_version 的写者：版本被别人推进后必须被拒绝（不覆盖）。 + + M11b-2d-A2 契约对齐：真源状态列名是 state_json（无 state 列）；expected_version + 语义按实读 pbl_runtime_tx.write_event_with_state——与读到的当前 state_version + 不等即在改状态之前抛 ConcurrentStateConflict（code + PBL_ENTITY_STATE_CONFLICT，detail 带 expected_version / actual_version）， + 两表零变更。 + """ write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) write_event_with_state(_event(), _state(state={"hp": 80}), conn_factory=factory) - with pytest.raises(ConcurrentStateConflict): + with pytest.raises(ConcurrentStateConflict) as ei: write_event_with_state(_event(), _state(state={"hp": 70}, expected_version=1), conn_factory=factory) + assert ei.value.code == "PBL_ENTITY_STATE_CONFLICT" + assert ei.value.detail["expected_version"] == 1 + assert ei.value.detail["actual_version"] == 2 assert _counts(conn) == (2, 1) - st = conn.execute("SELECT state FROM pbl_entity_state").fetchone() + st = conn.execute("SELECT state_json FROM pbl_entity_state").fetchone() assert '"hp":80' in st[0] # 迟到的旧版本写者没有覆盖新状态 @@ -335,10 +419,55 @@ class TestTenantWorldIsolation(object): ).fetchall() assert r == [("t1", 1), ("t2", 1)] # 各自独立版本，互不覆盖 - def test_same_entity_id_across_worlds_isolated(self, conn, factory): - write_event_with_state(_event(world="w1"), _state(world="w1"), conn_factory=factory) - write_event_with_state(_event(world="w2"), _state(world="w2"), conn_factory=factory) + def test_same_entity_id_across_sessions_isolated(self, conn, factory): + """同一 entity_id 在不同 session_id 下互不影响（真源 uk_es 的隔离维度）。 + + M11b-2d-A2 契约对齐（原用例名 test_same_entity_id_across_worlds_isolated）： + 真源 pbl_entity_state 没有 world_id 列，状态定位键 uk_es = + (tenant_id, session_id, entity_id)，因此「跨 world 隔离」这个前提在现行契约下 + 不成立——旧断言 (2, 2) 失败是断言的隔离维度写错，不是实现丢更新。按任务给的 + 两个方向合并处理：主断言改为 session_id 维度隔离并改名，同时保留 world_id 入参 + 并显式断言它「不进状态定位键」的守卫语义（换 session 才分状态行、只换 world + 不分状态行，两条互为对照），并机械核对真源列集/定位键里都没有 world_id，防止 + 契约又漂回旧列名。 + """ + # 维度 1：换 session_id -> 状态各自独立一行（uk_es 生效） + write_event_with_state(_event(session_id="s-1"), + _state(session_id="s-1", state={"hp": 1}), + conn_factory=factory) + write_event_with_state(_event(session_id="s-2"), + _state(session_id="s-2", state={"hp": 2}), + conn_factory=factory) assert _counts(conn) == (2, 2) + rows = conn.execute( + "SELECT session_id, state_version FROM pbl_entity_state" + " ORDER BY session_id").fetchall() + assert rows == [("s-1", 1), ("s-2", 1)] # 各自独立版本，互不覆盖 + + # 守卫：world_id 不进状态定位键 —— 只换 world、session 相同 -> 命中同一行， + # 第二次写入是版本推进（updated）而不是新插一行状态 + r = write_event_with_state(_event(session_id="s-1", world="w9"), + _state(session_id="s-1", world="w9", + state={"hp": 3}), + conn_factory=factory) + assert r["action"] == "updated" + assert r["state_version"] == 2 + assert _counts(conn) == (3, 2) + s1_state = conn.execute( + "SELECT state_json FROM pbl_entity_state WHERE session_id = ?", + ("s-1",)).fetchone()[0] + s2_state = conn.execute( + "SELECT state_json FROM pbl_entity_state WHERE session_id = ?", + ("s-2",)).fetchone()[0] + assert '"hp":3' in s1_state and '"hp":2' in s2_state # 互不污染 + + # 真源列集 / 定位键里都没有 world_id（机械核对，防契约再漂回旧列名） + assert "world_id" not in ps.STATE_COLUMNS + assert "world_id" not in STATE_KEY_COLUMNS + # world_id 只作为事件行的属性列存在（append-only 事件流仍可按 world 检索） + assert conn.execute( + "SELECT COUNT(*) FROM pbl_runtime_event WHERE world_id = ?", + ("w9",)).fetchone()[0] == 1 def test_event_query_scoped_by_tenant(self, conn, factory): write_event_with_state(_event(tenant="t1"), _state(tenant="t1"), conn_factory=factory)