# -*- coding: utf-8 -*-
"""M11b-2 单元测试：单事务写入（事件 + 实体状态）原子性 / 并发 / 时延。

运行方式（无需 appPublic/sqlor/ahserver，纯标准库 + sqlite3）::

    cd apps/scense/pkgs/world_sync && python3 -m pytest tests -q
    或 python3 tests/test_m11b2_single_tx.py          # 无 pytest 时自检

被测代码：world_sync/pbl_runtime_tx.py（事务与原子性）、
          world_sync/pbl_runtime_sql.py（列契约与 SQL 构造）、
          world_sync/pbl_runtime_errors.py（异常族）。

方言说明：生产为 MySQL（READ COMMITTED + SELECT ... FOR UPDATE）；测试用 sqlite
（SQLITE 方言：锁子句为空串，BEGIN IMMEDIATE 提供写串行化），因此并发正确性在
测试里主要由「乐观锁版本护栏」证明——这与 MySQL 下的第二重保险完全一致。
"""

import importlib.util
import os
import sqlite3
import sys
import threading
import time
import types
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
PKG_DIR = os.path.join(os.path.dirname(HERE), "world_sync")
_PKG = "_ws_m11b2"


def _load(name):
    """以合成包方式加载被测模块（绕开 world_sync/__init__.py 的运行时依赖）。"""
    if _PKG not in sys.modules:
        pkg = types.ModuleType(_PKG)
        pkg.__path__ = [PKG_DIR]
        sys.modules[_PKG] = pkg
    full = "%s.%s" % (_PKG, name)
    if full in sys.modules:
        return sys.modules[full]
    spec = importlib.util.spec_from_file_location(full, os.path.join(PKG_DIR, name + ".py"))
    mod = importlib.util.module_from_spec(spec)
    sys.modules[full] = mod
    spec.loader.exec_module(mod)
    return mod


tx = _load("pbl_runtime_tx")
sqlmod = _load("pbl_runtime_sql")
errs = _load("pbl_runtime_errors")

RuntimeStateEventWriter = tx.RuntimeStateEventWriter
write_event_with_state = tx.write_event_with_state
percentile = tx.percentile
build_plan = tx.build_plan
SQLITE = sqlmod.SQLITE
MYSQL = sqlmod.MYSQL
RuntimeWriteError = errs.RuntimeWriteError
InvalidWriteRequest = errs.InvalidWriteRequest
EventWriteError = errs.EventWriteError
StateWriteError = errs.StateWriteError
ConcurrentStateConflict = errs.ConcurrentStateConflict
TransactionAborted = errs.TransactionAborted

DDL_EVENT = """
CREATE TABLE IF NOT EXISTS pbl_runtime_event (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  tenant_id TEXT NOT NULL,
  event_uid TEXT NOT NULL,
  event_code TEXT,
  idem_key TEXT,
  world_id TEXT,
  scene_id TEXT,
  session_id TEXT,
  entity_id TEXT NOT NULL,
  actor_id TEXT,
  event_type TEXT NOT NULL,
  payload TEXT,
  payload_json TEXT,
  causation_id TEXT,
  seq INTEGER,
  seq_no INTEGER,
  source TEXT,
  state TEXT,
  state_version INTEGER,
  tx_group TEXT,
  broadcast TEXT,
  created_by TEXT,
  created_at TEXT NOT NULL,
  updated_at TEXT,
  occurred_at TEXT,
  UNIQUE (tenant_id, event_uid)
)
"""

DDL_STATE = """
CREATE TABLE IF NOT EXISTS pbl_entity_state (
  id TEXT,
  tenant_id TEXT NOT NULL,
  session_id TEXT NOT NULL,
  entity_id TEXT NOT NULL,
  state_json TEXT,
  state_version INTEGER NOT NULL DEFAULT 0,
  checksum TEXT,
  updated_by TEXT,
  created_at TEXT,
  updated_at TEXT,
  PRIMARY KEY (tenant_id, session_id, entity_id)
)
"""


def make_db(path=":memory:"):
    conn = sqlite3.connect(path, timeout=15.0)
    conn.isolation_level = None          # 由写入器显式 BEGIN/COMMIT/ROLLBACK
    conn.execute(DDL_EVENT)
    conn.execute(DDL_STATE)
    return conn


