450 lines
17 KiB
Python
450 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""M11b-2d-A 测试夹具层：**函数式 API 直连真源**的运行时事务引擎（仅测试使用）。
|
||
|
||
本文件的历史与本轮收敛（响应 QC #2 硬门禁）
|
||
==========================================================
|
||
M11b-2a「世界同步真源迁移」把写入实现迁到 **函数式 API**
|
||
``pbl_runtime_ext.tx_write.apply_runtime_event`` / ``read_states``
|
||
（见 ``modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py:344`` / ``:466``）。
|
||
|
||
本文件在 M11b-2c 时的形态是「类式垫片」：定义 ``RuntimeStateEventWriter`` 类，
|
||
持有 ``conn_factory`` / ``dialect`` / ``close_connection``，再把方法调用转发到
|
||
``world_sync/pbl_runtime_tx.write_event_with_state``。那是**第二条夹具路径**——
|
||
它把测试钉在「旧类式调用形态 + world_sync 本地同步实现」上，既没验证现行真源
|
||
``apply_runtime_event``，又与真源返回键（``action`` / ``states[]`` / ``stmt_count``
|
||
/ ``latency_ms``）不一致，垫片只能靠 ``_augment`` 补假键（``state_inserted`` 等）
|
||
维持旧断言，实测 22 failed（``Cannot operate on a closed database.`` /
|
||
``KeyError: 'expected_version'``）。
|
||
|
||
**本轮已删除该类别垫片**，本文件只保留一条夹具路径：
|
||
|
||
1. ``RuntimeEngine``：以 **真实 sqlite3** 为存储的事务引擎。表结构（列名/类型/
|
||
NOT NULL/DEFAULT/唯一索引）**从 ``pbl_runtime_ext/models/*.json`` 读取生成**
|
||
——不另立第二份 schema 定义，避免列名再次漂移（这正是 28 failed 的根因之一：
|
||
旧夹具里手写 DDL 用了 ``event_id`` / ``state`` 等真源不存在的列）。
|
||
2. ``install(engine)`` / ``uninstall()``：把 ``rtx_db.transaction`` /
|
||
``rtx_db.dbname`` / ``rtx_db.q_all`` 换成 sqlite 引擎实现，使
|
||
``tx_write.apply_runtime_event`` / ``read_states`` **走自身真实代码路径**
|
||
（幂等预检 → INSERT 事件 → UPSERT 状态 → 快照 → COMMIT → 提交后广播）。
|
||
与 ``modules/pbl_runtime_ext/scripts/m11b_selftest.py::install_fake`` 同一
|
||
思路，差别只在存储换成真 sqlite（可验证真实 ROLLBACK 与 DB 唯一索引兜底）。
|
||
3. ``apply(**kw)`` / ``read(**kw)``：函数式 API 的同步入口（``asyncio.run``）。
|
||
**不做任何返回键增补**——测试直接断言真源返回键。
|
||
4. ``percentile``：转发到 ``world_sync.pbl_runtime_sql.percentile``（真源唯一实现）。
|
||
|
||
边界（QC #3 允许路径）：只改测试夹具层，**不改业务实现、不改 models/*.json 表定义、
|
||
不改任何 load_path.py RBAC 注册**。本文件不含任何生产能力：只有被 pytest 夹具调用时
|
||
才安装引擎补丁，``uninstall()`` 逐样还原。
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
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") # modules/pbl_runtime_ext
|
||
|
||
for _p in (WS_ROOT, RTX_ROOT):
|
||
if _p not in sys.path:
|
||
sys.path.insert(0, _p)
|
||
|
||
# --- 真源导入（被测对象本体，不是副本）---------------------------------
|
||
from pbl_runtime_ext import broadcast as _bcast # noqa: E402
|
||
from pbl_runtime_ext import m11b_config as _cfg # noqa: E402
|
||
from pbl_runtime_ext import rtx_db as _rtx_db # noqa: E402
|
||
from pbl_runtime_ext import tx_write as _tx_write # noqa: E402
|
||
from world_sync import pbl_runtime_sql as ps # noqa: E402
|
||
|
||
# 旧夹具里以名字引用的常量：统一取真源，避免第二份字面量
|
||
SQLITE = ps.DIALECT_SQLITE
|
||
MYSQL = ps.DIALECT_MYSQL
|
||
percentile = ps.percentile # 真源唯一实现（world_sync.pbl_runtime_sql）
|
||
apply_runtime_event = _tx_write.apply_runtime_event # 现行函数式真源 API
|
||
read_states = _tx_write.read_states
|
||
build_row = _tx_write.build_row
|
||
assert_append_only = _tx_write.assert_append_only
|
||
RtxError = _rtx_db.RtxError
|
||
|
||
EVENT_TABLE = _cfg.EVENT_TABLE
|
||
STATE_TABLE = _cfg.STATE_TABLE
|
||
SNAPSHOT_TABLE = _cfg.SNAPSHOT_TABLE
|
||
PROTECTED_BASE_TABLES = _cfg.PROTECTED_BASE_TABLES
|
||
ERR_TENANT_MISSING = _cfg.ERR_TENANT_MISSING
|
||
ERR_BAD_PARAM = _cfg.ERR_BAD_PARAM
|
||
ERR_STATE_CONFLICT = _cfg.ERR_STATE_CONFLICT
|
||
ERR_APPEND_ONLY = _cfg.ERR_APPEND_ONLY
|
||
ERR_BROADCAST_AFTER_COMMIT = _cfg.ERR_BROADCAST_AFTER_COMMIT
|
||
# build_row 的三道门禁码（真源 tx_write.py 内以字面量抛出）
|
||
ERR_UNKNOWN_COLUMN = "PBL-RTX-0003"
|
||
ERR_MISSING_NOT_NULL = "PBL-RTX-0004"
|
||
ERR_IMMUTABLE_COLUMN = "PBL-RTX-0006"
|
||
|
||
_MODELS_DIR = os.path.join(RTX_ROOT, "models")
|
||
|
||
|
||
# ---------------------------------------------------------------- schema
|
||
def _sql_type(t):
|
||
"""抽象类型 → sqlite 存储类（sqlite 动态类型，映射只为可读性/亲和性）。"""
|
||
t = str(t or "").lower()
|
||
if t in ("long", "int", "bigint", "smallint", "short"):
|
||
return "INTEGER"
|
||
if t in ("float", "double", "ddouble", "decimal"):
|
||
return "REAL"
|
||
if t in ("text", "json"):
|
||
return "TEXT"
|
||
if t in ("date", "datetime", "time", "timestamp"):
|
||
return "TEXT"
|
||
return "TEXT"
|
||
|
||
|
||
def _not_null(fd):
|
||
if fd.get("null") is False:
|
||
return True
|
||
return str(fd.get("nullable") or "").strip().lower() in ("no", "false", "0")
|
||
|
||
|
||
def _default_clause(fd):
|
||
dv = fd.get("default")
|
||
if dv is None:
|
||
return ""
|
||
s = str(dv).strip()
|
||
if s.upper() in ("CURRENT_TIMESTAMP", "NULL", "NONE", ""):
|
||
return ""
|
||
if isinstance(dv, bool):
|
||
return " DEFAULT %d" % int(dv)
|
||
if isinstance(dv, (int, float)):
|
||
return " DEFAULT %s" % dv
|
||
return " DEFAULT '%s'" % s.replace("'", "''")
|
||
|
||
|
||
def model_fields(table):
|
||
"""表列定义（权威源 = pbl_runtime_ext/models/{table}.json）。"""
|
||
path = os.path.join(_MODELS_DIR, "%s.json" % table)
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
return json.load(fh)
|
||
|
||
|
||
def ddl_statements(table):
|
||
"""按 models/*.json 生成 sqlite DDL（列/NOT NULL/DEFAULT/唯一索引）。"""
|
||
spec = model_fields(table)
|
||
cols = []
|
||
for fd in spec.get("fields") or []:
|
||
name = fd.get("name")
|
||
if not name:
|
||
continue
|
||
if fd.get("auto_increment"):
|
||
cols.append('"%s" INTEGER PRIMARY KEY AUTOINCREMENT' % name)
|
||
continue
|
||
d = '"%s" %s' % (name, _sql_type(fd.get("type")))
|
||
if _not_null(fd):
|
||
d += " NOT NULL"
|
||
d += _default_clause(fd)
|
||
cols.append(d)
|
||
out = ['CREATE TABLE IF NOT EXISTS "%s" (\n %s\n)' % (table, ",\n ".join(cols))]
|
||
for idx in spec.get("indexes") or []:
|
||
fields = idx.get("idxfields") or []
|
||
if not fields:
|
||
continue
|
||
kind = "UNIQUE INDEX" if str(idx.get("idxtype")) == "unique" else "INDEX"
|
||
cols_sql = ", ".join('"%s"' % f for f in fields)
|
||
out.append('CREATE %s IF NOT EXISTS "%s" ON "%s" (%s)'
|
||
% (kind, idx.get("name") or ("%s_%d" % (table, len(out))), table, cols_sql))
|
||
return out
|
||
|
||
|
||
def _to_qmark(sql):
|
||
"""真源 SQL 用 sqlor 的 ``%s`` 占位；sqlite 用 ``?``。"""
|
||
return sql.replace("%s", "?")
|
||
|
||
|
||
# ---------------------------------------------------------------- 引擎
|
||
class RuntimeEngine(object):
|
||
"""真实 sqlite 事务引擎（测试专用），适配 ``rtx_db.Transaction`` 接口。
|
||
|
||
记录三条可断言的轨迹：
|
||
* ``trace`` 有序事务轨迹：``BEGIN`` / ``SQL`` / ``COMMIT`` / ``ROLLBACK`` /
|
||
``BROADCAST`` —— 用于「广播只在 commit 之后」「BEGIN/COMMIT 配平」
|
||
* ``sql_log`` 实际发出的 SQL（含参数），用于「SET 子句不含不可变列」「无广播/
|
||
轮询/睡眠语句」这类语句级断言
|
||
* ``broadcasts`` 提交后广播消息（hub.publish 录制）
|
||
"""
|
||
|
||
def __init__(self, path=":memory:", tables=(EVENT_TABLE, STATE_TABLE, SNAPSHOT_TABLE)):
|
||
self.path = path
|
||
self.tables = tuple(tables)
|
||
self.trace = []
|
||
self.sql_log = []
|
||
self.broadcasts = []
|
||
self.txn_depth = 0
|
||
self._lock = threading.RLock()
|
||
self._mem = None
|
||
if path == ":memory:":
|
||
self._mem = sqlite3.connect(":memory:", isolation_level=None,
|
||
check_same_thread=False)
|
||
self._mem.execute("PRAGMA busy_timeout=15000")
|
||
self._create(self._mem)
|
||
else:
|
||
c = sqlite3.connect(path, isolation_level=None)
|
||
c.execute("PRAGMA busy_timeout=15000")
|
||
self._create(c)
|
||
c.close()
|
||
|
||
# -- 建表 -------------------------------------------------------------
|
||
def _create(self, conn):
|
||
for t in self.tables:
|
||
for stmt in ddl_statements(t):
|
||
conn.execute(stmt)
|
||
conn.commit()
|
||
|
||
# -- 连接 -------------------------------------------------------------
|
||
def admin_conn(self):
|
||
"""给测试直接查库用的连接（文件库返回新连接，调用方负责 close）。"""
|
||
if self._mem is not None:
|
||
return self._mem, False
|
||
c = sqlite3.connect(self.path, isolation_level=None)
|
||
c.execute("PRAGMA busy_timeout=15000")
|
||
return c, True
|
||
|
||
def query(self, sql, params=()):
|
||
conn, own = self.admin_conn()
|
||
try:
|
||
cur = conn.execute(_to_qmark(sql), tuple(params or ()))
|
||
cols = [d[0] for d in cur.description] if cur.description else []
|
||
return [dict(zip(cols, r)) for r in cur.fetchall()]
|
||
finally:
|
||
if own:
|
||
conn.close()
|
||
|
||
def scalar(self, sql, params=(), default=None):
|
||
rows = self.query(sql, params)
|
||
if not rows:
|
||
return default
|
||
vals = list(rows[0].values())
|
||
return vals[0] if vals else default
|
||
|
||
def count(self, table):
|
||
return int(self.scalar("SELECT COUNT(*) AS c FROM \"%s\"" % table, (), 0) or 0)
|
||
|
||
def seed_state(self, tenant_id, session_id, entity_id, state_json,
|
||
state_version=0, checksum="", updated_by="seed"):
|
||
"""预置状态行（走真实 INSERT，不绕过列契约）。"""
|
||
now = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
conn, own = self.admin_conn()
|
||
try:
|
||
conn.execute("DELETE FROM \"%s\" WHERE tenant_id=? AND session_id=? AND entity_id=?"
|
||
% STATE_TABLE, (str(tenant_id), str(session_id), str(entity_id)))
|
||
conn.execute(
|
||
"INSERT INTO \"%s\" (tenant_id, session_id, entity_id, state_json,"
|
||
" state_version, checksum, updated_by, created_at, updated_at)"
|
||
" VALUES (?,?,?,?,?,?,?,?,?)" % STATE_TABLE,
|
||
(str(tenant_id), str(session_id), str(entity_id), state_json,
|
||
int(state_version), checksum, updated_by, now, now))
|
||
conn.commit()
|
||
finally:
|
||
if own:
|
||
conn.close()
|
||
|
||
# -- 事务轨迹 ---------------------------------------------------------
|
||
def _note(self, kind, detail=None):
|
||
with self._lock:
|
||
self.trace.append((kind, detail))
|
||
|
||
# -- 只读查询（供 rtx_db.q_all 补丁）----------------------------------
|
||
async def q_all(self, sql, params=None, db=None):
|
||
return _rtx_db._norm_rows(self.query(sql, params))
|
||
|
||
|
||
class _Tx(object):
|
||
"""sqlite 事务句柄：与 ``rtx_db.Transaction`` 同接口（execute/insert/update/
|
||
select/commit/rollback/stmt_count），**stmt_count 只计 DML**（与真源一致：
|
||
真源 ``execute`` 不计数、``insert``/``update`` 计数）。
|
||
"""
|
||
|
||
def __init__(self, engine):
|
||
self.e = engine
|
||
self.conn = None
|
||
self.committed = False
|
||
self.rolled_back = False
|
||
self.stmt_count = 0
|
||
|
||
async def __aenter__(self):
|
||
self.conn = self.e._mem if self.e._mem is not None else sqlite3.connect(
|
||
self.e.path, isolation_level=None)
|
||
if self.e._mem is None:
|
||
self.conn.execute("PRAGMA busy_timeout=15000")
|
||
with self.e._lock:
|
||
self.conn.execute("BEGIN IMMEDIATE")
|
||
self.e.txn_depth += 1
|
||
self.e._note("BEGIN")
|
||
return self
|
||
|
||
async def __aexit__(self, exc_type, exc, tb):
|
||
if exc_type is not None or not self.committed:
|
||
await self.rollback()
|
||
self._close()
|
||
return False
|
||
|
||
def _close(self):
|
||
if self.e._mem is None and self.conn is not None:
|
||
try:
|
||
self.conn.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self.conn = None
|
||
|
||
def _run(self, sql, params):
|
||
self.e.sql_log.append((" ".join(str(sql).split()), tuple(params or ())))
|
||
cur = self.conn.cursor()
|
||
try:
|
||
cur.execute(_to_qmark(sql), tuple(params or ()))
|
||
cols = [d[0] for d in cur.description] if cur.description else []
|
||
rows = [dict(zip(cols, r)) for r in cur.fetchall()] if cols else []
|
||
return rows
|
||
finally:
|
||
try:
|
||
cur.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
# -- DML / 查询 -------------------------------------------------------
|
||
async def execute(self, sql, params=None):
|
||
return self._run(sql, params)
|
||
|
||
async def insert(self, table, row):
|
||
row = dict(row)
|
||
if "tenant_id" in row and row.get("tenant_id") in (None, "", 0):
|
||
raise RtxError(ERR_TENANT_MISSING, "%s 写入缺少 tenant_id" % table)
|
||
cols = list(row.keys())
|
||
sql = 'INSERT INTO "%s" (%s) VALUES (%s)' % (
|
||
table, ", ".join('"%s"' % c for c in cols), ", ".join(["?"] * len(cols)))
|
||
self.e.sql_log.append((" ".join(sql.split()), tuple(row[c] for c in cols)))
|
||
cur = self.conn.cursor()
|
||
try:
|
||
cur.execute(sql, tuple(row[c] for c in cols))
|
||
rid = cur.lastrowid
|
||
finally:
|
||
try:
|
||
cur.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self.stmt_count += 1
|
||
return rid
|
||
|
||
async def update(self, table, values, where):
|
||
cols = list(dict(values).keys())
|
||
sql = 'UPDATE "%s" SET %s' % (table, ", ".join('"%s" = ?' % c for c in cols))
|
||
self.e.sql_log.append((" ".join(sql.split()), tuple(dict(values)[c] for c in cols)))
|
||
cur = self.conn.cursor()
|
||
try:
|
||
cur.execute(sql, tuple(dict(values)[c] for c in cols))
|
||
n = cur.rowcount
|
||
finally:
|
||
try:
|
||
cur.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self.stmt_count += 1
|
||
return n
|
||
|
||
async def select(self, table, where="", fields="*", order="", limit=200):
|
||
sql = 'SELECT %s FROM "%s"' % (fields or "*", table)
|
||
if where:
|
||
sql += " WHERE %s" % where
|
||
if order:
|
||
sql += " ORDER BY %s" % order
|
||
sql += " LIMIT %d" % int(limit or 200)
|
||
return self._run(sql, ())
|
||
|
||
# -- 事务控制 ---------------------------------------------------------
|
||
async def commit(self):
|
||
if self.committed or self.rolled_back:
|
||
return
|
||
self.conn.commit()
|
||
self.committed = True
|
||
with self.e._lock:
|
||
self.e.txn_depth -= 1
|
||
self.e._note("COMMIT")
|
||
|
||
async def rollback(self):
|
||
if self.committed or self.rolled_back:
|
||
return
|
||
try:
|
||
self.conn.rollback()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self.rolled_back = True
|
||
with self.e._lock:
|
||
self.e.txn_depth -= 1
|
||
self.e._note("ROLLBACK")
|
||
|
||
|
||
# ---------------------------------------------------------------- 安装 / 还原
|
||
_ORIG = {}
|
||
_INSTALLED = False
|
||
|
||
|
||
def install(engine):
|
||
"""把 ``rtx_db`` 的事务/查询入口换成 sqlite 引擎，并录制提交后广播。"""
|
||
global _INSTALLED
|
||
if _INSTALLED:
|
||
uninstall()
|
||
_ORIG["transaction"] = _rtx_db.transaction
|
||
_ORIG["dbname"] = _rtx_db.dbname
|
||
_ORIG["q_all"] = _rtx_db.q_all
|
||
_rtx_db.dbname = lambda env=None: "m11b2d_test_sqlite"
|
||
_rtx_db.transaction = lambda db=None: _Tx(engine)
|
||
_rtx_db.q_all = engine.q_all
|
||
|
||
_bcast.reset_hub()
|
||
hub = _bcast.get_hub()
|
||
original_publish = hub.publish
|
||
|
||
def _recording_publish(channel, event):
|
||
engine.broadcasts.append({"channel": channel, "event": event})
|
||
engine._note("BROADCAST", channel)
|
||
return original_publish(channel, event)
|
||
|
||
hub.publish = _recording_publish
|
||
_ORIG["hub"] = (hub, original_publish)
|
||
_INSTALLED = True
|
||
return engine
|
||
|
||
|
||
def uninstall():
|
||
global _INSTALLED
|
||
if not _INSTALLED:
|
||
return
|
||
for key in ("transaction", "dbname", "q_all"):
|
||
if key in _ORIG:
|
||
setattr(_rtx_db, key, _ORIG[key])
|
||
_INSTALLED = False
|
||
_ORIG.clear()
|
||
|
||
|
||
def apply(**kwargs):
|
||
"""函数式真源 API 的同步入口（不做返回键增补）。"""
|
||
return asyncio.run(apply_runtime_event(**kwargs))
|
||
|
||
|
||
def read(**kwargs):
|
||
return asyncio.run(read_states(**kwargs))
|
||
|
||
|
||
def trace_kinds(engine):
|
||
return [k for k, _d in engine.trace]
|
||
|
||
|
||
def update_statements(engine):
|
||
"""引擎实际发出的 UPDATE 语句（大写、压空格），用于不可变列/乐观锁断言。"""
|
||
return [(s, p) for s, p in engine.sql_log if s.upper().startswith("UPDATE ")]
|
||
|
||
|
||
def insert_statements(engine, table):
|
||
return [(s, p) for s, p in engine.sql_log
|
||
if s.upper().startswith('INSERT INTO "%s"' % table.upper())]
|