131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
import _bootstrap # noqa: F401
|
|
|
|
import json
|
|
import time
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
|
|
from fault_log_analyzer.api import FaultLogApiServer
|
|
from fault_log_analyzer.models import FaultLog, RootCause
|
|
from fault_log_analyzer.storage import (
|
|
InMemoryFaultLogRepository,
|
|
InMemoryFaultTypeRepository,
|
|
InMemoryFilterRuleRepository,
|
|
InMemoryRootCauseRepository,
|
|
)
|
|
|
|
|
|
class TestFaultLogApiServer(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.fl_repo = InMemoryFaultLogRepository()
|
|
cls.rc_repo = InMemoryRootCauseRepository()
|
|
cls.ft_repo = InMemoryFaultTypeRepository()
|
|
cls.fr_repo = InMemoryFilterRuleRepository()
|
|
|
|
cls.fl_repo.save(
|
|
FaultLog(
|
|
fault_log_id="fl-1",
|
|
host_id="h-1",
|
|
fingerprint="fp",
|
|
level="ERROR",
|
|
message="disk full",
|
|
occurred_at=datetime(2025, 1, 1, tzinfo=timezone.utc),
|
|
fault_type="disk_full",
|
|
)
|
|
)
|
|
cls.rc_repo.save(RootCause(fault_log_id="fl-1", cause_type="disk_full", confidence=0.9))
|
|
|
|
cls.server = FaultLogApiServer(
|
|
cls.fl_repo, cls.rc_repo, cls.ft_repo, cls.fr_repo, host="127.0.0.1", port=0
|
|
)
|
|
cls.server.start()
|
|
# 等待监听线程就绪
|
|
time.sleep(0.05)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.stop()
|
|
|
|
def _get(self, path):
|
|
with urllib.request.urlopen(f"http://127.0.0.1:{self.server.port}{path}", timeout=5) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
def _post(self, path, body):
|
|
req = urllib.request.Request(
|
|
f"http://127.0.0.1:{self.server.port}{path}",
|
|
data=json.dumps(body).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
def test_healthz(self):
|
|
data = self._get("/healthz")
|
|
self.assertEqual(data["code"], 0)
|
|
self.assertEqual(data["data"]["status"], "ok")
|
|
|
|
def test_readyz(self):
|
|
data = self._get("/readyz")
|
|
self.assertEqual(data["code"], 0)
|
|
self.assertIn("deps", data["data"])
|
|
|
|
def test_list_fault_logs(self):
|
|
data = self._get("/api/v1/fault-logs")
|
|
self.assertEqual(data["data"]["total"], 1)
|
|
self.assertEqual(data["data"]["items"][0]["fault_log_id"], "fl-1")
|
|
|
|
def test_list_fault_logs_filter(self):
|
|
data = self._get("/api/v1/fault-logs?host_id=h-1&level=ERROR")
|
|
self.assertEqual(data["data"]["total"], 1)
|
|
|
|
def test_get_fault_log(self):
|
|
data = self._get("/api/v1/fault-logs/fl-1")
|
|
self.assertEqual(data["data"]["fault_log_id"], "fl-1")
|
|
|
|
def test_get_fault_log_404(self):
|
|
with self.assertRaises(urllib.error.HTTPError) as ctx:
|
|
self._get("/api/v1/fault-logs/nope")
|
|
self.assertEqual(ctx.exception.code, 404)
|
|
|
|
def test_get_root_cause(self):
|
|
data = self._get("/api/v1/fault-logs/fl-1/root-cause")
|
|
self.assertEqual(data["data"]["cause_type"], "disk_full")
|
|
|
|
def test_list_fault_types(self):
|
|
data = self._get("/api/v1/fault-types")
|
|
self.assertEqual(data["code"], 0)
|
|
self.assertIn("items", data["data"])
|
|
|
|
def test_create_fault_type(self):
|
|
data = self._post("/api/v1/fault-types", {"fault_type": "oom", "name": "OOM"})
|
|
self.assertEqual(data["code"], 0)
|
|
self.assertEqual(data["data"]["fault_type"], "oom")
|
|
|
|
def test_create_fault_type_missing_fields(self):
|
|
with self.assertRaises(urllib.error.HTTPError) as ctx:
|
|
self._post("/api/v1/fault-types", {"name": "OOM"})
|
|
self.assertEqual(ctx.exception.code, 400)
|
|
|
|
def test_list_fault_filters(self):
|
|
data = self._get("/api/v1/fault-filters")
|
|
self.assertEqual(data["code"], 0)
|
|
|
|
def test_create_fault_filter(self):
|
|
data = self._post(
|
|
"/api/v1/fault-filters", {"name": "disk", "level": "ERROR", "pattern": "disk"}
|
|
)
|
|
self.assertEqual(data["code"], 0)
|
|
|
|
def test_unknown_route_404(self):
|
|
with self.assertRaises(urllib.error.HTTPError) as ctx:
|
|
self._get("/nope")
|
|
self.assertEqual(ctx.exception.code, 404)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|