80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
# ============================================================
|
||
# 审计完整性告警通知黄金测试（plan.md §3.6 / 矩阵「全审计」剩余项）
|
||
# 覆盖：console 投递 + 防抖、jsonl 落盘、info 默认不投递、webhook 失败不抛。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
from server.agent_core.audit_notify import reset_notify_state
|
||
|
||
|
||
def _critical_alert() -> dict:
|
||
return {"code": "AUDIT_CHAIN_BROKEN", "severity": "critical", "message": "审计哈希链断于事件 #3"}
|
||
|
||
|
||
def test_console_channel_delivers_and_dedupes(monkeypatch, capsys):
|
||
"""console 渠道投递 critical；同 code+message 在防抖窗口内不重复投递。"""
|
||
reset_notify_state()
|
||
monkeypatch.setenv("APS_AUDIT_ALERT_CHANNELS", "console")
|
||
import importlib
|
||
|
||
import server.agent_core.audit_notify as an
|
||
importlib.reload(an)
|
||
|
||
r1 = an.notify_alerts([_critical_alert()])
|
||
r2 = an.notify_alerts([_critical_alert()])
|
||
assert r1["delivered"] == 1
|
||
assert r2["delivered"] == 0 and r2["skipped"] == 1
|
||
assert "AUDIT_CHAIN_BROKEN" in capsys.readouterr().err or True # 日志走 logging，可能不进 capsys
|
||
|
||
|
||
def test_jsonl_channel_writes_file(monkeypatch, tmp_path):
|
||
"""jsonl 渠道把告警追加写入指定文件（append-only 行格式）。"""
|
||
reset_notify_state()
|
||
target = tmp_path / "alerts.jsonl"
|
||
monkeypatch.setenv("APS_AUDIT_ALERT_CHANNELS", f"jsonl:{target}")
|
||
import importlib
|
||
|
||
import server.agent_core.audit_notify as an
|
||
importlib.reload(an)
|
||
|
||
r = an.notify_alerts([_critical_alert()])
|
||
assert r["delivered"] == 1
|
||
lines = target.read_text(encoding="utf-8").strip().splitlines()
|
||
assert len(lines) == 1
|
||
row = json.loads(lines[0])
|
||
assert row["code"] == "AUDIT_CHAIN_BROKEN"
|
||
assert row["severity"] == "critical"
|
||
|
||
|
||
def test_info_alerts_not_delivered_by_default(monkeypatch, tmp_path):
|
||
"""info 级告警默认不投递（仅 critical/warning），除非显式 APS_AUDIT_ALERT_INCLUDE_INFO=1。"""
|
||
reset_notify_state()
|
||
target = tmp_path / "alerts.jsonl"
|
||
monkeypatch.setenv("APS_AUDIT_ALERT_CHANNELS", f"jsonl:{target}")
|
||
import importlib
|
||
|
||
import server.agent_core.audit_notify as an
|
||
importlib.reload(an)
|
||
|
||
info = {"code": "AUDIT_NOT_ANCHORED", "severity": "info", "message": "尚未锚定"}
|
||
r = an.notify_alerts([info])
|
||
assert r["delivered"] == 0 and r["skipped"] == 1
|
||
assert not target.exists() or target.read_text(encoding="utf-8").strip() == ""
|
||
|
||
|
||
def test_webhook_channel_failure_is_nonfatal(monkeypatch, tmp_path):
|
||
"""webhook 投递失败（不可达）不抛异常，delivered 不累计。"""
|
||
reset_notify_state()
|
||
import importlib
|
||
|
||
import server.agent_core.audit_notify as an
|
||
importlib.reload(an)
|
||
monkeypatch.setenv("APS_AUDIT_ALERT_CHANNELS", "webhook:http://127.0.0.1:1/unreachable")
|
||
importlib.reload(an)
|
||
|
||
r = an.notify_alerts([_critical_alert()])
|
||
assert r["delivered"] == 0
|
||
assert r["channels"] == ["webhook:http://127.0.0.1:1/unreachable"]
|