55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
import unittest
|
|
|
|
from detector.models import MetricRule, MetricSample, RuleScope
|
|
from detector.rule_loader import RuleLoader
|
|
from detector.storage import InMemoryCache, InMemoryRuleStore
|
|
|
|
|
|
class TestRuleLoader(unittest.TestCase):
|
|
def setUp(self):
|
|
self.store = InMemoryRuleStore()
|
|
self.cache = InMemoryCache()
|
|
self.loader = RuleLoader(self.store, self.cache, reload_interval="10s")
|
|
|
|
def test_load_and_match(self):
|
|
rule = MetricRule(
|
|
rule_id="r-1",
|
|
name="cpu",
|
|
metric="cpu_usage",
|
|
operator="gt",
|
|
threshold=90,
|
|
scope=RuleScope(scope_type="host_group", host_group="web"),
|
|
)
|
|
self.store.create_rule(rule)
|
|
self.loader.load()
|
|
|
|
sample = MetricSample(name="cpu_usage", value=95, timestamp=0)
|
|
matched = self.loader.match(sample, "h-1", host_group="web")
|
|
self.assertEqual([r.rule_id for r in matched], ["r-1"])
|
|
|
|
# 分组不匹配
|
|
self.assertEqual(self.loader.match(sample, "h-1", host_group="db"), [])
|
|
|
|
def test_hot_reload_by_version(self):
|
|
self.loader.load()
|
|
self.assertEqual(self.loader.get_rules(), [])
|
|
|
|
self.store.create_rule(
|
|
MetricRule(rule_id="r-2", name="mem", metric="mem_used_percent", threshold=85)
|
|
)
|
|
# 版本变化后手动 load 模拟热加载线程
|
|
self.assertNotEqual(self.store.get_version(), self.loader._version)
|
|
self.loader.load()
|
|
self.assertEqual(len(self.loader.get_rules()), 1)
|
|
|
|
def test_disabled_rule_not_matched(self):
|
|
rule = MetricRule(rule_id="r-3", name="disk", metric="disk_used_percent", enabled=False)
|
|
self.store.create_rule(rule)
|
|
self.loader.load()
|
|
sample = MetricSample(name="disk_used_percent", value=95, timestamp=0)
|
|
self.assertEqual(self.loader.match(sample, "h-1"), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|