63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
|
# 审计完整性告警黄金测试（plan.md §3.6 / 矩阵「全审计」）
|
|||
|
|
# ============================================================
|
|||
|
|
# 覆盖：正常无告警、链断裂告警、锚定不匹配告警、未锚定/账本损坏/镜像源告警。
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from server.agent_core.audit_alerts import build_alerts
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _chain_ok() -> dict:
|
|||
|
|
return {"ok": True, "checked": 3, "brokenAt": None}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _chain_broken() -> dict:
|
|||
|
|
return {"ok": False, "checked": 2, "brokenAt": 3}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _anchor_ok() -> dict:
|
|||
|
|
return {"anchored": True, "ok": True, "reason": None}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _anchor_mismatch() -> dict:
|
|||
|
|
return {"anchored": True, "ok": False, "reason": "mismatch"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _anchor_not_anchored() -> dict:
|
|||
|
|
return {"anchored": False, "ok": False, "reason": "not-anchored"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _anchor_corrupt() -> dict:
|
|||
|
|
return {"anchored": True, "ok": False, "reason": "corrupt-ledger"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_no_alerts_when_all_ok():
|
|||
|
|
"""正常态：链/锚定一致、镜像源 → 无告警。"""
|
|||
|
|
alerts = build_alerts(_chain_ok(), _anchor_ok(), source="mirror")
|
|||
|
|
assert alerts == []
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_chain_broken_alert_critical():
|
|||
|
|
"""链断裂 → critical 告警。"""
|
|||
|
|
alerts = build_alerts(_chain_broken(), _anchor_ok(), source="mirror")
|
|||
|
|
assert any(a["code"] == "AUDIT_CHAIN_BROKEN" and a["severity"] == "critical" for a in alerts)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_anchor_mismatch_alert_warning():
|
|||
|
|
"""锚定不匹配 → warning 告警。"""
|
|||
|
|
alerts = build_alerts(_chain_ok(), _anchor_mismatch(), source="mirror")
|
|||
|
|
assert any(a["code"] == "AUDIT_ANCHOR_MISMATCH" and a["severity"] == "warning" for a in alerts)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_not_anchored_and_corrupt_alerts():
|
|||
|
|
"""未锚定/账本损坏 → info/critical 告警。"""
|
|||
|
|
na = build_alerts(_chain_ok(), _anchor_not_anchored(), source="mirror")
|
|||
|
|
assert any(a["code"] == "AUDIT_NOT_ANCHORED" for a in na)
|
|||
|
|
cc = build_alerts(_chain_ok(), _anchor_corrupt(), source="mirror")
|
|||
|
|
assert any(a["code"] == "AUDIT_LEDGER_CORRUPT" and a["severity"] == "critical" for a in cc)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_world_source_info_alert():
|
|||
|
|
"""事件源为世界状态（镜像未启用）→ info 告警。"""
|
|||
|
|
alerts = build_alerts(_chain_ok(), _anchor_ok(), source="world")
|
|||
|
|
assert any(a["code"] == "AUDIT_SOURCE_WORLD" for a in alerts)
|