Codex
250 lines
10 KiB
Python
250 lines
10 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
|
|
|
|
|
|
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_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 = []
|
|
|
|
def fetch_latest_delivery(self, **kwargs):
|
|
return None
|
|
|
|
def fetch_latest_delivery_by_dedupe_scope(self, **kwargs):
|
|
return None
|
|
|
|
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.html",
|
|
"snapshot_payload": {"artifact_path": "captures/watch-44.json"},
|
|
}
|
|
]
|
|
|
|
|
|
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.html",
|
|
"feedback_summary": {"count": 1, "latest_label": "visual_helpful"},
|
|
"visual_notes": "structure aligns",
|
|
},
|
|
)
|
|
|
|
self.assertIn("visual=confirm", message)
|
|
self.assertIn("snapshot=captures/watch-44.html", message)
|
|
self.assertIn("feedback=1:visual_helpful", 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(delivery_repository.delivery_calls[0]["payload"]["visual"]["visual_state"], "confirm")
|
|
self.assertIn("visual=confirm", 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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|