Codex
71 lines
2.9 KiB
Python
71 lines
2.9 KiB
Python
import unittest
|
|
|
|
from src.services.visual_watch import (
|
|
VISUAL_TRIGGER_CANDIDATE_ACTIONABLE_WATCH,
|
|
VISUAL_TRIGGER_CANDIDATE_BLOCK_REASON_REVIEW,
|
|
VISUAL_TRIGGER_MARKET_ANOMALY_REVIEW,
|
|
VISUAL_TRIGGER_NEWS_WINDOW_REVIEW,
|
|
VISUAL_TRIGGER_POLICY_SKIPPED,
|
|
VISUAL_TRIGGER_POLICY_TRIGGERED,
|
|
VISUAL_TRIGGER_RECONCILIATION_REVIEW,
|
|
evaluate_visual_trigger_policy,
|
|
suggest_visual_trigger,
|
|
suggest_visual_triggers_from_council,
|
|
)
|
|
|
|
|
|
class VisualTriggerPolicyServiceTests(unittest.TestCase):
|
|
def test_observe_monitoring_skips_visual_review(self) -> None:
|
|
policy = evaluate_visual_trigger_policy(
|
|
action_recommendation="observe_only",
|
|
status="monitoring",
|
|
why_blocked=[],
|
|
)
|
|
|
|
self.assertEqual(policy["status"], VISUAL_TRIGGER_POLICY_SKIPPED)
|
|
self.assertFalse(policy["triggered"])
|
|
self.assertEqual(policy["trigger_types"], [])
|
|
self.assertEqual(policy["skip_reason"], "observe_monitoring_not_high_value")
|
|
self.assertIsNone(suggest_visual_trigger(action_recommendation="observe_only", status="monitoring"))
|
|
|
|
def test_actionable_candidate_triggers_high_value_watch(self) -> None:
|
|
policy = evaluate_visual_trigger_policy(
|
|
action_recommendation="confirm_ticket",
|
|
status="actionable",
|
|
why_blocked=[],
|
|
)
|
|
|
|
self.assertEqual(policy["status"], VISUAL_TRIGGER_POLICY_TRIGGERED)
|
|
self.assertTrue(policy["triggered"])
|
|
self.assertEqual(policy["primary_trigger_type"], VISUAL_TRIGGER_CANDIDATE_ACTIONABLE_WATCH)
|
|
self.assertTrue(policy["is_high_value"])
|
|
|
|
def test_degraded_market_and_broker_reasons_trigger_auditable_reviews(self) -> None:
|
|
policy = evaluate_visual_trigger_policy(
|
|
action_recommendation="do_not_execute",
|
|
status="blocked",
|
|
why_blocked=["market_stream_disconnected", "broker_read_transport_error"],
|
|
)
|
|
|
|
self.assertIn(VISUAL_TRIGGER_CANDIDATE_BLOCK_REASON_REVIEW, policy["trigger_types"])
|
|
self.assertIn(VISUAL_TRIGGER_MARKET_ANOMALY_REVIEW, policy["trigger_types"])
|
|
self.assertIn(VISUAL_TRIGGER_RECONCILIATION_REVIEW, policy["trigger_types"])
|
|
self.assertIn("market_anomaly_or_data_degraded", policy["reason_codes"])
|
|
self.assertIn("reconciliation_or_broker_degraded", policy["reason_codes"])
|
|
self.assertTrue(policy["is_high_value"])
|
|
|
|
def test_council_watch_items_map_to_visual_trigger_types(self) -> None:
|
|
triggers = suggest_visual_triggers_from_council(
|
|
action_recommendation="wait_for_confirmations",
|
|
status="blocked",
|
|
why_blocked=["mss_missing"],
|
|
recommended_watch_items=["news_window_review", "visual_review_missing"],
|
|
)
|
|
|
|
self.assertIn(VISUAL_TRIGGER_NEWS_WINDOW_REVIEW, triggers)
|
|
self.assertIn(VISUAL_TRIGGER_CANDIDATE_BLOCK_REASON_REVIEW, triggers)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|