222 lines
11 KiB
Python
222 lines
11 KiB
Python
# -*- 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）
|
||
T4c 同一探针的 **行为断言**（QC #5 修正）：真实调用 run() 校验返回的 rc 与
|
||
采样计数/百分位元数据，并把「环境受限」如实声明作为对模块元数据（docstring）
|
||
的校验 —— 不再用 open().read() 字符串包含式 grep 凑数。
|
||
|
||
边界：只测不改业务实现、不改 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"])
|
||
rows = [x for x in rows if str(x.get("entity_id")) == "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_runs_and_discloses_environment_limit(p99_probe, capsys):
|
||
"""QC #5 修正：行为断言取代 grep 式字符串包含检查。
|
||
|
||
历史版本本用例只做 `open(probe).read()` 的字符串包含（grep 伪测试），不执行任何
|
||
被测行为 —— 反证是 T4/T4b 全崩（探针 len(rows)==1 断言失败）时 T4c 仍 PASS。
|
||
现在真实调用 p99_probe.run()，断言：
|
||
· 退出码 rc ∈ {0,1}（0=达标，1=超门禁但如实判定；2=采样不足属探针缺陷）；
|
||
· write/read 两项样本数 n ≥ 30，且百分位元数据齐全（非 NaN、非 None）；
|
||
· 探针真实打印了 P99 数值行与结论行（从 capsys 输出中校验，属执行结果校验）；
|
||
· read 行 entity_id 一致性校验为 PASS（QC #3 的防静默放宽门禁落到断言上）。
|
||
「环境受限的诚实声明」改为对模块元数据（docstring）的校验，作为附加项而非唯一断言。
|
||
"""
|
||
capsys.readouterr() # 清掉 T4/T4b 的缓冲输出，只取本次
|
||
rc, rw, rr = p99_probe.run(P99_MIN_SAMPLES)
|
||
out = capsys.readouterr().out
|
||
|
||
# 1) 探针必须真的跑完并给出可判定结论（rc=2 表示采样不足 = 探针自身失效）
|
||
assert rc in (0, 1), "探针退出码异常 rc=%s（2=采样不足，探针未产出有效 P99）" % rc
|
||
# 2) 样本数与百分位元数据
|
||
assert rw["n"] >= P99_MIN_SAMPLES, "write 样本数 %d < %d" % (rw["n"], P99_MIN_SAMPLES)
|
||
assert rr["n"] >= P99_MIN_SAMPLES, "read 样本数 %d < %d" % (rr["n"], P99_MIN_SAMPLES)
|
||
for key in ("min_ms", "avg_ms", "p50_ms", "p95_ms", "p99_ms", "max_ms"):
|
||
assert isinstance(rw.get(key), float) and rw[key] == rw[key], "write %s 缺失/NaN" % key
|
||
assert isinstance(rr.get(key), float) and rr[key] == rr[key], "read %s 缺失/NaN" % key
|
||
assert rw["p99_ms"] <= SLA_P99_MS and rr["p99_ms"] <= SLA_P99_MS, (
|
||
"P99 超门禁：write=%.4fms read=%.4fms" % (rw["p99_ms"], rr["p99_ms"]))
|
||
# 3) 执行输出中确有 P99 数值行与结论行（真实执行的证据，非源码 grep）
|
||
assert "P99=" in out, "探针输出缺失 P99 数值行"
|
||
assert "P99 结论：" in out, "探针输出缺失 P99 门禁结论行"
|
||
assert "entity_id 一致性校验：PASS" in out, (
|
||
"read 返回行 entity_id 一致性校验未通过（QC #3 防静默放宽门禁）")
|
||
# 4) 环境受限的诚实声明（模块元数据校验，附加项）
|
||
doc = p99_probe.__doc__ or ""
|
||
assert ("不含网络 RTT" in doc) or ("真实库" in doc), "探针缺少环境受限的诚实声明"
|
||
assert "--live" in doc, "探针未指明真实库复测路径"
|