59 lines
2.3 KiB
Python
59 lines
2.3 KiB
Python
import _bootstrap # noqa: F401
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from fault_log_analyzer.filters import CaptureFilter
|
|
from fault_log_analyzer.models import FaultFilterRule, LogEntry
|
|
|
|
|
|
def entry(level="ERROR", message="something", host_id="h-1"):
|
|
return LogEntry(
|
|
timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc),
|
|
level=level,
|
|
message=message,
|
|
host_id=host_id,
|
|
)
|
|
|
|
|
|
class TestCaptureFilter(unittest.TestCase):
|
|
def test_default_levels(self):
|
|
f = CaptureFilter()
|
|
self.assertTrue(f.is_fault(entry(level="ERROR")))
|
|
self.assertTrue(f.is_fault(entry(level="FATAL")))
|
|
self.assertTrue(f.is_fault(entry(level="CRITICAL")))
|
|
self.assertFalse(f.is_fault(entry(level="WARN")))
|
|
self.assertFalse(f.is_fault(entry(level="INFO")))
|
|
|
|
def test_rule_pattern_hit(self):
|
|
f = CaptureFilter([FaultFilterRule(name="disk", level="ERROR", pattern="disk full")])
|
|
self.assertTrue(f.is_fault(entry(message="No space, disk full on device")))
|
|
self.assertFalse(f.is_fault(entry(message="connection refused")))
|
|
|
|
def test_rule_exclude_pattern(self):
|
|
f = CaptureFilter(
|
|
[FaultFilterRule(name="disk", level="ERROR", pattern="disk", exclude_pattern="ignore")]
|
|
)
|
|
self.assertTrue(f.is_fault(entry(message="disk full")))
|
|
self.assertFalse(f.is_fault(entry(message="disk ignore")))
|
|
|
|
def test_rule_level_mismatch(self):
|
|
f = CaptureFilter([FaultFilterRule(name="disk", level="FATAL", pattern="disk")])
|
|
self.assertFalse(f.is_fault(entry(level="ERROR", message="disk full")))
|
|
|
|
def test_configured_rules_do_not_fallback(self):
|
|
# 配置了规则但均未命中时不走默认级别兜底
|
|
f = CaptureFilter([FaultFilterRule(name="disk", level="ERROR", pattern="disk full")])
|
|
self.assertFalse(f.is_fault(entry(level="FATAL", message="connection refused")))
|
|
|
|
def test_disabled_rule_falls_back_to_default(self):
|
|
f = CaptureFilter(
|
|
[FaultFilterRule(name="disk", level="ERROR", pattern="disk", enabled=False)]
|
|
)
|
|
self.assertTrue(f.is_fault(entry(level="ERROR", message="anything")))
|
|
self.assertFalse(f.is_fault(entry(level="INFO", message="anything")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|