def seed_state(conn, tenant="t1", session="s1", entity="e1", state_json='{"hp":1}',
               version=0, checksum=None, updated_by=None):
    conn.execute("DELETE FROM pbl_entity_state WHERE tenant_id=? AND session_id=? AND entity_id=?",
                 (tenant, session, entity))
    conn.execute(
        "INSERT INTO pbl_entity_state (id,tenant_id,session_id,entity_id,state_json,"
        "state_version,checksum,updated_by,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?)",
        ("row-" + entity, tenant, session, entity, state_json, version, checksum,
         updated_by, "2026-01-01 00:00:00", "2026-01-01 00:00:00"))


def event(uid="ev-1", tenant="t1", session="s1", entity="e1", **kw):
    data = {"tenant_id": tenant, "event_uid": uid, "session_id": session,
            "entity_id": entity, "event_type": "unit.move", "world_id": "w1",
            "scene_id": "sc1", "actor_id": "u1", "payload": {"dx": 1},
            "seq_no": 7, "seq": 7}
    data.update(kw)
    return data


def state_update(**kw):
    data = {"session_id": "s1", "entity_id": "e1", "state_json": {"hp": 2},
            "updated_by": "u1"}
    data.update(kw)
    return data


class SingleConnProxy(object):
    """包装连接，可对指定阶段的 SQL 注入失败（验证原子回滚）。"""

    def __init__(self, conn, fail_when):
        self._conn = conn
        self._fail_when = fail_when

    def cursor(self):
        return _ProxyCursor(self._conn.cursor(), self._fail_when)

    def commit(self):
        self._conn.commit()

    def rollback(self):
        self._conn.rollback()

    def close(self):
        pass


class _ProxyCursor(object):
    def __init__(self, cur, fail_when):
        self._cur = cur
        self._fail_when = fail_when

    @property
    def rowcount(self):
        return self._cur.rowcount

    def execute(self, sql, params=None):
        if self._fail_when(sql):
            raise sqlite3.IntegrityError("injected failure: %s" % " ".join(sql.split())[:60])
        return self._cur.execute(sql, params or [])

    def fetchone(self):
        return self._cur.fetchone()

    def close(self):
        self._cur.close()


def is_event_insert(sql):
    head = " ".join(sql.split()).upper()
    return head.startswith("INSERT INTO PBL_RUNTIME_EVENT")


# --------------------------------------------------------------------------
# 1. 正常写入：单事务内事件 + 状态同时落库
# --------------------------------------------------------------------------

class TestNormalWrite(unittest.TestCase):

    def setUp(self):
        self.conn = make_db()
        seed_state(self.conn, version=3, state_json='{"hp":10}')
        self.writer = RuntimeStateEventWriter(lambda: self.conn, dialect=SQLITE, close_connection=False)

    def tearDown(self):
        self.conn.close()

    def test_writes_both_rows_in_one_transaction(self):
        res = self.writer.write_event_with_state(event("ev-ok"), state_update())
        self.assertEqual("ev-ok", res["event_uid"])
        self.assertEqual(4, res["state_version"])          # 3 -> 4
        self.assertEqual(3, res["expected_version"])
        self.assertFalse(res["state_inserted"])
        # 事务内语句清单：幂等预检 + 状态锁读 + 状态更新 + 事件插入（无广播/轮询语句）
        self.assertEqual(4, res["statement_count"])
        joined = " ; ".join(res["statements"]).upper()
        self.assertNotIn("NOTIFY", joined)
        self.assertNotIn("SLEEP", joined)
        rows = self.conn.execute("SELECT count(*) FROM pbl_runtime_event").fetchone()[0]
        self.assertEqual(1, rows)
        st = self.conn.execute(
            "SELECT state_json, state_version, updated_by FROM pbl_entity_state").fetchone()
        self.assertEqual('{"hp":2}', st[0])
        self.assertEqual(4, st[1])
        self.assertEqual("u1", st[2])

    def test_event_row_is_append_only_and_carries_aligned_version(self):
        self.writer.write_event_with_state(event("ev-a"), state_update())
        row = self.conn.execute(
            "SELECT event_uid, state_version, entity_id, session_id, seq_no, source, "
            "created_at FROM pbl_runtime_event").fetchone()
        self.assertEqual("ev-a", row[0])
        self.assertEqual(4, row[1])                        # 状态 3 -> 4，事件版本对齐
        self.assertEqual("e1", row[2])
        self.assertEqual("s1", row[3])
        self.assertEqual(7, row[4])
        self.assertEqual("runtime", row[5])
        self.assertTrue(row[6])                            # 分区键非空

    def test_state_row_inserted_when_absent(self):
        self.conn.execute("DELETE FROM pbl_entity_state")
        res = self.writer.write_event_with_state(event("ev-new"), state_update())
        self.assertTrue(res["state_inserted"])
        self.assertEqual(1, res["state_version"])
        self.assertEqual(1, self.conn.execute(
            "SELECT count(*) FROM pbl_entity_state").fetchone()[0])

    def test_module_level_helper_uses_configured_writer(self):
        tx.configure_writer(lambda: self.conn, dialect=SQLITE, close_connection=False)
        res = write_event_with_state(event("ev-helper"), state_update())
        self.assertEqual(4, res["state_version"])
        self.assertEqual(1, self.conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])

    def test_payload_dict_serialised_to_json_text(self):
        self.writer.write_event_with_state(event("ev-p", payload={"a": 1, "b": [1, 2]}),
                                           state_update())
        payload = self.conn.execute("SELECT payload FROM pbl_runtime_event").fetchone()[0]
        self.assertIn('"a":1', payload)


