65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
"""故障日志过滤规则测试。"""
|
|
import _bootstrap # noqa: F401
|
|
import unittest
|
|
|
|
from fault_log_analyzer.filters import DEFAULT_LEVELS, FaultFilter
|
|
from fault_log_analyzer.models import FaultFilterRule, ParsedLog
|
|
|
|
|
|
def _log(level="ERROR", message="boom"):
|
|
return ParsedLog(level=level, message=message)
|
|
|
|
|
|
class TestDefaultFilter(unittest.TestCase):
|
|
def test_error_captured(self):
|
|
self.assertTrue(FaultFilter().should_capture(_log("ERROR", "x")))
|
|
|
|
def test_fatal_captured(self):
|
|
self.assertTrue(FaultFilter().should_capture(_log("FATAL", "x")))
|
|
|
|
def test_info_not_captured(self):
|
|
self.assertFalse(FaultFilter().should_capture(_log("INFO", "x")))
|
|
|
|
def test_warn_not_captured(self):
|
|
self.assertFalse(FaultFilter().should_capture(_log("WARN", "x")))
|
|
|
|
def test_default_levels_set(self):
|
|
self.assertEqual(DEFAULT_LEVELS, {"ERROR", "FATAL"})
|
|
|
|
|
|
class TestRuleFilter(unittest.TestCase):
|
|
def test_level_rule(self):
|
|
f = FaultFilter([FaultFilterRule(name="warn", level="WARN")])
|
|
self.assertTrue(f.should_capture(_log("WARN", "x")))
|
|
self.assertFalse(f.should_capture(_log("ERROR", "x")))
|
|
|
|
def test_pattern_rule(self):
|
|
f = FaultFilter([FaultFilterRule(name="oom", level="INFO", pattern="out of memory")])
|
|
self.assertTrue(f.should_capture(_log("INFO", "out of memory occurred")))
|
|
self.assertFalse(f.should_capture(_log("INFO", "nothing")))
|
|
|
|
def test_exclude_pattern(self):
|
|
f = FaultFilter(
|
|
[FaultFilterRule(name="no_heartbeat", level="ERROR", pattern=".*", exclude_pattern="heartbeat")]
|
|
)
|
|
self.assertFalse(f.should_capture(_log("ERROR", "heartbeat timeout")))
|
|
self.assertTrue(f.should_capture(_log("ERROR", "disk full")))
|
|
|
|
def test_disabled_rule_skipped(self):
|
|
f = FaultFilter([FaultFilterRule(name="off", level="WARN", enabled=False)])
|
|
self.assertFalse(f.should_capture(_log("WARN", "x")))
|
|
|
|
def test_add_rule(self):
|
|
f = FaultFilter()
|
|
f.add_rule(FaultFilterRule(name="warn", level="WARN"))
|
|
self.assertTrue(f.should_capture(_log("WARN", "x")))
|
|
|
|
def test_compile_rules(self):
|
|
rules = FaultFilter.compile_rules([{"name": "r1", "level": "WARN"}])
|
|
self.assertEqual(len(rules), 1)
|
|
self.assertEqual(rules[0].name, "r1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|