ai-exchange/tests/unit/test_operator_digest_service.py

import unittest

from sqlalchemy.exc import OperationalError

from src.services.alerts import OperatorDigestService


class _DecisionRepository:
    def __init__(self, alerts=None) -> None:
        self.alerts = alerts if alerts is not None else [
            {
                "id": 1,
                "alert_type": "candidate_actionable",
                "status": "queued",
                "channel": "workbench",
                "created_ts": "2026-05-12T01:00:00+00:00",
                "payload": {"action_recommendation": "confirm_ticket"},
            },
            {
                "id": 2,
                "alert_type": "observe",
                "status": "queued",
                "channel": "workbench",
                "created_ts": "2026-05-12T00:59:00+00:00",
                "payload": {
                    "visual_trigger_policy": {"status": "triggered"},
                    "visual_trigger_types": ["market_degradation_review"],
                },
            },
            {
                "id": 3,
                "alert_type": "observe",
                "status": "queued",
                "channel": "workbench",
                "created_ts": "2026-05-12T00:58:00+00:00",
                "payload": {},
            },
        ]

    def fetch_recent_decision_alerts(self, *, instrument_id: int, limit: int = 50):
        return self.alerts[:limit]


class _DeliveryRepository:
    def __init__(self, deliveries=None) -> None:
        self.deliveries = deliveries if deliveries is not None else [
            {
                "id": 10,
                "alert_id": 1,
                "channel": "workbench",
                "status": "skipped",
                "attempted_ts": "2026-05-12T01:01:00+00:00",
                "payload": {"reason": "suppressed_noise_policy"},
            },
            {
                "id": 11,
                "alert_id": 2,
                "channel": "webhook",
                "status": "failed",
                "attempted_ts": "2026-05-12T01:00:30+00:00",
                "error": "transport_error",
                "payload": {},
            },
            {
                "id": 12,
                "alert_id": 2,
                "channel": "webhook",
                "status": "delivered",
                "attempted_ts": "2026-05-12T01:00:20+00:00",
                "payload": {"dispatch_mode": "failed_retry"},
            },
        ]

    def fetch_recent_deliveries_for_alerts(self, alert_ids, limit: int = 200):
        return self.deliveries[:limit]


class _FeedbackRepository:
    def fetch_feedback_for_alert_ids(self, alert_ids, limit: int = 200):
        return [{"alert_id": 1, "label": "useful", "created_ts": "2026-05-12T01:02:00+00:00"}]

    def fetch_feedback_for_visual_review_ids(self, visual_review_ids, limit: int = 200):
        return [{"visual_review_id": 32, "label": "visual_helpful", "created_ts": "2026-05-12T01:03:00+00:00"}]


class _VisualRepository:
    def __init__(self, reviews=None) -> None:
        self.reviews = reviews if reviews is not None else [
            {
                "id": 31,
                "watch_run_id": 21,
                "trigger_type": "market_degradation_review",
                "visual_state": "mismatch",
                "visual_confidence": 0.81,
                "reviewed_ts": "2026-05-12T01:04:00+00:00",
            },
            {
                "id": 32,
                "watch_run_id": 22,
                "trigger_type": "candidate_actionable_watch",
                "visual_state": "confirm",
                "visual_confidence": 0.92,
                "reviewed_ts": "2026-05-12T01:03:00+00:00",
            },
        ]

    def fetch_recent_reviews_for_instrument(self, instrument_id: int, limit: int = 50):
        return self.reviews[:limit]


class _QualityService:
    def __init__(self, warnings=None) -> None:
        self.warnings = warnings or []

    def build_summary(self, *, instrument_id: int, alert_limit: int = 50, visual_review_limit: int = 50):
        return {
            "instrument_id": instrument_id,
            "quality_flags": ["visual_mismatch_present"],
            "warnings": self.warnings,
        }


class OperatorDigestServiceTests(unittest.TestCase):
    def test_build_digest_prioritizes_operator_review_queue(self) -> None:
        service = OperatorDigestService(
            decision_repository=_DecisionRepository(),
            delivery_repository=_DeliveryRepository(),
            feedback_repository=_FeedbackRepository(),
            visual_repository=_VisualRepository(),
            quality_service=_QualityService(),
        )

        digest = service.build_digest(instrument_id=1, alert_limit=10, visual_review_limit=10, top_limit=6)

        self.assertEqual(digest["status"], "completed")
        self.assertEqual(digest["summary_counts"]["high_value_alert_count"], 2)
        self.assertEqual(digest["summary_counts"]["suppressed_delivery_count"], 1)
        self.assertEqual(digest["summary_counts"]["failed_or_retry_delivery_count"], 2)
        self.assertEqual(digest["summary_counts"]["visual_mismatch_count"], 1)
        self.assertEqual(digest["summary_counts"]["pending_feedback_count"], 3)
        self.assertIn("visual_mismatch_present", digest["quality_flags"])
        self.assertEqual(digest["review_queue"][0]["queue_type"], "failed_retry")
        self.assertIn("visual_mismatch", [item["queue_type"] for item in digest["review_queue"]])
        self.assertIn("pending feedback=3", digest["operator_summary"])

    def test_build_digest_is_idle_when_no_recent_data_exists(self) -> None:
        service = OperatorDigestService(
            decision_repository=_DecisionRepository(alerts=[]),
            delivery_repository=_DeliveryRepository(deliveries=[]),
            feedback_repository=_FeedbackRepository(),
            visual_repository=_VisualRepository(reviews=[]),
            quality_service=_QualityService(),
        )

        digest = service.build_digest(instrument_id=1)

        self.assertEqual(digest["status"], "idle")
        self.assertEqual(digest["summary_counts"]["recent_alert_count"], 0)
        self.assertEqual(digest["review_queue"], [])
        self.assertIn("idle", digest["operator_summary"])

    def test_build_digest_degrades_optional_feedback_tables(self) -> None:
        class _BrokenFeedbackRepository:
            def fetch_feedback_for_alert_ids(self, alert_ids, limit: int = 200):
                raise OperationalError("stmt", {}, Exception())

            def fetch_feedback_for_visual_review_ids(self, visual_review_ids, limit: int = 200):
                raise OperationalError("stmt", {}, Exception())

        service = OperatorDigestService(
            decision_repository=_DecisionRepository(),
            delivery_repository=_DeliveryRepository(),
            feedback_repository=_BrokenFeedbackRepository(),
            visual_repository=_VisualRepository(),
            quality_service=_QualityService(warnings=["quality_feedback_unavailable"]),
        )

        digest = service.build_digest(instrument_id=1)

        self.assertIn("alert_feedback_unavailable", digest["warnings"])
        self.assertIn("visual_feedback_unavailable", digest["warnings"])
        self.assertIn("quality_feedback_unavailable", digest["warnings"])


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