# -*- coding: utf-8 -*-
"""M11b-2 单事务写入 —— 可执行单元测试（sqlite 文件库真实跑通，非 import 面检查）。

覆盖需求点与验收标准：
- 正常写入：事件 + 状态在同一事务落库，两表各 1 行（TestAtomicWriteSuccess）；
- 事件插入失败 → 状态更新一并回滚（TestRollbackOnEventFailure）；
- 状态更新失败/并发冲突 → 整体回滚，事件不留（TestRollbackOnStateFailure）；
- 并发：多事务写同一实体，不出现脏写/丢失更新（TestConcurrencyNoLostUpdate）；
- 租户/世界隔离：跨租户同名 entity_id 互不影响（TestTenantWorldIsolation）；
- P99 ≤ 200ms（TestLatencyP99，percentile 与 m11b-selftest-run.log 同口径）；
- conn_factory 三级优先级 + fail-closed + 禁读 env/*.json（TestConnFactoryResolution）；
- 接线面：导出一致性、禁项（无广播/轮询）扫描（TestPackageExports / TestNoForbiddenLogic）。

运行：cd apps/scense/pkgs/world_sync && python -m pytest tests/test_m11b2_wiring.py -v
"""

import os
import sqlite3
import sys
import threading

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from world_sync import pbl_runtime_sql as ps  # noqa: E402
from world_sync.pbl_runtime_errors import (  # noqa: E402
    ConcurrentStateConflict,
    ConnFactoryNotConfigured,
    EventWriteError,
    PblRuntimeError,
    StateWriteError,
    TxConfigError,
)
from world_sync.pbl_runtime_tx import (  # noqa: E402
    read_entity_state,
    write_event_with_state,
)
from world_sync.pbl_runtime_tx_env import (  # noqa: E402
    build_sqlite_conn_factory,
    resolve_conn_factory,
)

percentile = ps.percentile


# --------------------------------------------------------------------------
# fixtures / helpers
# --------------------------------------------------------------------------

@pytest.fixture
def db_path(tmp_path):
    """文件型 sqlite 库（多线程共享同一库，比 :memory: 更适合并发用例）。"""
    return str(tmp_path / "m11b2.db")


@pytest.fixture
def factory(db_path):
    """sqlite 连接工厂，并保证两表 DDL 已建（CREATE TABLE IF NOT EXISTS 幂等）。"""
    f = build_sqlite_conn_factory(db_path)
    c = f()
    for stmt in ps.ddl_for(ps.DIALECT_SQLITE):
        c.execute(stmt)
    c.commit()
    c.close()
    return f


@pytest.fixture
def conn(factory):
    c = factory()
    yield c
    c.close()


def _event(tenant="t1", world="w1", entity="e1", **kw):
    base = {
        "tenant_id": tenant,
        "world_id": world,
        "entity_id": entity,
        "event_type": "state.updated",
        "payload": {"hp": 100},
        "session_id": "s-1",
        "source": "unit-test",
    }
    base.update(kw)
    return base


def _state(tenant="t1", world="w1", entity="e1", **kw):
    base = {"tenant_id": tenant, "world_id": world, "entity_id": entity,
            "state": {"hp": 100, "x": 1}}
    base.update(kw)
    return base


def _counts(conn):
    ev = conn.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0]
    st = conn.execute("SELECT COUNT(*) FROM pbl_entity_state").fetchone()[0]
    return ev, st


# --------------------------------------------------------------------------
# 1. 正常写入：单事务内事件 + 状态各 1 行
# --------------------------------------------------------------------------