# --------------------------------------------------------------------------
# 2. 原子性：任一步失败 → 全部回滚 + 明确异常
# --------------------------------------------------------------------------

class TestAtomicRollback(unittest.TestCase):

    def setUp(self):
        self.conn = make_db()
        seed_state(self.conn, version=5, state_json='{"hp":99}')
        self.snapshot = self.conn.execute(
            "SELECT state_json, state_version FROM pbl_entity_state").fetchone()

    def tearDown(self):
        self.conn.close()

    def _writer(self, **kw):
        return RuntimeStateEventWriter(lambda: self.conn, dialect=SQLITE, close_connection=False, **kw)

    def test_event_insert_failure_rolls_back_state_update(self):
        """事件插入失败（中途失败）⇒ 状态更新必须一并回滚。"""
        proxy = SingleConnProxy(self.conn, fail_when=is_event_insert)
        writer = RuntimeStateEventWriter(lambda: proxy, dialect=SQLITE, close_connection=False)
        with self.assertRaises(RuntimeWriteError) as cm:
            writer.write_event_with_state(event("ev-fail"), state_update())
        self.assertIsInstance(cm.exception, EventWriteError)
        self.assertEqual("ev-fail", cm.exception.detail.get("event_uid"))
        # 状态未被改写（回滚生效）
        now = self.conn.execute("SELECT state_json, state_version FROM pbl_entity_state").fetchone()
        self.assertEqual(self.snapshot, now)
        self.assertEqual(0, self.conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])

    def test_before_commit_hook_failure_rolls_back_everything(self):
        def boom(plan):
            raise RuntimeError("boom before commit")
        writer = self._writer(before_commit=boom)
        with self.assertRaises(TransactionAborted):
            writer.write_event_with_state(event("ev-hook"), state_update())
        now = self.conn.execute("SELECT state_json, state_version FROM pbl_entity_state").fetchone()
        self.assertEqual(self.snapshot, now)
        self.assertEqual(0, self.conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])

    def test_state_update_failure_rolls_back_and_is_explicit(self):
        def fail_state(sql):
            head = " ".join(sql.split()).upper()
            return head.startswith("UPDATE PBL_ENTITY_STATE")
        proxy = SingleConnProxy(self.conn, fail_when=fail_state)
        writer = RuntimeStateEventWriter(lambda: proxy, dialect=SQLITE, close_connection=False)
        with self.assertRaises(RuntimeWriteError) as cm:
            writer.write_event_with_state(event("ev-sf"), state_update())
        self.assertIsInstance(cm.exception, StateWriteError)
        self.assertEqual(0, self.conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])
        self.assertEqual(self.snapshot, self.conn.execute(
            "SELECT state_json, state_version FROM pbl_entity_state").fetchone())

    def test_errors_are_never_swallowed(self):
        """任何失败都必须抛 RuntimeWriteError 子类，且 detail 可定位。"""
        writer = self._writer(before_commit=lambda p: (_ for _ in ()).throw(ValueError("x")))
        try:
            writer.write_event_with_state(event("ev-e"), state_update())
        except RuntimeWriteError as exc:
            self.assertIn("reason", exc.detail)
            self.assertIn("event_uid", exc.detail)
            self.assertTrue(str(exc))
        else:
            self.fail("未抛出明确异常，属吞错误")

    def test_duplicate_event_is_rejected_and_state_untouched(self):
        writer = self._writer()
        writer.write_event_with_state(event("ev-dup"), state_update())
        with self.assertRaises(EventWriteError) as cm:
            writer.write_event_with_state(event("ev-dup"), state_update())
        self.assertEqual("DUPLICATE_EVENT", cm.exception.code)
        self.assertEqual(1, self.conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])
        self.assertEqual(6, self.conn.execute(
            "SELECT state_version FROM pbl_entity_state").fetchone()[0])


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

