cnbugs
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from backend.history_store import HistoryStore
|
|
|
|
|
|
class HistoryStoreTest(unittest.TestCase):
|
|
def setUp(self):
|
|
fd, self.path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
self.store = HistoryStore(self.path)
|
|
|
|
def tearDown(self):
|
|
os.remove(self.path)
|
|
|
|
def test_persists_diagnosis_and_can_query_after_reopen(self):
|
|
diagnosis = {
|
|
"timestamp": "2026-07-25T12:00:00",
|
|
"namespace": "prod",
|
|
"score": 65,
|
|
"status": "warning",
|
|
"critical_count": 1,
|
|
"warning_count": 4,
|
|
"issues": [{"level": "critical", "message": "pod failed"}],
|
|
"checks": {},
|
|
}
|
|
record_id = self.store.save_diagnosis(diagnosis, "report text")
|
|
|
|
reopened = HistoryStore(self.path)
|
|
records = reopened.list_diagnoses()
|
|
detail = reopened.get_diagnosis(record_id)
|
|
|
|
self.assertEqual(len(records), 1)
|
|
self.assertEqual(records[0]["namespace"], "prod")
|
|
self.assertEqual(detail["diagnosis"]["score"], 65)
|
|
self.assertEqual(detail["report"], "report text")
|
|
|
|
def test_records_all_processing_events_in_order(self):
|
|
diagnosis_id = self.store.save_diagnosis(
|
|
{"timestamp": "2026-07-25T12:00:00", "namespace": "all", "score": 80, "status": "warning"},
|
|
"report",
|
|
)
|
|
run_id = self.store.create_run(diagnosis_id, "agent", "ai", "check pods")
|
|
self.store.append_event(run_id, "command", {"command": "kubectl get pods -A"})
|
|
self.store.append_event(run_id, "result", {"returncode": 0, "stdout": "ok"})
|
|
self.store.finish_run(run_id, "completed", "done")
|
|
|
|
detail = self.store.get_diagnosis(diagnosis_id)
|
|
|
|
self.assertEqual(len(detail["runs"]), 1)
|
|
self.assertEqual(detail["runs"][0]["status"], "completed")
|
|
self.assertEqual([event["type"] for event in detail["runs"][0]["events"]], ["command", "result"])
|
|
|
|
def test_lists_all_diagnoses_newest_first_and_filters_status(self):
|
|
first = {"timestamp": "2026-07-25T11:00:00", "namespace": "dev", "score": 100, "status": "healthy"}
|
|
second = {"timestamp": "2026-07-25T12:00:00", "namespace": "prod", "score": 20, "status": "critical"}
|
|
self.store.save_diagnosis(first, "first")
|
|
self.store.save_diagnosis(second, "second")
|
|
|
|
all_records = self.store.list_diagnoses()
|
|
critical = self.store.list_diagnoses(status="critical")
|
|
|
|
self.assertEqual([item["namespace"] for item in all_records], ["prod", "dev"])
|
|
self.assertEqual([item["namespace"] for item in critical], ["prod"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|