class TestAtomicWriteSuccess(object):

    def test_first_write_inserts_state_and_event(self, conn, factory):
        result = write_event_with_state(_event(), _state(state={"hp": 90}),
                                        conn_factory=factory)
        assert result["action"] == "inserted"
        assert result["state_version"] == 1
        assert result["dialect"] == "sqlite"
        assert _counts(conn) == (1, 1)

        row = conn.execute(
            "SELECT event_id, entity_id, state_version, payload, created_at "
            "FROM pbl_runtime_event"
        ).fetchone()
        assert row[1] == "e1"
        assert row[2] == 1
        assert '"hp":100' in row[3]          # payload 存事件体
        assert row[4].endswith("Z")
        st = conn.execute(
            "SELECT state, state_version, updated_by_event FROM pbl_entity_state"
        ).fetchone()
        assert '"hp":90' in st[0]            # 状态存新值
        assert st[1] == 1
        assert st[2] == row[0]               # 状态行回指生效事件

    def test_second_write_bumps_version(self, conn, factory):
        write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
        r2 = write_event_with_state(_event(), _state(state={"hp": 70}), conn_factory=factory)
        assert r2["action"] == "updated"
        assert r2["state_version"] == 2
        assert _counts(conn) == (2, 1)

    def test_event_only_write_keeps_state(self, conn, factory):
        write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
        r = write_event_with_state(_event(event_type="chat"), None, conn_factory=factory)
        assert r["action"] == "event_only"
        assert _counts(conn) == (2, 1)

    def test_read_entity_state_helper(self, factory):
        write_event_with_state(_event(), _state(state={"hp": 42}), conn_factory=factory)
        got = read_entity_state(conn_factory=factory, tenant_id="t1",
                                world_id="w1", entity_id="e1")
        assert got["state_version"] == 1
        assert '"hp":42' in got["state"]
        assert read_entity_state(conn_factory=factory, tenant_id="t1",
                                 world_id="w1", entity_id="nope") is None

    def test_missing_tenant_rejected_before_touching_db(self, factory):
        with pytest.raises(TxConfigError) as ei:
            write_event_with_state({"world_id": "w1", "entity_id": "e1"},
                                   _state(), conn_factory=factory)
        assert ei.value.code == "PBL_TX_CONFIG_ERROR"
        assert ei.value.detail.get("field") == "tenant_id"
        assert _counts(factory()) == (0, 0)

    def test_conflicting_scope_rejected(self, factory):
        with pytest.raises(TxConfigError):
            write_event_with_state(_event(tenant="t1"),
                                   _state(tenant="t2"), conn_factory=factory)

    def test_result_is_serializable(self, factory):
        import json
        r = write_event_with_state(_event(), _state(), conn_factory=factory)
        json.dumps(r)  # 不抛即通过


# --------------------------------------------------------------------------
# 2. 失败整体回滚：事件插入失败 → 状态变更消失
# --------------------------------------------------------------------------

class TestRollbackOnEventFailure(object):

    def test_event_failure_rolls_back_state_update(self, conn, factory):
        """状态先更新成功，事件插入被触发器打坏 → 断言两表零增量、状态仍是旧值。"""
        write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
        conn.execute(
            "CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event "
            "BEGIN SELECT RAISE(ABORT, 'boom: event insert failed'); END"
        )

        with pytest.raises(EventWriteError) as ei:
            write_event_with_state(_event(), _state(state={"hp": 1}), conn_factory=factory)
        assert ei.value.code == "PBL_RUNTIME_EVENT_WRITE_FAILED"
        assert "boom" in str(ei.value)

        assert _counts(conn) == (1, 1)
        st = conn.execute("SELECT state, state_version FROM pbl_entity_state").fetchone()
        assert '"hp":90' in st[0]
        assert st[1] == 1

    def test_duplicate_event_id_rolls_back_state(self, conn, factory):
        """同 event_id 主键冲突 → 事件失败 → 状态更新回滚。"""
        write_event_with_state(_event(), _state(state={"hp": 90}),
                               conn_factory=factory, event_id="dup-1")
        with pytest.raises(EventWriteError):
            write_event_with_state(_event(), _state(state={"hp": 5}),
                                   conn_factory=factory, event_id="dup-1")
        assert _counts(conn) == (1, 1)
        st = conn.execute("SELECT state_version FROM pbl_entity_state").fetchone()
        assert st[0] == 1