class TestConcurrency(unittest.TestCase):

    def setUp(self):
        self.path = os.path.join(HERE, "_m11b2_concurrent.sqlite")
        if os.path.exists(self.path):
            os.remove(self.path)
        self.conn = make_db(self.path)
        seed_state(self.conn, version=0, state_json='{"hp":0}')
        self.conn.close()

    def tearDown(self):
        if os.path.exists(self.path):
            os.remove(self.path)

    def _factory(self):
        def f():
            return make_db(self.path)
        return f

    def test_stale_expected_version_raises_conflict(self):
        """乐观锁护栏：过期版本写入必须失败且不留痕（防丢失更新）。"""
        conn = make_db(self.path)
        seed_state(conn, version=8, state_json='{"hp":8}')
        writer = RuntimeStateEventWriter(lambda: conn, dialect=SQLITE, close_connection=False)
        with self.assertRaises(ConcurrentStateConflict) as cm:
            writer.write_event_with_state(event("ev-stale"),
                                          state_update(expected_version=7))
        self.assertEqual(7, cm.exception.detail.get("expected_version"))  # 调用方声明的过期基线
        self.assertEqual({"tenant_id": "t1", "session_id": "s1", "entity_id": "e1"},
                         cm.exception.detail.get("key"))
        self.assertEqual(8, conn.execute(
            "SELECT state_version FROM pbl_entity_state").fetchone()[0])
        self.assertEqual(0, conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])
        conn.close()

    def test_two_concurrent_writers_same_entity_no_lost_update(self):
        """两个并发事务写同一实体：全部成功、版本连续递增、事件数与成功数一致。"""
        n_each = 6
        errors = []
        results = []
        lock = threading.Lock()

        def worker(tag):
            conn = make_db(self.path)
            writer = RuntimeStateEventWriter(lambda: conn, dialect=SQLITE, close_connection=False)
            for i in range(n_each):
                try:
                    res = writer.write_event_with_state(
                        event("ev-%s-%d" % (tag, i)), state_update())
                    with lock:
                        results.append(res)
                except RuntimeWriteError as exc:
                    with lock:
                        errors.append(exc)
            conn.close()

        threads = [threading.Thread(target=worker, args=(t,)) for t in ("A", "B")]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        conn = sqlite3.connect(self.path)
        conn.isolation_level = None
        ok = len(results)
        self.assertEqual(0, len(errors), "并发写入不应有未恢复错误: %s" % errors)
        self.assertEqual(2 * n_each, ok)
        self.assertEqual(2 * n_each, conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])
        final = conn.execute("SELECT state_version FROM pbl_entity_state").fetchone()[0]
        self.assertEqual(2 * n_each, final)               # 无丢失更新
        versions = sorted(r["state_version"] for r in results)
        self.assertEqual(list(range(1, 2 * n_each + 1)), versions)  # 版本连续且互不重复 ⇒ 无脏写
        conn.close()

    def test_concurrent_duplicate_event_only_one_wins(self):
        """同一 event_uid 并发提交：只落一条事件，另一次被幂等拒绝。"""
        conn = make_db(self.path)
        outcomes = []
        lock = threading.Lock()

        def worker():
            c = make_db(self.path)
            writer = RuntimeStateEventWriter(lambda: c, dialect=SQLITE)
            try:
                writer.write_event_with_state(event("ev-race"), state_update())
                with lock:
                    outcomes.append("ok")
            except RuntimeWriteError as exc:
                with lock:
                    outcomes.append(type(exc).__name__)
            c.close()

        threads = [threading.Thread(target=worker) for _ in range(2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        self.assertEqual(1, conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])
        self.assertEqual(1, outcomes.count("ok"))
        conn.close()


# --------------------------------------------------------------------------
# 4. 入参校验（fail-closed）
# --------------------------------------------------------------------------

class TestValidation(unittest.TestCase):

    def setUp(self):
        self.conn = make_db()
        self.writer = RuntimeStateEventWriter(lambda: self.conn, dialect=SQLITE, close_connection=False)

    def tearDown(self):
        self.conn.close()

    def test_missing_required_event_fields(self):
        with self.assertRaises(InvalidWriteRequest) as cm:
            self.writer.write_event_with_state({"tenant_id": "t1"}, state_update())
        self.assertIn("event_uid", cm.exception.detail["missing"])
        self.assertEqual(0, self.conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])

    def test_missing_state_json_rejected(self):
        with self.assertRaises(InvalidWriteRequest):
            self.writer.write_event_with_state(event("ev-x"),
                                               {"session_id": "s1", "entity_id": "e1"})

    def test_key_mismatch_between_event_and_state_rejected(self):
        with self.assertRaises(InvalidWriteRequest) as cm:
            self.writer.write_event_with_state(event("ev-y"), state_update(entity_id="other"))
        self.assertEqual("entity_id", cm.exception.detail["field"])

    def test_unserialisable_payload_rejected(self):
        with self.assertRaises(InvalidWriteRequest):
            self.writer.write_event_with_state(event("ev-z", payload=object()), state_update())


