94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
"""存储抽象与内存实现测试。"""
|
|
import _bootstrap # noqa: F401
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from fault_log_analyzer.models import Event, FaultFilterRule, FaultLog, FaultType, RootCause
|
|
from fault_log_analyzer.storage import MemoryStorage
|
|
|
|
BASE = datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _log(fid, host="h-1", msg="boom", level="ERROR", ft="disk_full", at=None):
|
|
return FaultLog(
|
|
fault_log_id=fid,
|
|
host_id=host,
|
|
message=msg,
|
|
level=level,
|
|
fault_type=ft,
|
|
occurred_at=at or BASE,
|
|
)
|
|
|
|
|
|
class TestFaultLogStorage(unittest.TestCase):
|
|
def setUp(self):
|
|
self.s = MemoryStorage()
|
|
|
|
def test_save_and_get(self):
|
|
self.s.save_fault_log(_log("fl-1"))
|
|
self.assertIsNotNone(self.s.get_fault_log("fl-1"))
|
|
|
|
def test_get_missing(self):
|
|
self.assertIsNone(self.s.get_fault_log("nope"))
|
|
|
|
def test_list_sorted_by_time(self):
|
|
self.s.save_fault_log(_log("fl-1", at=datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc)))
|
|
self.s.save_fault_log(_log("fl-2", at=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone.utc)))
|
|
self.assertEqual([l.fault_log_id for l in self.s.list_fault_logs()], ["fl-2", "fl-1"])
|
|
|
|
def test_query_filters(self):
|
|
self.s.save_fault_log(_log("fl-1", host="h-1", ft="disk_full", msg="no space"))
|
|
self.s.save_fault_log(_log("fl-2", host="h-2", ft="memory_exhausted", msg="oom"))
|
|
self.assertEqual(self.s.query_fault_logs(host_id="h-1")["total"], 1)
|
|
self.assertEqual(self.s.query_fault_logs(fault_type="memory_exhausted")["total"], 1)
|
|
self.assertEqual(self.s.query_fault_logs(keyword="OOM")["total"], 1)
|
|
self.assertEqual(self.s.query_fault_logs(level="error")["total"], 2)
|
|
|
|
def test_query_pagination(self):
|
|
for i in range(5):
|
|
self.s.save_fault_log(_log(f"fl-{i}"))
|
|
r = self.s.query_fault_logs(page=2, page_size=2)
|
|
self.assertEqual(r["total"], 5)
|
|
self.assertEqual(len(r["items"]), 2)
|
|
|
|
|
|
class TestOtherEntities(unittest.TestCase):
|
|
def setUp(self):
|
|
self.s = MemoryStorage()
|
|
|
|
def test_root_cause(self):
|
|
self.s.save_root_cause(RootCause(fault_log_id="fl-1", cause_type="disk_full"))
|
|
self.assertEqual(self.s.get_root_cause("fl-1").cause_type, "disk_full")
|
|
self.assertIsNone(self.s.get_root_cause("nope"))
|
|
|
|
def test_fault_types(self):
|
|
self.s.add_fault_type(FaultType(fault_type="x", name="X"))
|
|
self.assertEqual(self.s.get_fault_type("x").name, "X")
|
|
self.assertEqual(len(self.s.list_fault_types()), 1)
|
|
|
|
def test_filter_rules(self):
|
|
self.s.add_filter_rule(FaultFilterRule(name="r1", level="ERROR"))
|
|
self.assertEqual(len(self.s.list_filter_rules()), 1)
|
|
|
|
def test_events_by_host_window(self):
|
|
e1 = Event(event_id="e-1", host_id="h-1", metric="m", value=1.0, threshold=0.0, fired_at=BASE)
|
|
e2 = Event(event_id="e-2", host_id="h-2", metric="m", value=1.0, threshold=0.0, fired_at=BASE)
|
|
e3 = Event(
|
|
event_id="e-3",
|
|
host_id="h-1",
|
|
metric="m",
|
|
value=1.0,
|
|
threshold=0.0,
|
|
fired_at=datetime(2025, 2, 1, 10, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
for e in (e1, e2, e3):
|
|
self.s.save_event(e)
|
|
start = datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
|
|
end = datetime(2025, 1, 31, 23, 59, 59, tzinfo=timezone.utc)
|
|
got = self.s.list_events_by_host("h-1", start, end)
|
|
self.assertEqual([e.event_id for e in got], ["e-1"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|