approve: 真实 DB 实测 + 三份验收证据落盘（QC #5）
This commit is contained in:
agent.develop
parent
13386a4e87
commit
eb7c242bb0
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user