# --------------------------------------------------------------------------
# 5. 时延：单次事务（不含广播）P99 ≤ 200ms
# --------------------------------------------------------------------------

class TestLatency(unittest.TestCase):

    def setUp(self):
        self.path = os.path.join(HERE, "_m11b2_latency.sqlite")
        if os.path.exists(self.path):
            os.remove(self.path)
        conn = make_db(self.path)
        seed_state(conn, version=0)
        conn.close()

    def tearDown(self):
        if os.path.exists(self.path):
            os.remove(self.path)

    def test_p99_under_200ms(self):
        samples = []
        conn = make_db(self.path)
        writer = RuntimeStateEventWriter(lambda: conn, dialect=SQLITE, close_connection=False)
        for i in range(100):
            res = writer.write_event_with_state(event("ev-l-%d" % i), state_update())
            samples.append(res["elapsed_ms"])
        conn.close()
        p50 = percentile(samples, 50)
        p99 = percentile(samples, 99)
        print("[M11b-2 latency] n=%d p50=%.3fms p99=%.3fms max=%.3fms"
              % (len(samples), p50, p99, max(samples)))
        self.assertLessEqual(p99, 200.0)

    def test_percentile_helper(self):
        self.assertEqual(None, percentile([], 99))
        self.assertEqual(5, percentile([5], 99))
        p99 = percentile(list(range(100)), 99)
        self.assertGreaterEqual(p99, 98.0)
        self.assertLessEqual(p99, 99.0)


# --------------------------------------------------------------------------
# 6. 范围约束：不引入广播 / 轮询；SQL 构造契约
# --------------------------------------------------------------------------

