From c1bce932e25cf852a58f1d0826b6a8199c37c008 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Tue, 22 Sep 2026 11:50:57 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/_m11b2_compat.py | 165 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 tests/_m11b2_compat.py diff --git a/tests/_m11b2_compat.py b/tests/_m11b2_compat.py new file mode 100644 index 0000000..b855bbd --- /dev/null +++ b/tests/_m11b2_compat.py @@ -0,0 +1,165 @@ +# -*- coding: utf-8 -*- +"""M11b-2c 测试夹具层兼容垫片（**仅测试使用，不属于模块运行时 API**）。 + +背景（QC #3 允许路径：只改夹具/断言，不改业务实现、不改表定义 JSON） +===================================================================== +``tests/test_m11b2_single_tx.py`` 是在 M11b-2a「世界同步真源迁移」之前写成的，它 +绑定的是当时的类式写入器 API： + + tx.RuntimeStateEventWriter / tx.build_plan / tx.percentile / tx.configure_writer + +M11b-2a 把真源实现改为**函数式 API**（``pbl_runtime_tx.write_event_with_state`` / +``read_entity_state``，见 ``docs/02-develop/dev-notes-m11b2a-world-sync-migration.md`` +与 ``world_sync/init.py`` 第 17 行的导出说明），类式写入器与模块级全局 writer 单例 +被一并移除——``git log -S RuntimeStateEventWriter -- world_sync/pbl_runtime_tx.py`` +为空即证据：该类从未存在于当前真源。 + +因此本垫片**不新增业务能力**，只做三件事： +1. ``RuntimeStateEventWriter``：把「持有 conn_factory + dialect + close_connection」 + 的旧调用形态适配到当前 ``write_event_with_state`` 函数（同一条代码路径）。 +2. ``build_plan``：转发到 ``world_sync.init.build_runtime_write_plan``（已存在的 + 不落库预演函数），保持旧名字可用。 +3. ``percentile``：转发到 ``pbl_runtime_sql.percentile``（真源唯一实现，避免第二份 + 分位数算法口径漂移）。 +4. ``configure_writer``：旧「模块级默认 writer」语义在真源已不存在；这里只在**测试 + 进程内**保存一个默认连接工厂，供 ``write_event_with_state`` 的兼容包装取用， + 不写 ServerEnv、不影响生产路径。 + +返回键差异也在垫片里收敛：真源返回 ``action``（``inserted``/``updated``/ +``event_only``），旧断言读 ``state_inserted``（bool）。垫片补 ``state_inserted`` += ``action == "inserted"``，不改动真源返回值。 +""" + +import importlib.util +import os +import sys +import types + +HERE = os.path.dirname(os.path.abspath(__file__)) +PKG_DIR = os.path.dirname(HERE) +_PKG = "_ws_m11b2_compat" + +# 与 test_m11b2_single_tx.py 使用同一个合成包名，确保加载的是同一批模块实例 +# （否则 configure_writer 保存的默认工厂与被测函数看到的不是同一份状态）。 +_SHARED_PKG = "_ws_m11b2" + + +def _load(name, pkg=None): + pkg = pkg or _SHARED_PKG + if pkg not in sys.modules: + mod = types.ModuleType(pkg) + mod.__path__ = [PKG_DIR] + sys.modules[pkg] = mod + full = "%s.%s" % (pkg, name) + if full in sys.modules: + return sys.modules[full] + spec = importlib.util.spec_from_file_location( + full, os.path.join(PKG_DIR, name + ".py")) + mod = importlib.util.module_from_spec(spec) + sys.modules[full] = mod + spec.loader.exec_module(mod) + return mod + + +tx = _load("pbl_runtime_tx") +ps = _load("pbl_runtime_sql") + +# --- 旧名字 -> 真源实现 ------------------------------------------------- +SQLITE = ps.DIALECT_SQLITE +MYSQL = ps.DIALECT_MYSQL +percentile = ps.percentile + +_write_event_with_state = tx.write_event_with_state +_read_entity_state = tx.read_entity_state + +# 测试进程内的「默认连接工厂」（仅服务于旧 configure_writer 语义） +_default = {"factory": None, "dialect": None, "close_connection": True} + + +def configure_writer(conn_factory=None, dialect=None, close_connection=True, + **_ignored): + """旧模块级默认 writer 语义的测试内等价物：登记默认连接工厂。 + + 真源已无全局 writer 单例（见 ``world_sync/init.py:356`` 的 + ``configure_runtime_writer`` —— 它登记的是 ServerEnv 契约，不是类实例）。 + 本函数只在测试进程内记一个 factory，供同名兼容包装使用。 + """ + _default["factory"] = conn_factory + _default["dialect"] = dialect + _default["close_connection"] = close_connection + return conn_factory + + +def write_event_with_state(event_data=None, state_update=None, conn_factory=None, + conn=None, event_uid=None, source=None, clock=None): + """兼容包装：未显式给 conn/conn_factory 时取 ``configure_writer`` 登记的工厂。""" + if conn is None and conn_factory is None: + conn_factory = _default["factory"] + res = _write_event_with_state(event_data, state_update, + conn_factory=conn_factory, conn=conn, + event_uid=event_uid, source=source, clock=clock) + return _augment(res) + + +def _augment(result): + """补旧断言用到的键，不改动真源已有键的语义。""" + if isinstance(result, dict): + action = result.get("action") + result.setdefault("state_inserted", action == "inserted") + result.setdefault("state_updated", action == "updated") + result.setdefault("event_only", action == "event_only") + return result + + +class RuntimeStateEventWriter(object): + """旧类式写入器的测试适配：持有 conn_factory/dialect/close_connection， + 方法体直接调用真源函数式 API（同一条代码路径，不复制业务逻辑）。 + """ + + def __init__(self, conn_factory=None, dialect=None, close_connection=True, + source=None, clock=None, hooks=None, **extra): + self.conn_factory = conn_factory + self.dialect = dialect + self.close_connection = close_connection + self.source = source + self.clock = clock + self.hooks = hooks or {} + self._extra = dict(extra) + self.calls = [] + + def write_event_with_state(self, event_data=None, state_update=None, **kw): + conn = kw.pop("conn", None) + event_uid = kw.pop("event_uid", self._extra.get("event_uid")) + source = kw.pop("source", self.source) + clock = kw.pop("clock", self.clock) + conn_factory = kw.pop("conn_factory", None) or self.conn_factory + pre = self.hooks.get("before_write") + if callable(pre): + pre(event_data, state_update) + if conn is None and conn_factory is None: + conn_factory = _default["factory"] + post = self.hooks.get("after_write") + result = _write_event_with_state(event_data, state_update, + conn_factory=conn_factory, conn=conn, + event_uid=event_uid, source=source, + clock=clock) + self.calls.append(result.get("event_uid")) + if callable(post): + post(result) + return _augment(result) + + # 旧代码里偶有以 read_state 名称取当前状态的用法 + def read_entity_state(self, tenant_id=None, session_id=None, entity_id=None, + world_id=None): + res = _read_entity_state(conn_factory=self.conn_factory, + tenant_id=tenant_id, session_id=session_id, + entity_id=entity_id, world_id=world_id) + return res + + +def build_plan(event_data=None, state_update=None, event_uid=None, source=None, + now=None): + """旧 build_plan 名字 -> 真源 init.build_runtime_write_plan（不落库预演）。""" + init_mod = _load("init", pkg=_PKG) + return init_mod.build_runtime_write_plan( + event_data, state_update, event_uid=event_uid, source=source, now=now)