Codex
495 lines
22 KiB
Python
495 lines
22 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from src.services.alerts import AlertDispatcherService
|
|
from src.services.alerts.message_template_service import build_alert_message
|
|
from src.services.feedback import AlertFeedbackService
|
|
from src.services.visual_watch import build_visual_media_evidence
|
|
|
|
|
|
class StubFeedbackRepository:
|
|
def __init__(self) -> None:
|
|
self.insert_calls = []
|
|
self.alert_rows = []
|
|
self.visual_rows = []
|
|
|
|
def insert_feedback(self, **kwargs):
|
|
self.insert_calls.append(kwargs)
|
|
return len(self.insert_calls) + 100
|
|
|
|
def fetch_feedback_for_alert(self, alert_id: int, limit: int = 20):
|
|
return [row for row in self.alert_rows if row.get("alert_id") == alert_id][:limit]
|
|
|
|
def fetch_feedback_for_visual_review(self, visual_review_id: int, limit: int = 20):
|
|
return [row for row in self.visual_rows if row.get("visual_review_id") == visual_review_id][:limit]
|
|
|
|
|
|
class StubDecisionRepository:
|
|
def __init__(self) -> None:
|
|
self.alerts = {}
|
|
self.packets = {}
|
|
self.status_updates = []
|
|
|
|
def fetch_pending_decision_alerts(self, limit: int = 50):
|
|
return [alert for alert in self.alerts.values() if str(alert.get("status") or "") == "queued"][:limit]
|
|
|
|
def fetch_decision_alert_by_id(self, alert_id: int):
|
|
return self.alerts.get(alert_id)
|
|
|
|
def fetch_decision_packet_by_id(self, packet_id: int):
|
|
return self.packets.get(packet_id)
|
|
|
|
def update_decision_alert_status(self, **kwargs):
|
|
self.status_updates.append(kwargs)
|
|
|
|
|
|
class StubVisualRepository:
|
|
def __init__(self) -> None:
|
|
self.reviews = {}
|
|
|
|
def fetch_visual_review_by_id(self, visual_review_id: int):
|
|
return self.reviews.get(visual_review_id)
|
|
|
|
|
|
class StubDeliveryRepository:
|
|
def __init__(self) -> None:
|
|
self.delivery_calls = []
|
|
self.failed_rows = []
|
|
self.recent_rows_by_alert_channel = {}
|
|
|
|
def fetch_latest_delivery(self, **kwargs):
|
|
return None
|
|
|
|
def fetch_latest_delivery_by_dedupe_scope(self, **kwargs):
|
|
return None
|
|
|
|
def fetch_failed_deliveries(self, limit: int = 100):
|
|
return list(self.failed_rows)[:limit]
|
|
|
|
def fetch_recent_deliveries_for_alert_channel(self, *, alert_id: int, channel: str, limit: int = 20):
|
|
return list(self.recent_rows_by_alert_channel.get((alert_id, channel), []))[:limit]
|
|
|
|
def insert_delivery(self, **kwargs):
|
|
self.delivery_calls.append(kwargs)
|
|
return len(self.delivery_calls) + 200
|
|
|
|
|
|
class StubVisualWatchService:
|
|
def fetch_reviews_for_packet(self, *, decision_packet_id: int, limit: int = 1):
|
|
return [
|
|
{
|
|
"id": 33,
|
|
"watch_run_id": 44,
|
|
"visual_state": "confirm",
|
|
"visual_confidence": 0.86,
|
|
"visual_notes": "structure aligns",
|
|
"snapshot_source_name": "auto_visual_watch_capture",
|
|
"snapshot_image_path": "captures/watch-44.png",
|
|
"snapshot_payload": {
|
|
"html_path": "captures/watch-44.html",
|
|
"artifact_path": "captures/watch-44.json",
|
|
"browser_capture": {"status": "captured"},
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
class StubFeedbackService:
|
|
def list_visual_review_feedback(self, visual_review_id: int, limit: int = 10):
|
|
return [{"visual_review_id": visual_review_id, "label": "visual_helpful", "created_ts": "2026-05-11T00:00:00+00:00"}]
|
|
|
|
def summarize_feedback(self, rows):
|
|
return {
|
|
"count": len(rows),
|
|
"labels": {"visual_helpful": len(rows)},
|
|
"latest_label": "visual_helpful" if rows else None,
|
|
"useful_count": len(rows),
|
|
"misleading_count": 0,
|
|
"actionable_count": 0,
|
|
}
|
|
|
|
|
|
class AlertFeedbackServiceTests(unittest.TestCase):
|
|
def test_records_alert_feedback_with_allowed_label(self) -> None:
|
|
repository = StubFeedbackRepository()
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[10] = {"id": 10, "status": "queued"}
|
|
service = AlertFeedbackService(
|
|
repository=repository,
|
|
decision_repository=decision_repository,
|
|
visual_repository=StubVisualRepository(),
|
|
)
|
|
|
|
result = service.record_alert_feedback(alert_id=10, label="useful", note="timely", created_by="operator")
|
|
|
|
self.assertEqual(result["status"], "created")
|
|
self.assertEqual(result["feedback_id"], 101)
|
|
self.assertEqual(repository.insert_calls[0]["subject_type"], "alert")
|
|
self.assertEqual(repository.insert_calls[0]["alert_id"], 10)
|
|
self.assertEqual(repository.insert_calls[0]["label"], "useful")
|
|
|
|
def test_records_visual_review_feedback(self) -> None:
|
|
repository = StubFeedbackRepository()
|
|
visual_repository = StubVisualRepository()
|
|
visual_repository.reviews[22] = {"id": 22, "visual_state": "confirm"}
|
|
service = AlertFeedbackService(
|
|
repository=repository,
|
|
decision_repository=StubDecisionRepository(),
|
|
visual_repository=visual_repository,
|
|
)
|
|
|
|
result = service.record_visual_review_feedback(visual_review_id=22, label="actionable")
|
|
|
|
self.assertEqual(result["status"], "created")
|
|
self.assertEqual(repository.insert_calls[0]["subject_type"], "visual_review")
|
|
self.assertEqual(repository.insert_calls[0]["visual_review_id"], 22)
|
|
self.assertEqual(repository.insert_calls[0]["label"], "actionable")
|
|
|
|
def test_rejects_unknown_label(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[10] = {"id": 10, "status": "queued"}
|
|
service = AlertFeedbackService(
|
|
repository=StubFeedbackRepository(),
|
|
decision_repository=decision_repository,
|
|
visual_repository=StubVisualRepository(),
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "invalid_feedback_label"):
|
|
service.record_alert_feedback(alert_id=10, label="rewrite_gate")
|
|
|
|
def test_summarizes_visual_feedback(self) -> None:
|
|
service = AlertFeedbackService(
|
|
repository=StubFeedbackRepository(),
|
|
decision_repository=StubDecisionRepository(),
|
|
visual_repository=StubVisualRepository(),
|
|
)
|
|
|
|
summary = service.summarize_feedback(
|
|
[
|
|
{"label": "visual_helpful", "created_ts": "2026-05-11T00:00:00+00:00"},
|
|
{"label": "visual_misleading", "created_ts": "2026-05-11T00:01:00+00:00"},
|
|
{"label": "actionable", "created_ts": "2026-05-11T00:02:00+00:00"},
|
|
]
|
|
)
|
|
|
|
self.assertEqual(summary["count"], 3)
|
|
self.assertEqual(summary["useful_count"], 1)
|
|
self.assertEqual(summary["misleading_count"], 1)
|
|
self.assertEqual(summary["actionable_count"], 1)
|
|
self.assertEqual(summary["latest_label"], "actionable")
|
|
|
|
|
|
class AlertVisualDispatchTests(unittest.TestCase):
|
|
def test_alert_message_includes_visual_context(self) -> None:
|
|
message = build_alert_message(
|
|
alert={"alert_type": "candidate_actionable", "payload": {"action_recommendation": "confirm_ticket", "status": "actionable"}},
|
|
packet={"payload": {"symbol": "BTC-USDT"}},
|
|
visual={
|
|
"visual_state": "confirm",
|
|
"visual_confidence": 0.86,
|
|
"snapshot_image_path": "captures/watch-44.png",
|
|
"feedback_summary": {"count": 1, "latest_label": "visual_helpful"},
|
|
"visual_notes": "structure aligns",
|
|
},
|
|
media_evidence=build_visual_media_evidence(
|
|
channel="webhook",
|
|
visual={"snapshot_image_path": "captures/watch-44.png"},
|
|
),
|
|
)
|
|
|
|
self.assertIn("visual=confirm", message)
|
|
self.assertIn("snapshot=captures/watch-44.png", message)
|
|
self.assertIn("feedback=1:visual_helpful", message)
|
|
self.assertIn("media=media_linked", message)
|
|
|
|
def test_dispatcher_attaches_visual_context_to_delivery_payload(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[9] = {
|
|
"id": 9,
|
|
"decision_packet_id": 55,
|
|
"alert_type": "candidate_actionable",
|
|
"status": "queued",
|
|
"payload": {"action_recommendation": "confirm_ticket", "status": "actionable"},
|
|
}
|
|
decision_repository.packets[55] = {"id": 55, "payload": {"symbol": "BTC-USDT"}}
|
|
delivery_repository = StubDeliveryRepository()
|
|
dispatcher = AlertDispatcherService(
|
|
decision_repository=decision_repository,
|
|
delivery_repository=delivery_repository,
|
|
visual_watch_service=StubVisualWatchService(),
|
|
feedback_service=StubFeedbackService(),
|
|
)
|
|
|
|
with patch("src.services.alerts.channel_router_service.settings.alert_default_channel", "workbench"), patch(
|
|
"src.services.alerts.channel_router_service.settings.alert_webhook_url",
|
|
"",
|
|
):
|
|
result = dispatcher.dispatch_alert(9)
|
|
|
|
self.assertEqual(result["status"], "skipped")
|
|
self.assertEqual(result["visual"]["visual_state"], "confirm")
|
|
self.assertEqual(result["media_status"], "media_attached")
|
|
self.assertEqual(result["media_evidence"]["primary_mime_type"], "image/png")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["payload"]["visual"]["visual_state"], "confirm")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["payload"]["media_evidence"]["status"], "media_attached")
|
|
self.assertIn("visual=confirm", delivery_repository.delivery_calls[0]["payload"]["message"])
|
|
self.assertIn("media=media_attached", delivery_repository.delivery_calls[0]["payload"]["message"])
|
|
|
|
def test_dispatcher_skips_webhook_when_silent_window_is_active(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[10] = {
|
|
"id": 10,
|
|
"decision_packet_id": 55,
|
|
"alert_type": "candidate_actionable",
|
|
"status": "queued",
|
|
"payload": {"action_recommendation": "confirm_ticket", "status": "actionable"},
|
|
}
|
|
decision_repository.packets[55] = {"id": 55, "payload": {"symbol": "BTC-USDT"}}
|
|
delivery_repository = StubDeliveryRepository()
|
|
silent_window_service = type(
|
|
"SilentWindowService",
|
|
(),
|
|
{
|
|
"build_status": lambda self, **kwargs: {"active": True, "window_label": "00:00-06:00"},
|
|
"is_active": lambda self, **kwargs: True,
|
|
},
|
|
)()
|
|
dispatcher = AlertDispatcherService(
|
|
decision_repository=decision_repository,
|
|
delivery_repository=delivery_repository,
|
|
visual_watch_service=StubVisualWatchService(),
|
|
feedback_service=StubFeedbackService(),
|
|
silent_window_service=silent_window_service,
|
|
)
|
|
|
|
with patch("src.services.alerts.channel_router_service.settings.alert_webhook_url", "https://example.test/hook"):
|
|
result = dispatcher.dispatch_alert(10)
|
|
|
|
self.assertEqual(result["status"], "skipped")
|
|
self.assertEqual(result["reason"], "alert_silent_window_active")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["error"], "alert_silent_window_active")
|
|
self.assertEqual(result["delivery_policy"]["action"], "suppressed_silent_window")
|
|
|
|
def test_dispatcher_suppresses_low_severity_external_alert_by_noise_policy(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[11] = {
|
|
"id": 11,
|
|
"decision_packet_id": 55,
|
|
"alert_type": "candidate_actionable",
|
|
"status": "queued",
|
|
"payload": {"action_recommendation": "confirm_ticket", "status": "actionable", "severity": "low"},
|
|
}
|
|
decision_repository.packets[55] = {"id": 55, "payload": {"symbol": "BTC-USDT"}}
|
|
delivery_repository = StubDeliveryRepository()
|
|
dispatcher = AlertDispatcherService(
|
|
decision_repository=decision_repository,
|
|
delivery_repository=delivery_repository,
|
|
visual_watch_service=StubVisualWatchService(),
|
|
feedback_service=StubFeedbackService(),
|
|
silent_window_service=type(
|
|
"SilentWindowService",
|
|
(),
|
|
{
|
|
"build_status": lambda self, **kwargs: {"active": False, "window_label": "00:00-06:00"},
|
|
"is_active": lambda self, **kwargs: False,
|
|
},
|
|
)(),
|
|
)
|
|
|
|
with patch("src.services.alerts.channel_router_service.settings.alert_webhook_url", "https://example.test/hook"), patch(
|
|
"src.services.alerts.delivery_policy_service.settings.alert_noise_external_min_severity",
|
|
"medium",
|
|
), patch.object(AlertDispatcherService, "_send") as send_mock:
|
|
result = dispatcher.dispatch_alert(11)
|
|
|
|
send_mock.assert_not_called()
|
|
self.assertEqual(result["status"], "skipped")
|
|
self.assertEqual(result["reason"], "suppressed_noise_policy")
|
|
self.assertEqual(result["delivery_policy"]["policy_rule"], "below_external_min_severity")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["error"], "suppressed_noise_policy")
|
|
self.assertEqual(
|
|
delivery_repository.delivery_calls[0]["payload"]["delivery_policy"]["action"],
|
|
"suppressed_noise_policy",
|
|
)
|
|
|
|
def test_dispatcher_records_disabled_channel_when_default_adapter_has_no_credentials(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[12] = {
|
|
"id": 12,
|
|
"decision_packet_id": 55,
|
|
"alert_type": "candidate_blocked",
|
|
"status": "queued",
|
|
"payload": {"action_recommendation": "wait_for_confirmations", "status": "blocked"},
|
|
}
|
|
decision_repository.packets[55] = {"id": 55, "payload": {"symbol": "BTC-USDT"}}
|
|
delivery_repository = StubDeliveryRepository()
|
|
dispatcher = AlertDispatcherService(
|
|
decision_repository=decision_repository,
|
|
delivery_repository=delivery_repository,
|
|
visual_watch_service=StubVisualWatchService(),
|
|
feedback_service=StubFeedbackService(),
|
|
)
|
|
|
|
with patch("src.services.alerts.channel_router_service.settings.alert_default_channel", "email"), patch(
|
|
"src.services.alerts.channel_router_service.settings.alert_email_smtp_url",
|
|
"",
|
|
), patch("src.services.alerts.channel_router_service.settings.alert_email_to", ""), patch.object(
|
|
AlertDispatcherService,
|
|
"_send",
|
|
) as send_mock:
|
|
result = dispatcher.dispatch_alert(12)
|
|
|
|
send_mock.assert_not_called()
|
|
self.assertEqual(result["status"], "skipped")
|
|
self.assertEqual(result["reason"], "channel_disabled")
|
|
self.assertEqual(result["channel"], "email")
|
|
self.assertEqual(result["channel_route"]["capability"]["reason"], "credential_missing")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["error"], "channel_disabled")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["payload"]["channel_route"]["selected_channel"], "email")
|
|
|
|
def test_dispatch_pending_retries_failed_alert_after_backoff(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[21] = {
|
|
"id": 21,
|
|
"decision_packet_id": 55,
|
|
"alert_type": "candidate_actionable",
|
|
"status": "failed",
|
|
"payload": {"action_recommendation": "confirm_ticket", "status": "actionable"},
|
|
}
|
|
decision_repository.packets[55] = {"id": 55, "payload": {"symbol": "BTC-USDT"}}
|
|
delivery_repository = StubDeliveryRepository()
|
|
delivery_repository.failed_rows = [
|
|
{"alert_id": 21, "status": "failed", "attempted_ts": "2026-05-11T00:00:00+00:00"}
|
|
]
|
|
delivery_repository.recent_rows_by_alert_channel[(21, "webhook")] = [
|
|
{"alert_id": 21, "channel": "webhook", "status": "failed", "attempted_ts": "2026-05-11T00:00:00+00:00"}
|
|
]
|
|
dispatcher = AlertDispatcherService(
|
|
decision_repository=decision_repository,
|
|
delivery_repository=delivery_repository,
|
|
visual_watch_service=StubVisualWatchService(),
|
|
feedback_service=StubFeedbackService(),
|
|
silent_window_service=type(
|
|
"SilentWindowService",
|
|
(),
|
|
{
|
|
"build_status": lambda self, **kwargs: {"active": False, "window_label": "00:00-06:00"},
|
|
"is_active": lambda self, **kwargs: False,
|
|
},
|
|
)(),
|
|
)
|
|
|
|
with patch("src.services.alerts.channel_router_service.settings.alert_webhook_url", "https://example.test/hook"), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_retry_failed_enabled",
|
|
True,
|
|
), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_retry_max_attempts",
|
|
3,
|
|
), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_retry_backoff_seconds",
|
|
60,
|
|
), patch.object(AlertDispatcherService, "_send", return_value={"external_message_id": "abc"}):
|
|
result = dispatcher.dispatch_pending(limit=5)
|
|
|
|
self.assertEqual(result["retry_candidates"], 1)
|
|
self.assertEqual(result["retried"], 1)
|
|
self.assertEqual(result["items"][0]["dispatch_mode"], "failed_retry")
|
|
self.assertEqual(result["items"][0]["status"], "delivered")
|
|
|
|
def test_dispatch_failed_alert_defers_when_retry_backoff_is_active(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[22] = {
|
|
"id": 22,
|
|
"decision_packet_id": 55,
|
|
"alert_type": "candidate_actionable",
|
|
"status": "failed",
|
|
"payload": {"action_recommendation": "confirm_ticket", "status": "actionable"},
|
|
}
|
|
decision_repository.packets[55] = {"id": 55, "payload": {"symbol": "BTC-USDT"}}
|
|
delivery_repository = StubDeliveryRepository()
|
|
delivery_repository.recent_rows_by_alert_channel[(22, "webhook")] = [
|
|
{"alert_id": 22, "channel": "webhook", "status": "failed", "attempted_ts": "2999-05-11T00:00:00+00:00"}
|
|
]
|
|
dispatcher = AlertDispatcherService(
|
|
decision_repository=decision_repository,
|
|
delivery_repository=delivery_repository,
|
|
visual_watch_service=StubVisualWatchService(),
|
|
feedback_service=StubFeedbackService(),
|
|
silent_window_service=type(
|
|
"SilentWindowService",
|
|
(),
|
|
{
|
|
"build_status": lambda self, **kwargs: {"active": False, "window_label": "00:00-06:00"},
|
|
"is_active": lambda self, **kwargs: False,
|
|
},
|
|
)(),
|
|
)
|
|
|
|
with patch("src.services.alerts.channel_router_service.settings.alert_webhook_url", "https://example.test/hook"), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_retry_failed_enabled",
|
|
True,
|
|
), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_retry_max_attempts",
|
|
3,
|
|
), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_retry_backoff_seconds",
|
|
60,
|
|
):
|
|
result = dispatcher.dispatch_alert(22)
|
|
|
|
self.assertEqual(result["status"], "deferred")
|
|
self.assertEqual(result["reason"], "retry_backoff_active")
|
|
self.assertEqual(delivery_repository.delivery_calls, [])
|
|
|
|
def test_dispatcher_retries_webhook_transport_before_success(self) -> None:
|
|
decision_repository = StubDecisionRepository()
|
|
decision_repository.alerts[30] = {
|
|
"id": 30,
|
|
"decision_packet_id": 55,
|
|
"alert_type": "candidate_actionable",
|
|
"status": "queued",
|
|
"payload": {"action_recommendation": "confirm_ticket", "status": "actionable"},
|
|
}
|
|
decision_repository.packets[55] = {"id": 55, "payload": {"symbol": "BTC-USDT"}}
|
|
delivery_repository = StubDeliveryRepository()
|
|
dispatcher = AlertDispatcherService(
|
|
decision_repository=decision_repository,
|
|
delivery_repository=delivery_repository,
|
|
visual_watch_service=StubVisualWatchService(),
|
|
feedback_service=StubFeedbackService(),
|
|
)
|
|
|
|
class _FailOnceNotifier:
|
|
calls = 0
|
|
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
def send_text(self, text: str):
|
|
type(self).calls += 1
|
|
if type(self).calls == 1:
|
|
raise RuntimeError("temporary transport error")
|
|
return {"external_message_id": "msg-1"}
|
|
|
|
with patch("src.services.alerts.channel_router_service.settings.alert_webhook_url", "https://example.test/hook"), patch(
|
|
"src.services.alerts.alert_dispatcher_service.WebhookNotifier",
|
|
_FailOnceNotifier,
|
|
), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_webhook_transport_retries",
|
|
2,
|
|
), patch(
|
|
"src.services.alerts.alert_dispatcher_service.settings.alert_webhook_retry_backoff_seconds",
|
|
0.0,
|
|
):
|
|
result = dispatcher.dispatch_alert(30)
|
|
|
|
self.assertEqual(result["status"], "delivered")
|
|
self.assertEqual(result["transport_attempt_count"], 2)
|
|
self.assertEqual(result["media_status"], "media_linked")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["payload"]["media_evidence"]["status"], "media_linked")
|
|
self.assertEqual(delivery_repository.delivery_calls[0]["payload"]["result"]["attempt_count"], 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|