# -*- coding: utf-8 -*- """M11b-2c-A pytest 用例：把既有 selftest / DDL harness 以 pytest 原生断言真实执行。 存在理由（QC #1 硬门禁）：`tests/test_m11b1a_ddl_runtime.py` 是脚本式 harness （main() + sys.exit），无 `test_` 函数 → `pytest tests/` 收集 0 项（退出码 5）， 无法作为「实跑留证」。本文件 **不重写被测逻辑**，而是导入既有真实实现并断言其结果： T1 scripts/pbl_runtime_event_ddl.py 的 A 层离线核验（% 转义 / DDL sha256 基线 / verify_db 判定要素 / CLI 冒烟）—— 复用 tests/test_m11b1a_ddl_runtime.layer_a() T2 scripts/m11b_selftest.py suite_offline()：模块 self_check + 三处同步 （impl/export/register）+ M11a/M11b 注册清单 fail-closed 门禁 T3 scripts/m11b_selftest.py suite_memory()：真实 pbl_runtime_ext.tx_write 主链路 （幂等 / 乐观锁 / 回滚无脏数据 / 提交后广播 / seq 单调 / SLA 打点 / 分区降级） T4 scripts/m11b2c_p99_probe.py：write_event_with_state(→apply_runtime_event) 与 read_entity_state(→read_states) 的 P99 实测（≥30 次采样，门禁 ≤200ms） 边界：只测不改业务实现、不改 models/*.json 表定义。 运行：.venv/bin/python -m pytest tests/ -v --tb=short """ import importlib.util import os import sys import pytest HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) SCRIPTS = os.path.join(ROOT, "scripts") for _p in (ROOT, SCRIPTS): if _p not in sys.path: sys.path.insert(0, _p) SLA_P99_MS = 200.0 P99_MIN_SAMPLES = 30 def _load_by_path(name, path): spec = importlib.util.spec_from_file_location(name, path) mod = importlib.util.module_from_spec(spec) sys.modules[name] = mod spec.loader.exec_module(mod) return mod @pytest.fixture(scope="module") def ddl_harness(): return _load_by_path("m11b2c_ddl_harness", os.path.join(HERE, "test_m11b1a_ddl_runtime.py")) @pytest.fixture(scope="module") def selftest(): return _load_by_path("m11b2c_selftest", os.path.join(SCRIPTS, "m11b_selftest.py")) @pytest.fixture(scope="module") def p99_probe(): return _load_by_path("m11b2c_p99_probe", os.path.join(SCRIPTS, "m11b2c_p99_probe.py")) def _fail_names(results, since): """从 selftest 风格的 RESULTS 切片中取 FAIL 项名。""" return ["%s(%s)" % (n, d) for n, ok, d in results[since:] if not ok] # ------------------------------------------------------------------ T1 DDL 离线层 def test_t1_ddl_offline_harness_all_pass(ddl_harness): """A 层：维护 SQL 的 % 转义、DDL sha256 基线、verify_db 判定要素、CLI 冒烟。""" before = len(ddl_harness.RESULTS) ddl_harness.layer_a() fails = _fail_names(ddl_harness.RESULTS, before) ran = len(ddl_harness.RESULTS) - before print("\n[T1] DDL 离线核验项数=%d, FAIL=%d" % (ran, len(fails))) assert ran >= 15, "A 层核验项异常偏少（%d），疑似未真正执行" % ran assert not fails, "DDL 离线核验存在 FAIL: %s" % " | ".join(fails) def test_t1b_ddl_baseline_sha256_matches(ddl_harness): """A4 的独立断言：DDL 逐字未动（基线哈希来自修复前 commit 049e09a）。""" import hashlib gen = ddl_harness.load_gen() digest = hashlib.sha256((gen.create_table_sql() + "\n").encode("utf-8")).hexdigest() assert digest == ddl_harness.DDL_BASELINE_SHA256, ( "create_table_sql 哈希漂移：本任务禁止改表定义/DDL，实测 %s != 基线 %s" % (digest, ddl_harness.DDL_BASELINE_SHA256)) # ------------------------------------------------- T2 selftest 离线套件（结构门禁） def test_t2_selftest_offline_all_pass(selftest): before = len(selftest.RESULTS) selftest.suite_offline() fails = _fail_names(selftest.RESULTS, before) ran = len(selftest.RESULTS) - before print("\n[T2] selftest 离线断言数=%d, FAIL=%d" % (ran, len(fails))) assert ran >= 20, "离线断言数异常偏少（%d）" % ran assert not fails, "selftest 离线套件 FAIL: %s" % " | ".join(fails) def test_t2b_three_place_sync_for_m11b_contracts(selftest): """三处同步铁律：apply_runtime_event/read_states 等 impl+export+register 齐备。""" from pbl_runtime_ext import tx_write as wtx import pbl_runtime_ext as pkg from pbl_runtime_ext import init as init_mod import inspect src = inspect.getsource(init_mod) for fn in ("apply_runtime_event", "poll_events", "read_states", "assert_append_only"): assert hasattr(wtx, fn), "实现层缺失 %s" % fn assert hasattr(pkg, fn), "包出口(__init__)缺失 %s" % fn assert ("env.%s" % fn) in src or fn in src, "init.py 注册缺失 %s" % fn # ------------------------------------- T3 selftest 内存事务引擎端到端（真实主链路） def test_t3_selftest_memory_engine_end_to_end(selftest): before = len(selftest.RESULTS) selftest.suite_memory() fails = _fail_names(selftest.RESULTS, before) ran = len(selftest.RESULTS) - before print("\n[T3] 端到端断言数=%d, FAIL=%d" % (ran, len(fails))) assert ran >= 25, "端到端断言数异常偏少（%d）" % ran assert not fails, "tx_write 主链路端到端 FAIL: %s" % " | ".join(fails) def test_t3b_state_conflict_rolls_back_cleanly(selftest): """乐观锁冲突：抛 PBL-STATE-CONFLICT 且事件/状态/广播三处零污染（独立复验）。""" engine = selftest.FakeEngine() from pbl_runtime_ext import broadcast as bcast from pbl_runtime_ext import rtx_db from pbl_runtime_ext import tx_write as wtx import asyncio bcast.reset_hub() hub = bcast.get_hub() engine.hub = hub selftest.install_fake(engine) async def run(): r1 = await wtx.apply_runtime_event( tenant_id=7, world_id=2, session_id=21, event_type="move", payload={"x": 1}, state_updates=[{"state_key": "pos", "state_value": {"x": 1}}], client_event_id="pytest-ok-1") assert r1.get("ok") and r1.get("dedup") is False n_ev = len(engine.committed_rows("pbl_runtime_event")) cur = hub.latest_cursor() try: await wtx.apply_runtime_event( tenant_id=7, world_id=2, session_id=21, event_type="move", payload={"x": 2}, state_updates=[{"state_key": "pos", "state_value": {"x": 2}}], base_version=99, client_event_id="pytest-conflict-1") raise AssertionError("乐观锁冲突未抛异常") except rtx_db.RtxError as exc: assert exc.code == "PBL-STATE-CONFLICT", exc.code assert len(engine.committed_rows("pbl_runtime_event")) == n_ev assert hub.latest_cursor() == cur rows = await wtx.read_states(tenant_id=7, session_id=21, entity_ids=["pos"]) assert len(rows) == 1 and int(rows[0]["state_version"]) == 1 asyncio.new_event_loop().run_until_complete(run()) # ------------------------------------------------------------------- T4 P99 实测 def test_t4_p99_write_event_with_state_within_sla(p99_probe): rc, rw, _rr = p99_probe.run(P99_MIN_SAMPLES) print("\n[T4] write P99=%.4fms (n=%d, avg=%.4fms, max=%.4fms) rc=%d" % (rw["p99_ms"], rw["n"], rw["avg_ms"], rw["max_ms"], rc)) assert rw["n"] >= P99_MIN_SAMPLES, "采样不足：%d < %d" % (rw["n"], P99_MIN_SAMPLES) assert rw["p99_ms"] <= SLA_P99_MS, ( "write_event_with_state P99=%.4fms 超 200ms 门禁" % rw["p99_ms"]) def test_t4b_p99_read_entity_state_within_sla(p99_probe): _rc, _rw, rr = p99_probe.run(P99_MIN_SAMPLES) print("\n[T4b] read P99=%.4fms (n=%d, avg=%.4fms, max=%.4fms)" % (rr["p99_ms"], rr["n"], rr["avg_ms"], rr["max_ms"])) assert rr["n"] >= P99_MIN_SAMPLES assert rr["p99_ms"] <= SLA_P99_MS, ( "read_entity_state P99=%.4fms 超 200ms 门禁" % rr["p99_ms"]) def test_t4c_probe_declares_environment_limit_honestly(p99_probe, capsys): """探针必须如实声明：内存引擎 P99 不含真实库 RTT/InnoDB 提交开销。""" text = open(os.path.join(SCRIPTS, "m11b2c_p99_probe.py"), encoding="utf-8").read() assert "不含网络 RTT" in text or "真实库" in text, "探针缺少环境受限的诚实声明" assert "--live" in text, "探针未指明真实库复测路径"