# --------------------------------------------------------------------------
# 3. 失败整体回滚：状态更新失败 / 并发冲突
# --------------------------------------------------------------------------

class TestRollbackOnStateFailure(object):

    def test_expected_version_conflict_rolls_back_both(self, conn, factory):
        write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)

        with pytest.raises(ConcurrentStateConflict) as ei:
            write_event_with_state(
                _event(), _state(state={"hp": 0}, expected_version=99),
                conn_factory=factory)
        assert ei.value.code == "PBL_ENTITY_STATE_CONFLICT"
        assert ei.value.detail["expected_version"] == 99
        assert ei.value.detail["actual_version"] == 1
        assert _counts(conn) == (1, 1)      # 冲突时事件一条都不能多留

    def test_state_trigger_failure_rolls_back_event(self, conn, factory):
        """状态 UPDATE 失败（触发器报错）→ 本事务的事件也不能落库。"""
        write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
        conn.execute(
            "CREATE TRIGGER fail_state BEFORE UPDATE ON pbl_entity_state "
            "BEGIN SELECT RAISE(ABORT, 'boom: state update failed'); END"
        )
        with pytest.raises(StateWriteError):
            write_event_with_state(_event(), _state(state={"hp": 3}), conn_factory=factory)
        # 事件仍只有首条（失败事务的事件被回滚），状态版本仍为 1
        assert _counts(conn) == (1, 1)
        assert conn.execute(
            "SELECT state_version FROM pbl_entity_state").fetchone()[0] == 1

    def test_failed_write_leaves_no_open_transaction(self, factory):
        """失败后连接必须已回滚，后续写入不受影响（不吞错误、不留半截事务）。"""
        c = factory()
        c.execute(
            "CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event "
            "BEGIN SELECT RAISE(ABORT, 'boom'); END"
        )
        with pytest.raises(EventWriteError):
            write_event_with_state(_event(), _state(), conn=c)
        c.execute("DROP TRIGGER fail_event")
        r = write_event_with_state(_event(), _state(), conn=c)
        assert r["state_version"] == 1
        assert c.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0] == 1
        assert c.execute("SELECT COUNT(*) FROM pbl_entity_state").fetchone()[0] == 1
        c.close()


# --------------------------------------------------------------------------
# 4. 并发：不出现脏写 / 丢失更新
# --------------------------------------------------------------------------

class TestConcurrencyNoLostUpdate(object):

    def test_concurrent_writers_no_lost_update(self, factory):
        """8 线程同时写同一实体：成功版本号连续无空洞，事件数 == 成功数。"""
        threads = 8
        results = []
        errors = []
        lock = threading.Lock()
        barrier = threading.Barrier(threads)

        def worker(i):
            barrier.wait()
            try:
                r = write_event_with_state(
                    _event(payload={"writer": i}),
                    _state(state={"hp": 100 + i}),
                    conn_factory=factory,
                )
                with lock:
                    results.append(r["state_version"])
            except PblRuntimeError as exc:
                with lock:
                    errors.append(exc.code)
            except sqlite3.OperationalError as exc:
                # sqlite 文件库锁超时属环境限制，不算脏写
                with lock:
                    errors.append("locked:%s" % str(exc)[:40])

        ts = [threading.Thread(target=worker, args=(i,)) for i in range(threads)]
        for t in ts:
            t.start()
        for t in ts:
            t.join()

        c = factory()
        try:
            n_events = c.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0]
            st = c.execute("SELECT state_version FROM pbl_entity_state").fetchone()
        finally:
            c.close()

        assert len(results) >= 1, "all writers failed: %s" % errors
        assert n_events == len(results)      # 没有「事件写了但状态没推进」的半截事务
        assert st[0] == len(results)         # 版本 == 成功提交数，无丢失更新
        assert sorted(results) == list(range(1, len(results) + 1))  # 连续无空洞
        assert not [e for e in errors if not e.startswith("locked")]

    def test_optimistic_lock_rejects_stale_writer(self, conn, factory):
        """显式 expected_version 的写者：版本被别人推进后必须被拒绝（不覆盖）。"""
        write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
        write_event_with_state(_event(), _state(state={"hp": 80}), conn_factory=factory)
        with pytest.raises(ConcurrentStateConflict):
            write_event_with_state(_event(), _state(state={"hp": 70}, expected_version=1),
                                   conn_factory=factory)
        assert _counts(conn) == (2, 1)
        st = conn.execute("SELECT state FROM pbl_entity_state").fetchone()
        assert '"hp":80' in st[0]  # 迟到的旧版本写者没有覆盖新状态


