108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
import unittest
|
||
|
||
from detector.engine import DetectionEngine, aggregate_value, compare
|
||
from detector.models import MetricRule, MetricSample, RuleScope
|
||
from detector.rule_loader import RuleLoader
|
||
from detector.storage import InMemoryCache, InMemoryRuleStore
|
||
|
||
|
||
def make_rule(rule_id="r-cpu", metric="cpu_usage", operator="gt", threshold=90.0, for_duration="60s"):
|
||
return MetricRule(
|
||
rule_id=rule_id,
|
||
name="cpu high",
|
||
metric=metric,
|
||
aggregation="avg",
|
||
operator=operator,
|
||
threshold=threshold,
|
||
for_duration=for_duration,
|
||
severity="critical",
|
||
scope=RuleScope(scope_type="all"),
|
||
)
|
||
|
||
|
||
def make_engine(rule):
|
||
store = InMemoryRuleStore()
|
||
store.create_rule(rule)
|
||
loader = RuleLoader(store, InMemoryCache())
|
||
loader.load()
|
||
return DetectionEngine(loader, evaluation_interval="15s", recover_duration="15s")
|
||
|
||
|
||
class TestAggregateCompare(unittest.TestCase):
|
||
def test_aggregate(self):
|
||
self.assertEqual(aggregate_value([1, 2, 3], "avg"), 2)
|
||
self.assertEqual(aggregate_value([1, 2, 3], "min"), 1)
|
||
self.assertEqual(aggregate_value([1, 2, 3], "max"), 3)
|
||
self.assertEqual(aggregate_value([1, 2, 3], "sum"), 6)
|
||
self.assertEqual(aggregate_value([1, 2, 3], "last"), 3)
|
||
|
||
def test_compare(self):
|
||
self.assertTrue(compare(91, "gt", 90))
|
||
self.assertFalse(compare(90, "gt", 90))
|
||
self.assertTrue(compare(90, "gte", 90))
|
||
self.assertTrue(compare(89, "lt", 90))
|
||
self.assertTrue(compare(95, "between", 90, 100))
|
||
self.assertFalse(compare(105, "between", 90, 100))
|
||
|
||
|
||
class TestStateMachine(unittest.TestCase):
|
||
def test_firing_after_sustained(self):
|
||
rule = make_rule(for_duration="60s")
|
||
engine = make_engine(rule)
|
||
events = []
|
||
for i in range(5):
|
||
ts = float(i * 15)
|
||
sample = MetricSample(name="cpu_usage", value=95.0, timestamp=ts)
|
||
events.extend(engine.handle(sample, "h-1"))
|
||
|
||
self.assertEqual(len(events), 1)
|
||
self.assertEqual(events[0].status, "firing")
|
||
self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "firing")
|
||
|
||
def test_no_fire_before_for_duration(self):
|
||
rule = make_rule(for_duration="60s")
|
||
engine = make_engine(rule)
|
||
events = []
|
||
for i in range(3):
|
||
ts = float(i * 15)
|
||
events.extend(engine.handle(MetricSample(name="cpu_usage", value=95.0, timestamp=ts), "h-1"))
|
||
self.assertEqual(events, [])
|
||
self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "ok")
|
||
|
||
def test_resolved(self):
|
||
rule = make_rule(for_duration="60s")
|
||
engine = make_engine(rule)
|
||
events = []
|
||
for i in range(5):
|
||
events.extend(engine.handle(MetricSample(name="cpu_usage", value=95.0, timestamp=float(i * 15)), "h-1"))
|
||
self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "firing")
|
||
|
||
# 一次回落进入 pending
|
||
events.clear()
|
||
events.extend(engine.handle(MetricSample(name="cpu_usage", value=50.0, timestamp=75.0), "h-1"))
|
||
self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "pending")
|
||
|
||
# 观察期后仍回落 -> resolved
|
||
events.clear()
|
||
events.extend(engine.handle(MetricSample(name="cpu_usage", value=50.0, timestamp=90.0), "h-1"))
|
||
self.assertEqual(len(events), 1)
|
||
self.assertEqual(events[0].status, "resolved")
|
||
self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "ok")
|
||
|
||
def test_pending_back_to_firing_no_duplicate_event(self):
|
||
rule = make_rule(for_duration="60s")
|
||
engine = make_engine(rule)
|
||
for i in range(5):
|
||
engine.handle(MetricSample(name="cpu_usage", value=95.0, timestamp=float(i * 15)), "h-1")
|
||
engine.handle(MetricSample(name="cpu_usage", value=50.0, timestamp=75.0), "h-1")
|
||
self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "pending")
|
||
|
||
# 观察期内再次满足（低值样本滑出窗口后聚合值回升）-> 回到 firing，不重复触发
|
||
events = engine.handle(MetricSample(name="cpu_usage", value=96.0, timestamp=150.0), "h-1")
|
||
self.assertEqual(events, [])
|
||
self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "firing")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|