#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""M5a 幂等 / C-3 fail-closed 测试（pbl_evidence）。

覆盖两条硬约束：

C-3 fail-closed
    ``pbl_runtime_event``（scense_runtime M11b 落库的事件表）不存在时，
    ``collect_evidence_from_events()`` **必须抛 CollectError 且消息含 'M11b'**，
    绝不允许「静默返回空 stats 当成功」——静默空成功会让 M5b 回放/cron 轮询
    误判采集已完成，证据链永久缺失。

幂等（idempotency）
    去重三元组 ``(tenant_id, source_event_id, evidence_type)`` 同时是
    ``models/pbl_evidence.json`` 的唯一索引 ``uk_ev_dedup`` 列组合，
    也是 ``build_dedup_key()`` 的输入；同一输入重复计算必须得到同一 32 位键。

无 DB 环境
    本文件默认只跑 fail-closed + 纯函数/契约断言（零 DB 依赖，可离线执行）。
    真实双次采集用例由 ``PBL_EVIDENCE_TEST_DB=1`` 开启，否则 skip 并写明原因。

运行：
    cd modules/pbl_evidence && python3 -m pytest tests/test_m5a_idempotency.py -v
"""

import asyncio
import importlib.util
import json
import os
import pathlib
import sys
import types

import pytest

REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent          # modules/pbl_evidence
PKG_DIR = REPO_ROOT / 'pbl_evidence'                                # 包目录
MODELS_DIR = REPO_ROOT / 'models'

# 让「模块仓库未 pip install -e」时也能定位包目录（build.sh 里是 pkgs/ 软链布局）
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))


def _load_collector():
    """隔离加载 ``pbl_evidence.collector``。

    直接 ``import pbl_evidence.collector`` 会先执行包 ``__init__.py``，
    连带拉起 ``pbl_common`` / ``ahserver`` 等宿主依赖；单元测试要能在
    裸环境（无宿主应用、无 DB）里跑，故用 importlib 造一个只带 ``__path__``
    的假包对象，绕过 ``__init__.py``，只加载 collector 及其同包内
    ``db`` / ``evidence_map``（两者仅依赖标准库）。
    """
    pkg_name = 'pbl_evidence'
    if pkg_name not in sys.modules:
        pkg = types.ModuleType(pkg_name)
        pkg.__path__ = [str(PKG_DIR)]
        pkg.__package__ = pkg_name
        sys.modules[pkg_name] = pkg
    full = pkg_name + '.collector'
    if full in sys.modules:
        return sys.modules[full]
    spec = importlib.util.spec_from_file_location(full, str(PKG_DIR / 'collector.py'))
    if spec is None or spec.loader is None:
        raise RuntimeError('cannot locate %s' % (PKG_DIR / 'collector.py',))
    mod = importlib.util.module_from_spec(spec)
    sys.modules[full] = mod
    spec.loader.exec_module(mod)
    return mod


collector = _load_collector()
CollectError = collector.CollectError
collect_evidence_from_events = collector.collect_evidence_from_events
build_dedup_key = None  # 延迟取，见下方 fixture（来自 collector 同包 evidence_map）

evidence_map = sys.modules.get('pbl_evidence.evidence_map')
if evidence_map is None:
    _em_spec = importlib.util.spec_from_file_location(
        'pbl_evidence.evidence_map', str(PKG_DIR / 'evidence_map.py'))
    evidence_map = importlib.util.module_from_spec(_em_spec)
    sys.modules['pbl_evidence.evidence_map'] = evidence_map
    _em_spec.loader.exec_module(evidence_map)
build_dedup_key = evidence_map.build_dedup_key


# ── 工具：证明「fail-closed 之前不碰数据库」────────────────────────────────
class _DbGuard:
    """替换 collector 的 DB 入口，任何调用都会被记录（用于断言零 DB 访问）。"""

    def __init__(self, monkeypatch, allow=False):
        self.calls = []
        self.allow = allow
        monkeypatch.setattr(collector, 'q_all', self._q_all, raising=True)
        monkeypatch.setattr(collector, 'q_one', self._q_one, raising=True)
        monkeypatch.setattr(collector, 'q_exec', self._q_exec, raising=True)
        monkeypatch.setattr(collector, 'table_exists', self._table_exists, raising=True)
        monkeypatch.setattr(collector, 'table_columns', self._table_columns, raising=True)

    async def _boom(self, name):
        self.calls.append(name)
        if not self.allow:
            raise AssertionError(
                'fail-closed 路径不得访问数据库，但调用了 %s' % (name,))
        return []

    async def _q_all(self, sql, params=None):
        await self._boom('q_all')
        return []

    async def _q_one(self, sql, params=None):
        await self._boom('q_one')
        return None

    async def _q_exec(self, sql, params=None):
        await self._boom('q_exec')
        return None

    async def _table_exists(self, table):
        await self._boom('table_exists')
        return False

    async def _table_columns(self, table):
        await self._boom('table_columns')
        return set()


def _patch_event_table(monkeypatch, value):
    """mock ``resolve_event_table()`` 返回指定值（本任务要求返回 None）。"""

    async def _fake():
        return value

    monkeypatch.setattr(collector, 'resolve_event_table', _fake, raising=True)
    return _fake


def _db_enabled():
    """是否有可用测试库（默认无 → 只跑 fail-closed 路径）。"""
    if os.environ.get('PBL_EVIDENCE_TEST_DB', '').strip().lower() not in (
            '1', 'true', 'yes', 'on'):
        return False
    try:
        collector.get_dbname()
    except Exception:
        return False
    return True


# ══════════════════════════════════════════════════════════════════════════
# C-3 fail-closed
# ══════════════════════════════════════════════════════════════════════════
def test_collect_fail_closed_when_runtime_event_table_missing(monkeypatch):
    """resolve_event_table() → None 时必须抛 CollectError 且消息含 'M11b'。"""
    guard = _DbGuard(monkeypatch)
    _patch_event_table(monkeypatch, None)

    with pytest.raises(CollectError) as excinfo:
        asyncio.run(collect_evidence_from_events(tenant_id='t1'))

    msg = str(excinfo.value)
    assert 'M11b' in msg, 'fail-closed 消息必须点明硬依赖 M11b，实际: %s' % (msg,)
    assert 'pbl_runtime_event' in msg, '消息必须给出缺失的事件表名，实际: %s' % (msg,)
    # 事件表都解析不出来，绝不该走到任何查询
    assert guard.calls == [], 'fail-closed 前不应访问 DB: %s' % (guard.calls,)


def test_collect_fail_closed_is_not_silent_empty_success(monkeypatch):
    """反证：不能返回 ``{'scanned': 0, ...}`` 冒充采集成功（静默空成功 = 证据链永久缺失）。"""
    _patch_event_table(monkeypatch, None)
    monkeypatch.setattr(collector, 'q_all',
                        lambda *a, **k: _unexpected_db(), raising=True)
    monkeypatch.setattr(collector, 'table_columns',
                        lambda *a, **k: _unexpected_db(), raising=True)

    result = None
    raised = None
    try:
        result = asyncio.run(collect_evidence_from_events(tenant_id='t1'))
    except CollectError as exc:
        raised = exc

    assert raised is not None, '事件表缺失必须抛 CollectError，不得静默返回'
    assert result is None, '抛错路径不得同时产出 stats（防「空成功」）'


def _unexpected_db():
    raise AssertionError('fail-closed 路径不得访问数据库')


def test_collect_fail_closed_when_tenant_missing(monkeypatch):
    """多租户铁律：tenant_id 缺失先于事件表解析 fail-closed，且完全不碰 DB。"""
    guard = _DbGuard(monkeypatch)
    _patch_event_table(monkeypatch, 'pbl_runtime_event')  # 即便表存在也不该被走到

    with pytest.raises(CollectError) as excinfo:
        asyncio.run(collect_evidence_from_events(tenant_id=''))

    msg = str(excinfo.value)
    assert 'tenant_id' in msg and 'fail-closed' in msg, '实际: %s' % (msg,)
    assert guard.calls == [], '租户缺失时不得访问 DB: %s' % (guard.calls,)

    # None 与纯空白同样拒绝
    for bad in (None, '   '):
        with pytest.raises(CollectError):
            asyncio.run(collect_evidence_from_events(tenant_id=bad))


def test_event_table_candidates_include_m11b_table():
    """候选表名必须含 M11b 真实落库表 pbl_runtime_event（否则永远 fail-closed）。"""
    assert 'pbl_runtime_event' in collector.EVENT_TABLE_CANDIDATES
    assert isinstance(collector.EVENT_TABLE_CANDIDATES, tuple)


def test_resolve_event_table_returns_none_without_db(monkeypatch):
    """resolve_event_table 自身在无表时返回 None（不抛错），由上层决定 fail-closed。"""

    async def _no_table(table):
        return False

    monkeypatch.setattr(collector, 'table_exists', _no_table, raising=True)
    assert asyncio.run(collector.resolve_event_table()) is None


# ══════════════════════════════════════════════════════════════════════════
# 幂等契约（纯函数 + 表定义，零 DB）
# ══════════════════════════════════════════════════════════════════════════
def test_dedup_key_is_idempotent_and_distinguishes_each_dimension():
    """同一三元组重复计算必须同键；任一维度变化必须换键（否则幂等会误合并）。"""
    k1 = build_dedup_key('t1', 'evt-001', 'observation')
    k2 = build_dedup_key('t1', 'evt-001', 'observation')
    assert k1 == k2, '幂等键必须可重放（同输入同键）'
    assert isinstance(k1, str) and len(k1) == 32, 'dedup_key 须为 32 位定长（VARCHAR(32)）'
    assert all(c in '0123456789abcdef' for c in k1), 'dedup_key 须为十六进制串'

    assert build_dedup_key('t2', 'evt-001', 'observation') != k1
    assert build_dedup_key('t1', 'evt-002', 'observation') != k1
    assert build_dedup_key('t1', 'evt-001', 'artifact') != k1


def _load_model(table):
    path = MODELS_DIR / ('%s.json' % table)
    with open(path, 'r', encoding='utf-8') as fh:
        return json.load(fh)


def test_unique_index_matches_idempotency_triple():
    """uk_ev_dedup 列组合必须 == (tenant_id, source_event_id, evidence_type)。"""
    model = _load_model('pbl_evidence')
    fields = [f['name'] for f in model['fields']]
    for col in ('tenant_id', 'source_event_id', 'evidence_type', 'dedup_key'):
        assert col in fields, 'models/pbl_evidence.json 缺列 %s' % (col,)

    uk = [i for i in model.get('indexes', []) if i.get('name') == 'uk_ev_dedup']
    assert len(uk) == 1, '必须且只能有一条 uk_ev_dedup 唯一索引，实际 %s' % (uk,)
    assert uk[0]['idxtype'] == 'unique', 'uk_ev_dedup 必须是 unique 索引'
    assert list(uk[0]['idxfields']) == ['tenant_id', 'source_event_id', 'evidence_type'], \
        '唯一索引列组合与幂等键语义不一致: %s' % (uk[0]['idxfields'],)
    assert model['summary'][0]['primary'] == ['id'], 'primary 必须是数组 ["id"]'


def test_crud_json_uses_same_dedup_semantics():
    """CRUD 定义不得把幂等列暴露为可编辑（否则人工改数据破坏去重）。"""
    with open(REPO_ROOT / 'json' / 'pbl_evidence.json', 'r', encoding='utf-8') as fh:
        crud = json.load(fh)
    assert crud['tblname'] == 'pbl_evidence'
    params = crud['params']
    hidden = set(params.get('browserfields', {}).get('exclouded', [])) | \
        set(params.get('editexclouded', [])) | \
        set(params.get('editable', {}).get('exclouded', []) if isinstance(
            params.get('editable', {}).get('exclouded', []), list) else [])
    for col in ('dedup_key', 'tenant_id'):
        assert col in hidden, '%s 不应在列表/编辑面暴露' % (col,)


# ══════════════════════════════════════════════════════════════════════════
# 真实双次采集（有 DB 才跑；无 DB 明确 skip，不伪造通过）
# ══════════════════════════════════════════════════════════════════════════
@pytest.mark.skipif(not _db_enabled(),
                    reason='无可用测试库（PBL_EVIDENCE_TEST_DB 未开启）：'
                           '本环境仅验证 C-3 fail-closed 路径')
def test_double_collect_is_idempotent_with_db():
    """同一 source_event_id + evidence_type 连续采集两次：第二次 skipped≥1 且表内仅一行。"""
    tid = os.environ.get('PBL_EVIDENCE_TEST_TENANT', 'm5a-idem-test')
    first = asyncio.run(collect_evidence_from_events(tenant_id=tid, limit=5))
    second = asyncio.run(collect_evidence_from_events(tenant_id=tid, limit=5))

    assert first['scanned'] == second['scanned']
    assert second['created'] == 0, '重放不得新建: %s' % (second,)
    assert second['skipped'] >= 1, '重放应命中已存在证据: %s' % (second,)

    rows = asyncio.run(collector.q_all(
        'SELECT `source_event_id`, `evidence_type`, COUNT(*) AS c FROM `pbl_evidence` '
        'WHERE `tenant_id`=${tenant_id}$ GROUP BY `source_event_id`, `evidence_type` '
        'HAVING COUNT(*) > 1', {'tenant_id': tid}))
    assert not rows, '出现重复行，幂等失效: %s' % (rows,)


# =============================================================================
# M5a 收口验证记录（Git 选择性收口任务 task_id=92mX2ANX5qYN4d9Q9ZOEy, 2026-09-22）
# 本块为可追溯性元数据，不参与 pytest 断言。
# =============================================================================
M5A_CLOSURE_EVIDENCE = {
    "scope": [
        "scripts/selfcheck_m5a.py",
        "tests/test_m5a_idempotency.py",
    ],
    "selfcheck_log": "projects/pbls/docs/02-develop/m5a-selfcheck-run.log",
    "idempotency_log": "projects/pbls/docs/02-develop/m5a-idempotency-run.log",
    "upstream_commits": {
        "scripts/selfcheck_m5a.py": "0f5e22f",
        "tests/m5a_live_db_harness.py": "80156fd",
    },
    "db_available": False,
    "note": "环境无 DB，仅验证 fail-closed 路径（resolve_event_table() -> None 时抛 CollectError 且消息含 M11b）",
}