# --------------------------------------------------------------------------
# 5. 租户 / 世界隔离
# --------------------------------------------------------------------------

class TestTenantWorldIsolation(object):

    def test_same_entity_id_across_tenants_isolated(self, conn, factory):
        write_event_with_state(_event(tenant="t1"), _state(tenant="t1", state={"hp": 1}),
                               conn_factory=factory)
        write_event_with_state(_event(tenant="t2"), _state(tenant="t2", state={"hp": 2}),
                               conn_factory=factory)
        assert _counts(conn) == (2, 2)
        r = conn.execute(
            "SELECT tenant_id, state_version FROM pbl_entity_state ORDER BY tenant_id"
        ).fetchall()
        assert r == [("t1", 1), ("t2", 1)]  # 各自独立版本，互不覆盖

    def test_same_entity_id_across_worlds_isolated(self, conn, factory):
        write_event_with_state(_event(world="w1"), _state(world="w1"), conn_factory=factory)
        write_event_with_state(_event(world="w2"), _state(world="w2"), conn_factory=factory)
        assert _counts(conn) == (2, 2)

    def test_event_query_scoped_by_tenant(self, conn, factory):
        write_event_with_state(_event(tenant="t1"), _state(tenant="t1"), conn_factory=factory)
        write_event_with_state(_event(tenant="t2"), _state(tenant="t2"), conn_factory=factory)
        n = conn.execute(
            "SELECT COUNT(*) FROM pbl_runtime_event WHERE tenant_id = ? AND world_id = ?",
            ("t1", "w1")).fetchone()[0]
        assert n == 1


# --------------------------------------------------------------------------
# 6. 时延：P99 ≤ 200ms（不含广播）
# --------------------------------------------------------------------------

class TestLatencyP99(object):

    def test_percentile_linear_interpolation(self):
        assert percentile([], 99) == 0.0
        assert percentile([5], 99) == 5.0
        assert percentile([1, 2, 3, 4], 50) == 2.5
        assert abs(percentile(list(range(1, 101)), 99) - 99.01) < 1e-9
        assert percentile([10, 20, 30], 0) == 10
        assert percentile([10, 20, 30], 100) == 30

    def test_single_tx_p99_under_200ms(self, factory):
        """100 次单事务写入（事件+状态），P99 ≤ 200ms（sqlite 本地口径）。"""
        samples = []
        for i in range(100):
            r = write_event_with_state(
                _event(entity="perf-%d" % (i % 10), payload={"i": i}),
                _state(entity="perf-%d" % (i % 10), state={"i": i}),
                conn_factory=factory,
            )
            samples.append(r["elapsed_ms"])
        p99 = percentile(samples, 99)
        assert p99 <= 200.0, "P99=%.3fms exceeds 200ms budget" % p99
        assert percentile(samples, 50) <= 200.0


# --------------------------------------------------------------------------
# 7. conn_factory 三级优先级 + fail-closed + 禁读 env/*.json
# --------------------------------------------------------------------------