class TestScopeAndSqlContract(unittest.TestCase):

    def test_no_broadcast_or_polling_in_production_sources(self):
        """范围约束：本模块不得出现广播/轮询/睡眠的可执行调用（AST 级检查，不看散文）。"""
        import ast
        banned_calls = {"sleep", "poll", "long_poll", "publish", "subscribe",
                        "broadcast", "notify", "websocket", "push", "fetch_events"}
        banned_imports = {"asyncio", "socket", "threading", "queue"}
        for name in ("pbl_runtime_tx.py", "pbl_runtime_sql.py", "pbl_runtime_errors.py"):
            path = os.path.join(PKG_DIR, name)
            with open(path, encoding="utf-8") as fh:
                tree = ast.parse(fh.read(), filename=path)
            for node in ast.walk(tree):
                if isinstance(node, (ast.Import, ast.ImportFrom)):
                    mods = ([a.name for a in node.names]
                            if isinstance(node, ast.Import) else [node.module or ""])
                    for mod in mods:
                        root = (mod or "").split(".")[0]
                        self.assertNotIn(root, banned_imports,
                                         "%s 不应导入 %s" % (name, root))
                if isinstance(node, ast.Call):
                    fn = node.func
                    ident = fn.id if isinstance(fn, ast.Name) else (
                        fn.attr if isinstance(fn, ast.Attribute) else "")
                    self.assertNotIn(ident, banned_calls,
                                     "%s 不应调用广播/轮询/睡眠: %s" % (name, ident))

    def test_event_sql_is_insert_only(self):
        cols = list(sqlmod.EventSchema.columns)
        stmt = sqlmod.build_insert_event(MYSQL, ["tenant_id", "event_uid", "created_at"])
        self.assertTrue(stmt.startswith("INSERT INTO pbl_runtime_event"))
        self.assertNotIn("UPDATE", stmt.upper())
        self.assertIn("%s", stmt)
        self.assertEqual(len(cols), len(set(cols)))
        self.assertIn("created_at", cols)                  # 分区键在契约内

    def test_mysql_select_for_update_present(self):
        stmt = sqlmod.build_select_state_for_update(MYSQL)
        self.assertTrue(stmt.endswith("FOR UPDATE"))
        self.assertIn("pbl_entity_state", stmt)
        self.assertNotIn("?", stmt)

    def test_sqlite_lock_clause_degrades_gracefully(self):
        stmt = sqlmod.build_select_state_for_update(SQLITE)
        self.assertNotIn("FOR UPDATE", stmt)

    def test_update_carries_version_guard_and_increment(self):
        stmt, set_cols, count = sqlmod.build_update_state(
            MYSQL, ["state_json", "updated_at"])
        self.assertIn("state_version = state_version + 1", stmt)
        self.assertIn("AND state_version = %s", stmt)
        self.assertEqual(("state_json", "updated_at"), set_cols)
        self.assertEqual(2 + 3 + 1, count)

    def test_plan_assembly_is_side_effect_free(self):
        conn = make_db()
        plan = build_plan(event("ev-plan"), state_update(expected_version=11))
        self.assertEqual(11, plan.expected_version)
        self.assertEqual("ev-plan", plan.event_uid)
        self.assertEqual({"tenant_id": "t1", "session_id": "s1", "entity_id": "e1"}, plan.key)
        self.assertEqual("created_at", sqlmod.EventSchema.partition_key)
        self.assertEqual(0, conn.execute(
            "SELECT count(*) FROM pbl_runtime_event").fetchone()[0])
        conn.close()

    def test_transaction_lifecycle_is_balanced(self):
        """BEGIN / COMMIT 各一次；失败路径必须 ROLLBACK。"""
        conn = make_db()
        seed_state(conn, version=0)
        calls = []

        class Spy(sqlite3.Connection):
            pass

        class Traced(object):
            def __init__(self, real):
                self._real = real

            def cursor(self):
                cur = self._real.cursor()
                outer = self

                class C(object):
                    @property
                    def rowcount(self):
                        return cur.rowcount

                    def execute(self, sql, params=None):
                        head = " ".join(sql.split()).upper()
                        if head.startswith("BEGIN"):
                            calls.append("BEGIN")
                        elif head.startswith("COMMIT"):
                            calls.append("COMMIT")
                        elif head.startswith("ROLLBACK"):
                            calls.append("ROLLBACK")
                        if "INSERT INTO PBL_RUNTIME_EVENT" in head:
                            raise sqlite3.OperationalError("injected")
                        return cur.execute(sql, params or [])

                    def fetchone(self):
                        return cur.fetchone()

                    def close(self):
                        cur.close()
                return C()

            def commit(self):
                self._real.commit()

            def rollback(self):
                calls.append("ROLLBACK")
                self._real.rollback()

            def close(self):
                pass

        traced = Traced(conn)
        writer = RuntimeStateEventWriter(lambda: traced, dialect=SQLITE)
        with self.assertRaises(EventWriteError):
            writer.write_event_with_state(event("ev-trace"), state_update())
        self.assertIn("BEGIN", calls)
        self.assertIn("ROLLBACK", calls)
        self.assertNotIn("COMMIT", calls)
        conn.close()


if __name__ == "__main__":
    unittest.main(verbosity=2)