From 36a9f795436b80d4a94383cca3c7e7e0a4ba357c Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Tue, 22 Sep 2026 15:14:52 +0800 Subject: [PATCH] test(m5a): idempotency + C-3 fail-closed pytest (resolve_event_table None -> CollectError M11b) --- tests/test_m5a_idempotency.py | 296 ++++++++++++++++++++++++++++++++++ 1 file changed, 296 insertions(+) create mode 100644 tests/test_m5a_idempotency.py diff --git a/tests/test_m5a_idempotency.py b/tests/test_m5a_idempotency.py new file mode 100644 index 0000000..dc2748e --- /dev/null +++ b/tests/test_m5a_idempotency.py @@ -0,0 +1,296 @@ +#!/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,)