Codex
103 lines
4.4 KiB
Python
103 lines
4.4 KiB
Python
import unittest
|
|
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
from src.services.runtime.runtime_incident_journal_service import RuntimeIncidentJournalService
|
|
|
|
|
|
class _IncidentRepository:
|
|
def __init__(self) -> None:
|
|
self.rows_by_key = {}
|
|
self.upserts = []
|
|
|
|
def upsert_incident(self, **kwargs):
|
|
self.upserts.append(kwargs)
|
|
row = dict(kwargs)
|
|
existing = self.rows_by_key.get(kwargs["incident_key"])
|
|
row["id"] = existing.get("id", len(self.rows_by_key) + 1) if existing else len(self.rows_by_key) + 1
|
|
row["status"] = "open"
|
|
row["first_seen_ts"] = existing.get("first_seen_ts", kwargs["observed_ts"]) if existing else kwargs["observed_ts"]
|
|
row["last_seen_ts"] = kwargs["observed_ts"]
|
|
row["occurrence_count"] = int(existing.get("occurrence_count", 0)) + 1 if existing else 1
|
|
self.rows_by_key[kwargs["incident_key"]] = row
|
|
return dict(row)
|
|
|
|
def fetch_recent_incidents(self, *, limit: int = 50, status=None):
|
|
rows = sorted(self.rows_by_key.values(), key=lambda item: str(item["last_seen_ts"]), reverse=True)
|
|
return [dict(row) for row in rows[:limit]]
|
|
|
|
|
|
class RuntimeIncidentJournalServiceTests(unittest.TestCase):
|
|
def test_build_from_runtime_status_persists_deduped_incidents_with_actions(self) -> None:
|
|
repository = _IncidentRepository()
|
|
service = RuntimeIncidentJournalService(repository=repository)
|
|
runtime_status = {
|
|
"current_blockers": ["market_stream_degraded:BTC-USDT:market_stream_disconnected"],
|
|
"current_warnings": [],
|
|
"current_noisy_incidents": ["watch_skipped_market_degraded:BTC-USDT"],
|
|
"incidents": [
|
|
{
|
|
"task_name": "market_state",
|
|
"symbol": "BTC-USDT",
|
|
"code": "market_stream_degraded:BTC-USDT:market_stream_disconnected",
|
|
"severity": "blocker",
|
|
"observed_ts": "2026-05-12T01:00:00+00:00",
|
|
}
|
|
],
|
|
"runtime_watchdog": {
|
|
"checked_ts": "2026-05-12T01:01:00+00:00",
|
|
"stalled_tasks": [{"name": "watch", "reason": "running_too_long", "age_seconds": 300}],
|
|
},
|
|
}
|
|
|
|
summary = service.build_from_runtime_status(runtime_status, limit=10, persist=True)
|
|
|
|
self.assertEqual(summary["status"], "completed")
|
|
self.assertTrue(summary["persisted"])
|
|
self.assertEqual(len(repository.upserts), 3)
|
|
self.assertEqual(summary["severity_counts"]["blocker"], 1)
|
|
self.assertEqual(summary["severity_counts"]["warning"], 1)
|
|
self.assertEqual(summary["severity_counts"]["noisy_incident"], 1)
|
|
self.assertIn("check_market_truth_stream_proxy_and_backfill", summary["recommended_actions"])
|
|
self.assertIn("inspect_runtime_task_and_restart_if_stale", summary["recommended_actions"])
|
|
first = summary["items"][0]
|
|
self.assertIn("severity", first)
|
|
self.assertIn("source", first)
|
|
self.assertIn("reason", first)
|
|
self.assertIn("first_seen_ts", first)
|
|
self.assertIn("last_seen_ts", first)
|
|
self.assertIn("recommended_action", first)
|
|
|
|
def test_build_from_runtime_status_is_idle_without_incidents(self) -> None:
|
|
service = RuntimeIncidentJournalService(repository=_IncidentRepository())
|
|
|
|
summary = service.build_from_runtime_status({"incidents": [], "runtime_watchdog": {}}, persist=True)
|
|
|
|
self.assertEqual(summary["status"], "idle")
|
|
self.assertEqual(summary["incident_count"], 0)
|
|
self.assertEqual(summary["items"], [])
|
|
|
|
def test_build_from_runtime_status_reports_repository_unavailable(self) -> None:
|
|
class _BrokenRepository:
|
|
def upsert_incident(self, **kwargs):
|
|
raise OperationalError("stmt", {}, Exception())
|
|
|
|
def fetch_recent_incidents(self, *, limit: int = 50, status=None):
|
|
return []
|
|
|
|
service = RuntimeIncidentJournalService(repository=_BrokenRepository())
|
|
|
|
summary = service.build_from_runtime_status(
|
|
{"current_blockers": ["db_unavailable:runtime"], "incidents": []},
|
|
persist=True,
|
|
)
|
|
|
|
self.assertEqual(summary["status"], "unavailable")
|
|
self.assertFalse(summary["persisted"])
|
|
self.assertEqual(summary["volatile_incident_count"], 1)
|
|
self.assertIn("runtime_incident_journal_unavailable", summary["warnings"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|