# -*- coding: utf-8 -*- """炉层杂质预警规则引擎单元测试（Issue #72）。 覆盖： - AlertCondition 运算匹配（含缺失值不触发、未知 op 拒绝）； - AlertRule AND 语义、空条件拒绝、id 重复拒绝； - AlertSeverity 排序（primary_alert 取最高）； - AlertRuleEngine.evaluate 命中（多规则按 severity 降序）； - 模板配置 YAML 加载（含 flow map condition、错误 severity/op 拒绝）； - 端到端：模板资产加载 → 急升温特征向量 → P0 命中（场景A 红色告警）。 """ import math import os import sys import unittest sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import _bootstrap # noqa: F401 挂载 impurity_forecast 包 from impurity_forecast import ( # noqa: E402 Alert, AlertCondition, AlertRule, AlertRuleEngine, AlertSeverity, load_alert_rules_config, ) NAN = float("nan") CONFIG_PATH = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "alert_rules.template.yaml") # --------------------------------------------------------------------------- # 1. AlertCondition # --------------------------------------------------------------------------- class AlertConditionTest(unittest.TestCase): def test_ops(self): self.assertTrue(AlertCondition("x", ">", 10).matches({"x": 11})) self.assertTrue(AlertCondition("x", ">=", 10).matches({"x": 10})) self.assertTrue(AlertCondition("x", "<", 10).matches({"x": 9})) self.assertTrue(AlertCondition("x", "<=", 10).matches({"x": 10})) self.assertTrue(AlertCondition("x", "==", 10).matches({"x": 10})) self.assertFalse(AlertCondition("x", ">", 10).matches({"x": 10})) def test_missing_value_not_match(self): self.assertFalse(AlertCondition("x", ">", 10).matches({"x": NAN})) self.assertFalse(AlertCondition("x", ">", 10).matches({"y": 100})) def test_unknown_op_rejected(self): with self.assertRaises(ValueError): AlertCondition("x", "!=", 10) # --------------------------------------------------------------------------- # 2. AlertRule # --------------------------------------------------------------------------- class AlertRuleTest(unittest.TestCase): def test_and_semantics(self): rule = AlertRule(id="r1", severity=AlertSeverity.P0, conditions=[ AlertCondition("a", ">", 10), AlertCondition("b", "<", 5), ]) self.assertTrue(rule.matches({"a": 11, "b": 4})) self.assertFalse(rule.matches({"a": 11, "b": 6})) # b 不满足 self.assertFalse(rule.matches({"a": 9, "b": 4})) # a 不满足 def test_empty_conditions_rejected(self): with self.assertRaises(ValueError): AlertRule(id="r", severity=AlertSeverity.P0, conditions=[]) def test_empty_id_rejected(self): with self.assertRaises(ValueError): AlertRule(id="", severity=AlertSeverity.P0, conditions=[AlertCondition("a", ">", 1)]) def test_describe(self): rule = AlertRule(id="r1", severity=AlertSeverity.P0, conditions=[ AlertCondition("炉温_ema5", ">", 900.0), ]) self.assertIn("P0", rule.describe()) self.assertIn("炉温_ema5>900.0", rule.describe()) # --------------------------------------------------------------------------- # 3. AlertRuleEngine # --------------------------------------------------------------------------- class AlertRuleEngineTest(unittest.TestCase): def _engine(self) -> AlertRuleEngine: return AlertRuleEngine([ AlertRule(id="p0_rule", severity=AlertSeverity.P0, conditions=[ AlertCondition("炉温", ">", 900.0)]), AlertRule(id="p1_rule", severity=AlertSeverity.P1, conditions=[ AlertCondition("氯气", ">", 8.0)]), AlertRule(id="p2_rule", severity=AlertSeverity.P2, conditions=[ AlertCondition("炉温", ">", 880.0)]), ]) def test_empty_rules_rejected(self): with self.assertRaises(ValueError): AlertRuleEngine([]) def test_duplicate_id_rejected(self): with self.assertRaises(ValueError): AlertRuleEngine([ AlertRule(id="dup", severity=AlertSeverity.P0, conditions=[AlertCondition("a", ">", 1)]), AlertRule(id="dup", severity=AlertSeverity.P1, conditions=[AlertCondition("b", ">", 1)]), ]) def test_evaluate_returns_sorted_by_severity(self): eng = self._engine() # 炉温=890 同时命中 p2（>880）；氯气=10 命中 p1 alerts = eng.evaluate(100, {"炉温": 890.0, "氯气": 10.0}) ids = [a.rule_id for a in alerts] self.assertEqual(ids, ["p1_rule", "p2_rule"]) # P1 > P2 self.assertEqual([a.severity for a in alerts], [AlertSeverity.P1, AlertSeverity.P2]) def test_primary_alert_picks_highest(self): eng = self._engine() # 炉温=920 命中 p0 + p2 → 主告警 P0 alerts = eng.evaluate(100, {"炉温": 920.0}) primary = eng.primary_alert(alerts) self.assertIsNotNone(primary) self.assertEqual(primary.severity, AlertSeverity.P0) def test_no_hit_returns_empty(self): eng = self._engine() self.assertEqual(eng.evaluate(100, {"炉温": 850.0, "氯气": 5.0}), []) self.assertIsNone(eng.primary_alert([])) # --------------------------------------------------------------------------- # 4. 模板配置 YAML 加载 # --------------------------------------------------------------------------- class ConfigLoadTest(unittest.TestCase): def test_load_template_config(self): cfg = load_alert_rules_config(CONFIG_PATH) self.assertEqual(cfg.template, "ti-cl4") self.assertGreaterEqual(len(cfg.rules), 5) ids = [r.id for r in cfg.rules] self.assertIn("bed_temp_critical", ids) self.assertIn("cl2_flow_warning", ids) def test_flow_map_condition_parsed(self): cfg = load_alert_rules_config(CONFIG_PATH) r = next(x for x in cfg.rules if x.id == "bed_temp_critical") self.assertEqual(r.severity, AlertSeverity.P0) self.assertEqual(r.conditions[0].feature, "炉温_ema5") self.assertEqual(r.conditions[0].op, ">") self.assertEqual(r.conditions[0].threshold, 900.0) self.assertEqual(r.sop, "SOP-CL-001") def test_engine_from_template_config(self): eng = AlertRuleEngine.from_template_config(CONFIG_PATH) # 炉温_ema5=920 命中 bed_temp_critical (P0) + bed_temp_near_limit (P2) alerts = eng.evaluate(1, {"炉温_ema5": 920.0}) ids = [a.rule_id for a in alerts] self.assertIn("bed_temp_critical", ids) self.assertEqual(eng.primary_alert(alerts).severity, AlertSeverity.P0) # --------------------------------------------------------------------------- # 5. 端到端：急升温场景命中 P0（PRD 场景A 红色告警） # --------------------------------------------------------------------------- class EndToEndScenarioATest(unittest.TestCase): def test_rising_temp_triggers_p0(self): """模拟炉层杂质富集的急升温：规则引擎应在 ema 平滑值越界时产出 P0 告警。""" eng = AlertRuleEngine.from_template_config(CONFIG_PATH) # 特征向量：炉温_ema5 越过 900 上限 values = {"炉温_ema5": 905.0, "炉温_rate10": 0.06, "氯气流量_std10": 5.0, "炉压_rate10": 0.02, "炉层状态_mean10": 60.0} alerts = eng.evaluate(timestamp=100, values=values) primary = eng.primary_alert(alerts) self.assertIsNotNone(primary, "急升温应触发预警") self.assertEqual(primary.severity, AlertSeverity.P0, "主告警应为 P0 红色告警") # 命中的 P0 规则应有处置 SOP（供 LLM 报警解释 + 值班长确认） self.assertTrue(any(a.sop for a in alerts if a.severity == AlertSeverity.P0)) if __name__ == "__main__": unittest.main(verbosity=2)