ai-exchange/tests/unit/test_maintenance_summary_service.py

import unittest

from src.services.maintenance import MaintenanceSummaryService


class _CaptureRetentionService:
    def __init__(self, payload=None) -> None:
        self.payload = payload or {
            "status": "completed",
            "eligible_file_count": 2,
            "deleted_file_count": 2,
            "deleted_bytes": 100,
        }
        self.calls = []

    def cleanup_captures(self, **kwargs):
        self.calls.append(kwargs)
        return dict(self.payload)


class _AlertQualityService:
    def __init__(self, payload=None) -> None:
        self.payload = payload or {
            "delivery_summary": {
                "failed_count": 1,
                "retry_exhausted_skip_count": 0,
                "channel_disabled_count": 1,
            },
            "alert_feedback_summary": {"count": 0},
            "visual_feedback_summary": {"count": 0},
            "quality_flags": ["feedback_sample_missing"],
        }

    def build_summary(self, **kwargs):
        return dict(self.payload)


class _IncidentJournalService:
    def __init__(self, payload=None) -> None:
        self.payload = payload or {
            "status": "completed",
            "incident_count": 1,
            "severity_counts": {"blocker": 1},
            "recommended_actions": ["check_market_truth_stream_proxy_and_backfill"],
        }

    def build_summary(self, *, limit: int = 50):
        return dict(self.payload)


class MaintenanceSummaryServiceTests(unittest.TestCase):
    def test_build_summary_rolls_up_storage_delivery_feedback_and_runtime(self) -> None:
        capture_service = _CaptureRetentionService()
        service = MaintenanceSummaryService(
            capture_retention_service=capture_service,
            alert_quality_service=_AlertQualityService(),
            incident_journal_service=_IncidentJournalService(),
            runtime_status_provider=lambda: {
                "current_blockers": ["market_stream_degraded:BTC-USDT:market_stream_disconnected"],
                "current_warnings": [],
                "runtime_watchdog": {"stalled_task_count": 1},
                "runtime_health": {"level": "blocked"},
            },
        )

        summary = service.build_summary(instrument_id=1)

        self.assertEqual(summary["status"], "blocked")
        self.assertTrue(capture_service.calls[0]["dry_run"])
        self.assertIn("storage", summary)
        self.assertIn("delivery", summary)
        self.assertIn("feedback", summary)
        self.assertIn("runtime", summary)
        self.assertIn("visual_capture_retention_cleanup", summary["maintenance_actions"]["automatic_actions"])
        self.assertIn("inspect_alert_delivery_failures", summary["maintenance_actions"]["operator_actions"])
        self.assertIn("check_market_truth_stream_proxy_and_backfill", summary["maintenance_actions"]["operator_actions"])
        self.assertTrue(summary["p23_assertions"]["storage_visible"])
        self.assertTrue(summary["p23_assertions"]["no_unscoped_deletion"])
        self.assertTrue(summary["p23_assertions"]["retention_policy_dry_run_only"])

    def test_build_summary_is_healthy_when_no_maintenance_findings_exist(self) -> None:
        service = MaintenanceSummaryService(
            capture_retention_service=_CaptureRetentionService(
                {"status": "completed", "eligible_file_count": 0, "deleted_file_count": 0}
            ),
            alert_quality_service=_AlertQualityService(
                {
                    "delivery_summary": {"failed_count": 0, "retry_exhausted_skip_count": 0, "channel_disabled_count": 0},
                    "alert_feedback_summary": {"count": 1},
                    "visual_feedback_summary": {"count": 1},
                    "quality_flags": [],
                }
            ),
            incident_journal_service=_IncidentJournalService(
                {"status": "idle", "incident_count": 0, "severity_counts": {}, "recommended_actions": []}
            ),
            runtime_status_provider=lambda: {
                "current_blockers": [],
                "current_warnings": [],
                "runtime_watchdog": {"stalled_task_count": 0},
                "runtime_health": {"level": "ok"},
            },
        )

        summary = service.build_summary(instrument_id=1)

        self.assertEqual(summary["status"], "healthy")
        self.assertEqual(summary["blockers"], [])
        self.assertEqual(summary["maintenance_actions"]["automatic_actions"], [])

    def test_build_summary_suppresses_stream_blocker_when_rest_market_is_fresh(self) -> None:
        service = MaintenanceSummaryService(
            capture_retention_service=_CaptureRetentionService(
                {"status": "completed", "eligible_file_count": 0, "deleted_file_count": 0}
            ),
            alert_quality_service=_AlertQualityService(
                {
                    "delivery_summary": {"failed_count": 0, "retry_exhausted_skip_count": 0, "channel_disabled_count": 0},
                    "alert_feedback_summary": {"count": 1},
                    "visual_feedback_summary": {"count": 1},
                    "quality_flags": [],
                }
            ),
            incident_journal_service=_IncidentJournalService(
                {"status": "idle", "incident_count": 0, "severity_counts": {}, "recommended_actions": []}
            ),
            runtime_status_provider=lambda: {
                "current_blockers": ["market_stream_degraded:BTC-USDT:market_stream_connecting"],
                "current_warnings": [],
                "runtime_watchdog": {"stalled_task_count": 0},
                "runtime_health": {"level": "blocked"},
                "latest_market_state_by_symbol": {
                    "BTC-USDT": {
                        "freshness": {"code": "market_data_fresh"},
                        "data_gap_state": {"code": "market_data_gap_clear"},
                    }
                },
            },
        )

        summary = service.build_summary(instrument_id=1)

        self.assertEqual(summary["status"], "healthy")
        self.assertEqual(summary["runtime"]["current_blockers"], [])
        self.assertEqual(summary["runtime"]["suppressed_blockers"], ["market_stream_degraded:BTC-USDT:market_stream_connecting"])

    def test_build_summary_suppresses_broker_sync_blocker_for_readonly(self) -> None:
        service = MaintenanceSummaryService(
            capture_retention_service=_CaptureRetentionService(
                {"status": "completed", "eligible_file_count": 0, "deleted_file_count": 0}
            ),
            alert_quality_service=_AlertQualityService(
                {
                    "delivery_summary": {"failed_count": 0, "retry_exhausted_skip_count": 0, "channel_disabled_count": 0},
                    "alert_feedback_summary": {"count": 1},
                    "visual_feedback_summary": {"count": 1},
                    "quality_flags": [],
                }
            ),
            incident_journal_service=_IncidentJournalService(
                {"status": "idle", "incident_count": 0, "severity_counts": {}, "recommended_actions": []}
            ),
            runtime_status_provider=lambda: {
                "current_blockers": ["broker_sync_degraded:BTC-USDT:okx_read_credentials_missing"],
                "current_warnings": [],
                "runtime_watchdog": {"stalled_task_count": 0},
                "runtime_health": {"level": "blocked"},
            },
        )

        summary = service.build_summary(instrument_id=1)

        self.assertEqual(summary["status"], "healthy")
        self.assertEqual(summary["runtime"]["current_blockers"], [])
        self.assertEqual(summary["runtime"]["suppressed_blockers"], ["broker_sync_degraded:BTC-USDT:okx_read_credentials_missing"])


if __name__ == "__main__":
    unittest.main()