class TestConnFactoryResolution(object):

    def test_explicit_factory_wins(self, tmp_path, monkeypatch):
        path = str(tmp_path / "explicit.db")
        called = []
        real = build_sqlite_conn_factory(path)

        def counting():
            called.append(1)
            return real()

        monkeypatch.setenv("PBL_RUNTIME_DB_URL", "sqlite:///%s" % (tmp_path / "env.db"))
        f = resolve_conn_factory(counting)
        c = f()
        c.execute("SELECT 1").fetchone()
        c.close()
        assert called == [1]

    def test_env_url_used_when_no_injection(self, tmp_path, monkeypatch):
        path = str(tmp_path / "from-env.db")
        monkeypatch.setenv("PBL_RUNTIME_DB_URL", "sqlite:///%s" % path)
        f = resolve_conn_factory(None)
        c = f()
        assert c is not None
        c.close()
        assert os.path.isfile(path)

    def test_db_url_string_injection(self, tmp_path, monkeypatch):
        path = str(tmp_path / "by-string.db")
        f = resolve_conn_factory("sqlite:///%s" % path)
        c = f()
        c.close()
        assert os.path.isfile(path)

    def test_fail_closed_when_nothing_configured(self, monkeypatch):
        import world_sync.pbl_runtime_tx_env as txenv
        monkeypatch.delenv("PBL_RUNTIME_DB_URL", raising=False)
        monkeypatch.delenv("WORLD_SYNC_DB_URL", raising=False)
        # 屏蔽模块自有 conf/db.json，模拟「三级全空」
        monkeypatch.setattr(txenv, "load_db_config", lambda conf_dir=None: None)
        with pytest.raises(ConnFactoryNotConfigured) as ei:
            resolve_conn_factory(None)
        assert ei.value.code == "PBL_TX_CONN_FACTORY_MISSING"
        assert "forbidden" in str(ei.value)
        assert ei.value.detail["tried"] == [
            "explicit", "server_env", "env_vars", "module_conf/db.json"]

    def test_plaintext_password_in_module_conf_rejected(self, monkeypatch):
        import world_sync.pbl_runtime_tx_env as txenv
        monkeypatch.setattr(
            txenv, "load_db_config",
            lambda conf_dir=None: {"dialect": "mysql", "host": "h",
                                   "user": "u", "database": "d",
                                   "password": "s3cr3t"})
        with pytest.raises(TxConfigError):
            txenv._from_module_conf()

    def test_unsupported_url_scheme_rejected(self):
        import world_sync.pbl_runtime_tx_env as txenv
        with pytest.raises(TxConfigError):
            txenv._parse_db_url("oracle://x/y")

    def test_no_env_test_json_candidate_in_source(self):
        """QC #5 回归守卫：源码里不得再出现 env/test.json 之类的凭据文件候选路径。"""
        root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        pkg = os.path.join(root, "world_sync")
        banned = ('"env/test.json"', '"env/prod.json"', "'env/test.json'",
                  "'env/prod.json'", "projects/pbls/env")
        for name in sorted(os.listdir(pkg)):
            if not name.endswith(".py"):
                continue
            with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh:
                text = fh.read()
            for token in banned:
                assert token not in text, "%s still references %s" % (name, token)


# --------------------------------------------------------------------------
# 8. 方言层单测
# --------------------------------------------------------------------------

