124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
"""根因分析（RootCauseAnalyzer）单元测试。"""
|
||
import _bootstrap # noqa: F401
|
||
import unittest
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from fault_log_analyzer.models import Event, FaultLog
|
||
from fault_log_analyzer.root_cause import RootCauseAnalyzer, _as_aware
|
||
|
||
BASE = datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
|
||
NAIVE_BASE = datetime(2025, 1, 1, 10, 0, 0)
|
||
|
||
|
||
def _log(message, occurred_at=BASE, host_id="h-1", fault_log_id="fl-1"):
|
||
return FaultLog(
|
||
fault_log_id=fault_log_id,
|
||
host_id=host_id,
|
||
message=message,
|
||
level="ERROR",
|
||
occurred_at=occurred_at,
|
||
)
|
||
|
||
|
||
class TestAsAware(unittest.TestCase):
|
||
def test_aware_kept(self):
|
||
self.assertEqual(_as_aware(BASE), BASE)
|
||
|
||
def test_naive_becomes_utc(self):
|
||
out = _as_aware(NAIVE_BASE)
|
||
self.assertEqual(out.tzinfo, timezone.utc)
|
||
self.assertEqual(out, BASE)
|
||
|
||
def test_none(self):
|
||
self.assertIsNone(_as_aware(None))
|
||
|
||
|
||
class TestRootCauseAnalyzer(unittest.TestCase):
|
||
def setUp(self):
|
||
self.analyzer = RootCauseAnalyzer(window_minutes=5)
|
||
|
||
def test_log_hint_only(self):
|
||
rc = self.analyzer.analyze(_log("No space left on device /var"), [])
|
||
self.assertEqual(rc.cause_type, "disk_full")
|
||
self.assertEqual(rc.confidence, 0.6)
|
||
self.assertEqual(rc.fault_log_id, "fl-1")
|
||
|
||
def test_unknown_without_evidence(self):
|
||
rc = self.analyzer.analyze(_log("some unrelated message"), [])
|
||
self.assertEqual(rc.cause_type, "unknown")
|
||
self.assertEqual(rc.confidence, 0.0)
|
||
|
||
def test_matching_event_corroborates(self):
|
||
event = Event(
|
||
event_id="e-1",
|
||
host_id="h-1",
|
||
metric="disk_used_percent",
|
||
value=97.2,
|
||
threshold=90.0,
|
||
fired_at=BASE,
|
||
)
|
||
rc = self.analyzer.analyze(_log("No space left on device"), [event])
|
||
self.assertEqual(rc.cause_type, "disk_full")
|
||
self.assertEqual(rc.confidence, 1.0)
|
||
self.assertTrue(any(e["type"] == "event" and e["event_id"] == "e-1" for e in rc.evidence))
|
||
|
||
def test_naive_datetime_aware_event(self):
|
||
"""naive 日志时间 + aware 事件时间混用不应抛异常（鲁棒性）。"""
|
||
event = Event(
|
||
event_id="e-1",
|
||
host_id="h-1",
|
||
metric="mem_used_percent",
|
||
value=95.0,
|
||
threshold=90.0,
|
||
fired_at=BASE,
|
||
)
|
||
rc = self.analyzer.analyze(_log("out of memory", occurred_at=NAIVE_BASE), [event])
|
||
self.assertEqual(rc.cause_type, "memory_exhausted")
|
||
self.assertEqual(rc.confidence, 1.0)
|
||
|
||
def test_naive_event_outside_window(self):
|
||
"""窗口外的 naive 事件不应被关联，也不应抛异常（鲁棒性）。"""
|
||
outside = Event(
|
||
event_id="e-1",
|
||
host_id="h-1",
|
||
metric="disk_used_percent",
|
||
value=97.2,
|
||
threshold=90.0,
|
||
fired_at=NAIVE_BASE + timedelta(hours=1),
|
||
)
|
||
rc = self.analyzer.analyze(_log("No space left on device"), [outside])
|
||
self.assertEqual(rc.cause_type, "disk_full")
|
||
self.assertEqual(rc.confidence, 0.6)
|
||
self.assertEqual(len(rc.evidence), 1)
|
||
|
||
def test_host_mismatch_excluded(self):
|
||
event = Event(
|
||
event_id="e-1",
|
||
host_id="h-other",
|
||
metric="disk_used_percent",
|
||
value=97.2,
|
||
threshold=90.0,
|
||
fired_at=BASE,
|
||
)
|
||
rc = self.analyzer.analyze(_log("No space left on device"), [event])
|
||
self.assertEqual(rc.cause_type, "disk_full")
|
||
self.assertEqual(rc.confidence, 0.6)
|
||
|
||
def test_event_only_cause(self):
|
||
"""日志无关键字但事件命中规则库时，以事件规则为准。"""
|
||
event = Event(
|
||
event_id="e-1",
|
||
host_id="h-1",
|
||
metric="cpu_usage",
|
||
value=99.0,
|
||
threshold=95.0,
|
||
fired_at=BASE,
|
||
)
|
||
rc = self.analyzer.analyze(_log("service degraded"), [event])
|
||
self.assertEqual(rc.cause_type, "cpu_saturation")
|
||
self.assertEqual(rc.confidence, 0.25)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|