# -*- 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, "探针未指明真实库复测路径"