class TestDialectLayer(object):

    def test_detect_sqlite(self, conn):
        assert ps.detect_dialect(conn) == ps.DIALECT_SQLITE

    def test_placeholders_per_dialect(self):
        assert ps.placeholder(ps.DIALECT_SQLITE, 0) == "?"
        assert ps.placeholder(ps.DIALECT_MYSQL, 0) == "%s"
        assert ps.placeholder(ps.DIALECT_POSTGRES, 2) == "$3"
        assert ps.render_placeholders(ps.DIALECT_MYSQL, 3) == "%s, %s, %s"

    def test_for_update_only_on_row_lock_dbs(self):
        assert ps.for_update_clause(ps.DIALECT_SQLITE) == ""
        assert ps.for_update_clause(ps.DIALECT_MYSQL).strip() == "FOR UPDATE"
        assert ps.for_update_clause(ps.DIALECT_POSTGRES).strip() == "FOR UPDATE"

    def test_read_committed_stmt(self):
        assert "READ COMMITTED" in ps.read_committed_stmt(ps.DIALECT_MYSQL)
        assert "READ COMMITTED" in ps.read_committed_stmt(ps.DIALECT_POSTGRES)
        assert ps.read_committed_stmt(ps.DIALECT_SQLITE) is None

    def test_begin_stmt(self):
        assert ps.begin_stmt(ps.DIALECT_SQLITE) == "BEGIN IMMEDIATE"
        assert ps.begin_stmt(ps.DIALECT_MYSQL) is None

    def test_unknown_dialect_raises(self):
        class Weird(object):
            pass
        with pytest.raises(ValueError):
            ps.detect_dialect(Weird())

    def test_ddl_covers_both_tables(self):
        text = " ".join(ps.ddl_for(ps.DIALECT_SQLITE))
        assert ps.EVENT_TABLE in text and ps.STATE_TABLE in text
        assert "PRIMARY KEY (tenant_id, world_id, entity_id)" in text
        assert "ENGINE=InnoDB" in " ".join(ps.ddl_for(ps.DIALECT_MYSQL))

    def test_no_ddl_for_unknown_dialect(self):
        with pytest.raises(ValueError):
            ps.ddl_for("oracle")


# --------------------------------------------------------------------------
# 9. 接线面 / 禁项扫描
# --------------------------------------------------------------------------

class TestPackageExports(object):

    def test_all_is_importable(self):
        import world_sync as ws
        for name in ws.__all__:
            assert hasattr(ws, name), "missing export: %s" % name

    def test_public_api_surface(self):
        import world_sync as ws
        assert callable(ws.write_event_with_state)
        assert callable(ws.read_entity_state)
        assert "ConcurrentStateConflict" in ws.__all__
        assert "EventWriteError" in ws.__all__

    def test_sqlor_layer_delegates_to_tx_core(self):
        from world_sync import pbl_runtime_tx, pbl_runtime_tx_sqlor
        assert callable(pbl_runtime_tx_sqlor.write_event_with_state)
        assert pbl_runtime_tx_sqlor._write_event_with_state is \
            pbl_runtime_tx.write_event_with_state


class TestNoForbiddenLogic(object):
    """验收标准「不引入广播逻辑、不引入轮询逻辑」的机械守卫。"""

    FORBIDDEN = ("broadcast", "publish_to", "subscribe", "poll(", "polling",
                 "while True:", "time.sleep(", "websocket", "event_source")

    TARGETS = ("pbl_runtime_tx.py", "pbl_runtime_tx_sqlor.py",
               "pbl_runtime_sql.py", "pbl_runtime_errors.py",
               "pbl_runtime_tx_env.py")

    def test_no_broadcast_or_polling_in_m11b2_modules(self):
        root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        pkg = os.path.join(root, "world_sync")
        for name in self.TARGETS:
            with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh:
                lines = fh.readlines()
            for idx, ln in enumerate(lines, 1):
                low = ln.lower()
                for token in self.FORBIDDEN:
                    if token in low and not low.lstrip().startswith("#"):
                        pytest.fail("%s:%d contains forbidden %r" % (name, idx, token))

    def test_no_network_imports(self):
        root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        pkg = os.path.join(root, "world_sync")
        for name in self.TARGETS:
            with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh:
                text = fh.read()
            for mod in ("import asyncio", "import socket", "import requests",
                        "import aiohttp"):
                assert mod not in text, "%s %s" % (name, mod)