317 lines
14 KiB
Python
317 lines
14 KiB
Python
#!/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）",
|
||
}
|