92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
"""根因分析测试。"""
|
|
import _bootstrap # noqa: F401
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from fault_log_analyzer.models import Event, FaultLog
|
|
from fault_log_analyzer.root_cause import RootCauseAnalyzer
|
|
|
|
BASE = datetime(2025, 1, 1, 10, 18, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _log(msg, host="h-1", trace=None, at=None, fid="fl-1"):
|
|
return FaultLog(
|
|
fault_log_id=fid,
|
|
host_id=host,
|
|
message=msg,
|
|
trace_id=trace,
|
|
occurred_at=at or BASE,
|
|
)
|
|
|
|
|
|
def _event(host="h-1", metric="disk_used_percent", value=97.2, threshold=90.0, at=None):
|
|
return Event(
|
|
event_id="e-1",
|
|
host_id=host,
|
|
metric=metric,
|
|
value=value,
|
|
threshold=threshold,
|
|
fired_at=at or BASE,
|
|
)
|
|
|
|
|
|
class TestLogHints(unittest.TestCase):
|
|
def test_disk_hint(self):
|
|
rc = RootCauseAnalyzer().analyze(_log("no space left on device"), [])
|
|
self.assertEqual(rc.cause_type, "disk_full")
|
|
self.assertGreater(rc.confidence, 0.0)
|
|
|
|
def test_memory_hint(self):
|
|
rc = RootCauseAnalyzer().analyze(_log("out of memory killed"), [])
|
|
self.assertEqual(rc.cause_type, "memory_exhausted")
|
|
|
|
def test_unknown_when_no_signal(self):
|
|
rc = RootCauseAnalyzer().analyze(_log("hello world"), [])
|
|
self.assertEqual(rc.cause_type, "unknown")
|
|
self.assertEqual(rc.confidence, 0.0)
|
|
|
|
|
|
class TestMetricEvents(unittest.TestCase):
|
|
def test_metric_event_rule(self):
|
|
rc = RootCauseAnalyzer().analyze(_log("some error"), [_event()])
|
|
self.assertEqual(rc.cause_type, "disk_full")
|
|
self.assertGreater(rc.confidence, 0.0)
|
|
|
|
def test_event_outside_window_ignored(self):
|
|
later = datetime(2025, 1, 1, 11, 0, 0, tzinfo=timezone.utc)
|
|
rc = RootCauseAnalyzer().analyze(_log("some error"), [_event(at=later)])
|
|
self.assertEqual(rc.cause_type, "unknown")
|
|
|
|
def test_different_host_event_ignored(self):
|
|
rc = RootCauseAnalyzer().analyze(_log("some error"), [_event(host="h-2")])
|
|
self.assertEqual(rc.cause_type, "unknown")
|
|
|
|
def test_memory_metric_rule(self):
|
|
event = _event(metric="mem_used_percent", value=95.0, threshold=90.0)
|
|
rc = RootCauseAnalyzer().analyze(_log("some error"), [event])
|
|
self.assertEqual(rc.cause_type, "memory_exhausted")
|
|
|
|
|
|
class TestTraceAssociation(unittest.TestCase):
|
|
def test_trace_association(self):
|
|
log = _log("request failed", trace="tr-1")
|
|
related = FaultLog(
|
|
fault_log_id="fl-2",
|
|
host_id="h-1",
|
|
message="out of memory",
|
|
trace_id="tr-1",
|
|
occurred_at=BASE,
|
|
)
|
|
rc = RootCauseAnalyzer().analyze(log, [], related_logs=[related])
|
|
self.assertEqual(rc.cause_type, "memory_exhausted")
|
|
|
|
def test_evidence_collected(self):
|
|
rc = RootCauseAnalyzer().analyze(_log("no space left on device"), [_event()])
|
|
types = {e["type"] for e in rc.evidence}
|
|
self.assertIn("log", types)
|
|
self.assertIn("event", types)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|