diff --git a/scripts/m11b4_selftest.py b/scripts/m11b4_selftest.py index 5244f65..c0a8109 100644 --- a/scripts/m11b4_selftest.py +++ b/scripts/m11b4_selftest.py @@ -310,12 +310,20 @@ def try_mysql_explain(): def test_e2e_latency(dbpath): section("AC3 端到端时延基准:裁决→write_event_with_state→COMMIT→广播→消费端感知") n = 200 + # 写入端连接走模块统一解析链的「环境变量」级(project-directory-spec 8.1: + # 模块禁止读部署 env/{test,prod}.json,凭据只在环境里),不注入私有工厂 + os.environ["PBL_RUNTIME_DB_URL"] = "sqlite:///" + dbpath hooked = install_broadcast_hook(force=True) reset_fallback(factory_for(dbpath)) events, seq = [], 0 def _state_builder(ev): - return None # 只测事件写入链路,不污染状态表 + # 真实链路 = 事件 + 实体状态同事务写入(M11b-2),故给出合法 state_update + return { + "tenant_id": ev["tenant_id"], "session_id": "ss_bench", + "entity_id": ev["entity_id"], "state": {"hp": 100, "i": ev["seq"]}, + "updated_by": "m11b4_bench", + } for i in range(n): seq += 1 @@ -417,6 +425,7 @@ def main(): dbpath = os.path.join(tmpdir, "m11b4.sqlite") conn = make_db(dbpath) conn.close() + os.environ["PBL_RUNTIME_DB_URL"] = "sqlite:///" + dbpath scene, uid, t_created = test_broadcast_lost_recovered_within_3s(dbpath) test_no_duplicate_delivery(dbpath, scene, uid) test_watermark_format_guard(dbpath)