@@ -0,0 +1,338 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""报警看板配置化单元测试（issue #52 / PRD 5.5「⑤ 配置化驾驶舱」）。
|
||||
|
||||
覆盖：
|
||||
1. 合法配置（Ti 模板）→ 校验通过、可解析为 ``AlarmPanelConfig``、序列化往返一致；
|
||||
2. 默认值与字段别名（snake_case / camelCase 都接受）、thresholds_source 选填；
|
||||
3. ``render_alarm_panel_props`` 正确编译 severity 样式 / 规则源 / 确认策略；
|
||||
4. ``filter_alarms_by_mute`` 静默下限 + severity 降序排序；
|
||||
5. 各类非法情况：非对象、错误 $schema、severityColors 缺级/重复/非法、
|
||||
rulesSource 缺失/asset 缺 ref/inline 缺 data、布尔/整数/枚举/区间非法；
|
||||
6. ``load_alarm_config`` 校验失败抛 ``AlarmConfigError`` 并携带全部错误。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import unittest
|
||||
|
||||
from cockpit import ( # type: ignore[import-not-found]
|
||||
ALARM_CONFIG_SCHEMA_ID,
|
||||
DEFAULT_ACK_TIMEOUT_S,
|
||||
DEFAULT_MAX_ITEMS,
|
||||
AlarmConfigError,
|
||||
AlarmPanelConfig,
|
||||
filter_alarms_by_mute,
|
||||
load_alarm_config,
|
||||
render_alarm_panel_props,
|
||||
validate_alarm_config,
|
||||
)
|
||||
|
||||
|
||||
def _valid_config() -> dict:
|
||||
"""返回一份合法的告警面板配置（Ti 模板风格）。"""
|
||||
return {
|
||||
"$schema": ALARM_CONFIG_SCHEMA_ID,
|
||||
"severityColors": [
|
||||
{"severity": "P0", "fg": "#ff3b30", "bg": "rgba(255,59,48,0.12)",
|
||||
"icon": "alert-octagon", "border": "#ff3b30"},
|
||||
{"severity": "P1", "fg": "#f5a623", "bg": "rgba(245,166,35,0.12)"},
|
||||
{"severity": "P2", "fg": "#3aa0ff", "bg": "rgba(58,160,255,0.10)"},
|
||||
],
|
||||
"rulesSource": {"kind": "asset", "ref": "ti-cl4/cfg/alert_rules.yaml"},
|
||||
"showSop": True,
|
||||
"requireAck": True,
|
||||
"ackTimeoutS": 300,
|
||||
"muteLower": "P2",
|
||||
"groupBy": "severity",
|
||||
"sortBy": "severity",
|
||||
"maxItems": 50,
|
||||
}
|
||||
|
||||
|
||||
class ValidateTest(unittest.TestCase):
|
||||
def test_valid_config_passes(self):
|
||||
res = validate_alarm_config(_valid_config())
|
||||
self.assertTrue(res.ok, msg=res.errors)
|
||||
self.assertEqual(res.errors, [])
|
||||
norm = res.normalized
|
||||
assert norm is not None
|
||||
# 规范化：severityColors 按 P0/P1/P2 顺序回填
|
||||
self.assertEqual([c["severity"] for c in norm["severityColors"]], ["P0", "P1", "P2"])
|
||||
# P0 的可选 icon/border 保留
|
||||
self.assertEqual(norm["severityColors"][0]["icon"], "alert-octagon")
|
||||
|
||||
def test_non_dict_rejected(self):
|
||||
res = validate_alarm_config(["not", "a", "dict"])
|
||||
self.assertFalse(res.ok)
|
||||
self.assertIn("对象（dict）", res.errors[0])
|
||||
|
||||
def test_wrong_schema_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["$schema"] = "something-else"
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("$schema" in e for e in res.errors))
|
||||
|
||||
def test_snake_case_aliases_accepted(self):
|
||||
# snake_case 字段名同样接受（向后兼容手写 YAML）
|
||||
cfg = {
|
||||
"severity_colors": [
|
||||
{"severity": "P0", "fg": "#f", "bg": "#0"},
|
||||
{"severity": "P1", "fg": "#f", "bg": "#0"},
|
||||
{"severity": "P2", "fg": "#f", "bg": "#0"},
|
||||
],
|
||||
"rules_source": {"kind": "asset", "ref": "a.yaml"},
|
||||
"show_sop": False,
|
||||
"require_ack": False,
|
||||
"mute_lower": "P2",
|
||||
"group_by": "severity",
|
||||
"sort_by": "severity",
|
||||
"max_items": 10,
|
||||
}
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertTrue(res.ok, msg=res.errors)
|
||||
|
||||
def test_missing_severity_level_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["severityColors"] = cfg["severityColors"][:2] # 只留 P0/P1
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("缺少 P2" in e for e in res.errors))
|
||||
|
||||
def test_duplicate_and_invalid_severity_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["severityColors"] = [
|
||||
{"severity": "P0", "fg": "#f", "bg": "#0"},
|
||||
{"severity": "P0", "fg": "#f", "bg": "#0"}, # 重复
|
||||
{"severity": "P9", "fg": "#f", "bg": "#0"}, # 非法
|
||||
]
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
joined = "; ".join(res.errors)
|
||||
self.assertIn("重复", joined)
|
||||
self.assertIn("P9", joined)
|
||||
|
||||
def test_missing_required_color_key_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["severityColors"][0] = {"severity": "P0", "fg": "#f"} # 缺 bg
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("bg" in e for e in res.errors))
|
||||
|
||||
def test_missing_rules_source_rejected(self):
|
||||
cfg = _valid_config()
|
||||
del cfg["rulesSource"]
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("rulesSource" in e for e in res.errors))
|
||||
|
||||
def test_asset_source_without_ref_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["rulesSource"] = {"kind": "asset"} # 缺 ref
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("ref" in e for e in res.errors))
|
||||
|
||||
def test_inline_source_without_data_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["rulesSource"] = {"kind": "inline"} # 缺 data
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("data" in e for e in res.errors))
|
||||
|
||||
def test_inline_source_item_without_id_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["rulesSource"] = {"kind": "inline", "data": [{"severity": "P0"}]} # 缺 id
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("id" in e for e in res.errors))
|
||||
|
||||
def test_invalid_source_kind_rejected(self):
|
||||
cfg = _valid_config()
|
||||
cfg["rulesSource"] = {"kind": "graphql", "ref": "x"}
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("kind" in e for e in res.errors))
|
||||
|
||||
def test_bool_fields_type_checked(self):
|
||||
for field in ("showSop", "requireAck"):
|
||||
cfg = _valid_config()
|
||||
cfg[field] = "yes"
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok, msg=field)
|
||||
self.assertTrue(any(field in e for e in res.errors), msg=field)
|
||||
|
||||
def test_ack_timeout_must_be_positive_int(self):
|
||||
cfg = _valid_config()
|
||||
cfg["ackTimeoutS"] = 0
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertTrue(any("ackTimeoutS" in e for e in res.errors))
|
||||
# bool 不应被当成 int 接受
|
||||
cfg["ackTimeoutS"] = True
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok)
|
||||
|
||||
def test_enum_fields_validated(self):
|
||||
for field, valid, invalid in [
|
||||
("muteLower", ["P0", "P1", "P2"], "P9"),
|
||||
("groupBy", ["severity", "rule", "process"], "color"),
|
||||
("sortBy", ["severity", "time"], "color"),
|
||||
]:
|
||||
for v in valid:
|
||||
cfg = _valid_config()
|
||||
cfg[field] = v
|
||||
# muteLower=P0 合法但会过滤掉 P1/P2，不影响校验合法性
|
||||
self.assertTrue(validate_alarm_config(cfg).ok, msg=f"{field}={v}")
|
||||
cfg = _valid_config()
|
||||
cfg[field] = invalid
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertFalse(res.ok, msg=f"{field}={invalid}")
|
||||
self.assertTrue(any(field in e for e in res.errors), msg=field)
|
||||
|
||||
def test_max_items_range(self):
|
||||
cfg = _valid_config()
|
||||
cfg["maxItems"] = 0
|
||||
self.assertFalse(validate_alarm_config(cfg).ok)
|
||||
cfg["maxItems"] = 501
|
||||
self.assertFalse(validate_alarm_config(cfg).ok)
|
||||
cfg["maxItems"] = 500
|
||||
self.assertTrue(validate_alarm_config(cfg).ok)
|
||||
cfg["maxItems"] = True # bool 不被当成 int
|
||||
self.assertFalse(validate_alarm_config(cfg).ok)
|
||||
|
||||
def test_defaults_applied_when_optional_omitted(self):
|
||||
cfg = {
|
||||
"severityColors": [
|
||||
{"severity": "P0", "fg": "#f", "bg": "#0"},
|
||||
{"severity": "P1", "fg": "#f", "bg": "#0"},
|
||||
{"severity": "P2", "fg": "#f", "bg": "#0"},
|
||||
],
|
||||
"rulesSource": {"kind": "asset", "ref": "a.yaml"},
|
||||
}
|
||||
res = validate_alarm_config(cfg)
|
||||
self.assertTrue(res.ok, msg=res.errors)
|
||||
assert res.normalized is not None
|
||||
self.assertEqual(res.normalized["showSop"], True)
|
||||
self.assertEqual(res.normalized["requireAck"], False)
|
||||
self.assertEqual(res.normalized["ackTimeoutS"], DEFAULT_ACK_TIMEOUT_S)
|
||||
self.assertEqual(res.normalized["muteLower"], "P2")
|
||||
self.assertEqual(res.normalized["groupBy"], "severity")
|
||||
self.assertEqual(res.normalized["sortBy"], "severity")
|
||||
self.assertEqual(res.normalized["maxItems"], DEFAULT_MAX_ITEMS)
|
||||
# thresholds_source 选填，未给出则 normalized 不含它
|
||||
self.assertNotIn("thresholdsSource", res.normalized)
|
||||
|
||||
def test_multiple_errors_aggregated(self):
|
||||
# 一次性破坏多个字段，校验器应聚合多条错误而非首条即返
|
||||
bad = {
|
||||
"severityColors": [{"severity": "P0", "fg": "#f", "bg": "#0"}], # 缺 P1/P2
|
||||
"rulesSource": {"kind": "asset"}, # 缺 ref
|
||||
"ackTimeoutS": -1,
|
||||
"sortBy": "color",
|
||||
}
|
||||
res = validate_alarm_config(bad)
|
||||
self.assertFalse(res.ok)
|
||||
self.assertGreaterEqual(len(res.errors), 4)
|
||||
|
||||
|
||||
class LoadTest(unittest.TestCase):
|
||||
def test_load_valid_returns_model(self):
|
||||
cfg = load_alarm_config(_valid_config())
|
||||
self.assertIsInstance(cfg, AlarmPanelConfig)
|
||||
self.assertEqual(cfg.schema, ALARM_CONFIG_SCHEMA_ID)
|
||||
self.assertEqual(len(cfg.severity_colors), 3)
|
||||
self.assertEqual(cfg.rules_source.kind, "asset")
|
||||
self.assertEqual(cfg.rules_source.ref, "ti-cl4/cfg/alert_rules.yaml")
|
||||
self.assertTrue(cfg.require_ack)
|
||||
|
||||
def test_load_roundtrip(self):
|
||||
original = _valid_config()
|
||||
cfg = load_alarm_config(original)
|
||||
# 内存模型序列化后再次校验仍合法（往返一致）
|
||||
res = validate_alarm_config(cfg.to_dict())
|
||||
self.assertTrue(res.ok, msg=res.errors)
|
||||
|
||||
def test_load_invalid_raises_with_errors(self):
|
||||
bad = copy.deepcopy(_valid_config())
|
||||
del bad["rulesSource"]
|
||||
with self.assertRaises(AlarmConfigError) as cm:
|
||||
load_alarm_config(bad)
|
||||
# 异常携带全部错误清单
|
||||
self.assertGreaterEqual(len(cm.exception.errors), 1)
|
||||
self.assertTrue(any("rulesSource" in e for e in cm.exception.errors))
|
||||
|
||||
|
||||
class RenderPropsTest(unittest.TestCase):
|
||||
def test_props_contain_severity_styles_and_sources(self):
|
||||
cfg = load_alarm_config(_valid_config())
|
||||
props = render_alarm_panel_props(cfg)
|
||||
# 保留 #51 的订阅语义（向后兼容）
|
||||
self.assertEqual(props["subscribe"], "alarm_stream")
|
||||
# severity 样式：P0 红色 + 可选 border
|
||||
self.assertEqual(props["severityStyles"]["P0"]["fg"], "#ff3b30")
|
||||
self.assertEqual(props["severityStyles"]["P0"]["border"], "#ff3b30")
|
||||
# asset 规则源透传 ref
|
||||
self.assertEqual(props["rulesSource"]["kind"], "asset")
|
||||
self.assertTrue(props["rulesSource"]["ref"].endswith("alert_rules.yaml"))
|
||||
# require_ack=True 携带 ackTimeoutS
|
||||
self.assertEqual(props["ackTimeoutS"], 300)
|
||||
self.assertTrue(props["showSop"])
|
||||
|
||||
def test_props_ack_omitted_when_not_required(self):
|
||||
cfg = _valid_config()
|
||||
cfg["requireAck"] = False
|
||||
props = render_alarm_panel_props(load_alarm_config(cfg))
|
||||
self.assertNotIn("ackTimeoutS", props)
|
||||
self.assertFalse(props["requireAck"])
|
||||
|
||||
def test_props_inline_source_data_passed_through(self):
|
||||
cfg = _valid_config()
|
||||
cfg["rulesSource"] = {
|
||||
"kind": "inline",
|
||||
"data": [{"id": "r1", "severity": "P0"}, {"id": "r2", "severity": "P1"}],
|
||||
}
|
||||
props = render_alarm_panel_props(load_alarm_config(cfg))
|
||||
self.assertEqual(props["rulesSource"]["kind"], "inline")
|
||||
self.assertEqual(len(props["rulesSource"]["data"]), 2)
|
||||
|
||||
def test_props_thresholds_source_optional(self):
|
||||
# 未声明 thresholds_source → props 不含该键
|
||||
props = render_alarm_panel_props(load_alarm_config(_valid_config()))
|
||||
self.assertNotIn("thresholdsSource", props)
|
||||
# 声明后 → 透传
|
||||
cfg = _valid_config()
|
||||
cfg["thresholdsSource"] = {"kind": "asset", "ref": "thr.yaml"}
|
||||
props = render_alarm_panel_props(load_alarm_config(cfg))
|
||||
self.assertEqual(props["thresholdsSource"]["ref"], "thr.yaml")
|
||||
|
||||
|
||||
class FilterMuteTest(unittest.TestCase):
|
||||
def test_filter_keeps_above_floor_and_sorts_desc(self):
|
||||
alarms = [
|
||||
{"severity": "P2", "id": "a"},
|
||||
{"severity": "P0", "id": "b"},
|
||||
{"severity": "P1", "id": "c"},
|
||||
]
|
||||
# mute=P2：全保留，按严重度降序
|
||||
kept = filter_alarms_by_mute(alarms, "P2")
|
||||
self.assertEqual([a["id"] for a in kept], ["b", "c", "a"])
|
||||
# mute=P1：滤掉 P2
|
||||
kept = filter_alarms_by_mute(alarms, "P1")
|
||||
self.assertEqual([a["id"] for a in kept], ["b", "c"])
|
||||
# mute=P0：只留 P0
|
||||
kept = filter_alarms_by_mute(alarms, "P0")
|
||||
self.assertEqual([a["id"] for a in kept], ["b"])
|
||||
|
||||
def test_filter_ignores_unknown_severity(self):
|
||||
alarms = [{"severity": "P9", "id": "x"}, {"severity": "P0", "id": "y"}]
|
||||
kept = filter_alarms_by_mute(alarms, "P2")
|
||||
self.assertEqual([a["id"] for a in kept], ["y"])
|
||||
|
||||
def test_filter_invalid_mute_raises(self):
|
||||
with self.assertRaises(AlarmConfigError):
|
||||
filter_alarms_by_mute([], "P9")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user