import contextlib import io import json import os import unittest from types import SimpleNamespace from unittest.mock import patch from sqlalchemy.exc import OperationalError os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") from src.api import route_request from scripts import run_execution, run_execution_settlement, run_offline_checks, run_pipeline class ApiServerRoutingTests(unittest.TestCase): def test_routes_trading_plan_generate(self) -> None: status, payload = route_request( "/trading/plans/generate", {}, method="POST", body={"instrument_id": 1, "activate": True}, dependencies={ "trading_plan_generate": lambda **kwargs: {"id": 1, "status": "active", **kwargs}, }, ) self.assertEqual(status, 200) self.assertEqual(payload["instrument_id"], 1) self.assertTrue(payload["activate"]) def test_routes_trading_plan_latest(self) -> None: status, payload = route_request( "/trading/plans/latest", {"instrument_id": ["1"]}, dependencies={"trading_plan_latest": lambda **kwargs: {"id": 9, "status": "active"}}, ) self.assertEqual(status, 200) self.assertEqual(payload["id"], 9) def test_routes_unattended_harness(self) -> None: status, payload = route_request( "/trading/unattended-harness", {"instrument_id": ["2"]}, dependencies={"unattended_harness": lambda **kwargs: {"status": "passed", "instrument_id": kwargs["instrument_id"]}}, ) self.assertEqual(status, 200) self.assertEqual(payload["status"], "passed") self.assertEqual(payload["instrument_id"], 2) def test_routes_trading_market_state(self) -> None: status, payload = route_request( "/trading/market-state", {"instrument_id": ["1"], "timeframe": ["5m"]}, dependencies={"trading_market_state": lambda **kwargs: {"snapshot_id": 1, "market_state": kwargs}}, ) self.assertEqual(status, 200) self.assertEqual(payload["snapshot_id"], 1) self.assertEqual(payload["market_state"]["timeframe"], "5m") def test_routes_trading_decision_packet_and_alerts(self) -> None: status_packet, payload_packet = route_request( "/trading/decision-packet", {"instrument_id": ["1"]}, dependencies={"trading_decision_packet": lambda **kwargs: {"status": "actionable", "action_recommendation": "confirm_ticket"}}, ) status_alerts, payload_alerts = route_request( "/trading/alerts", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={"trading_decision_alerts": lambda **kwargs: [{"id": 1, "alert_type": "candidate_actionable"}]}, ) self.assertEqual(status_packet, 200) self.assertEqual(payload_packet["action_recommendation"], "confirm_ticket") self.assertEqual(status_alerts, 200) self.assertEqual(payload_alerts["items"], [{"id": 1, "alert_type": "candidate_actionable"}]) def test_routes_trading_backtests(self) -> None: captured = {} def fake_backtests(**kwargs): captured.update(kwargs) return { "instrument_id": kwargs["instrument_id"], "latest": {"id": 7, "summary": {"quality_gate": {"passed": True}}}, "quality_gate": {"passed": True}, "items": [{"id": 7}], } status, payload = route_request( "/trading/backtests", {"instrument_id": ["1"], "limit": ["3"]}, dependencies={"trading_backtests": fake_backtests}, ) self.assertEqual(status, 200) self.assertEqual(captured, {"instrument_id": 1, "limit": 3}) self.assertEqual(payload["latest"]["id"], 7) self.assertTrue(payload["quality_gate"]["passed"]) def test_routes_trading_doctrine_playbooks(self) -> None: status, payload = route_request( "/trading/doctrine-playbooks", {}, dependencies={ "trading_doctrine_playbooks": lambda: { "phase": "D2_PLAYBOOK_BACKTEST_COHORT", "playbook_count": 1, } }, ) self.assertEqual(status, 200) self.assertEqual(payload["phase"], "D2_PLAYBOOK_BACKTEST_COHORT") self.assertEqual(payload["playbook_count"], 1) def test_routes_trading_liquidity_kernel(self) -> None: status, payload = route_request( "/trading/liquidity-kernel", {}, dependencies={ "trading_liquidity_kernel": lambda: { "phase": "D3_LIQUIDITY_KERNEL", "rule_count": 6, } }, ) self.assertEqual(status, 200) self.assertEqual(payload["phase"], "D3_LIQUIDITY_KERNEL") self.assertEqual(payload["rule_count"], 6) def test_routes_trading_pd_array_kernel(self) -> None: status, payload = route_request( "/trading/pd-array-kernel", {}, dependencies={ "trading_pd_array_kernel": lambda: { "phase": "D4_PD_ARRAY_REACTION_KERNEL", "rule_count": 5, } }, ) self.assertEqual(status, 200) self.assertEqual(payload["phase"], "D4_PD_ARRAY_REACTION_KERNEL") self.assertEqual(payload["rule_count"], 5) def test_routes_trading_time_session_kernel(self) -> None: status, payload = route_request( "/trading/time-session-kernel", {}, dependencies={ "trading_time_session_kernel": lambda: { "phase": "D5_TIME_SESSION_KERNEL", "rule_count": 6, } }, ) self.assertEqual(status, 200) self.assertEqual(payload["phase"], "D5_TIME_SESSION_KERNEL") self.assertEqual(payload["rule_count"], 6) def test_routes_trading_model_playbook_kernel(self) -> None: status, payload = route_request( "/trading/model-playbook-kernel", {}, dependencies={ "trading_model_playbook_kernel": lambda: { "phase": "D6_MODEL_PLAYBOOK_KERNEL", "rule_count": 6, } }, ) self.assertEqual(status, 200) self.assertEqual(payload["phase"], "D6_MODEL_PLAYBOOK_KERNEL") self.assertEqual(payload["rule_count"], 6) def test_routes_trading_alert_quality_summary(self) -> None: status, payload = route_request( "/trading/alerts/quality-summary", {"instrument_id": ["1"], "limit": ["25"], "visual_limit": ["15"]}, dependencies={ "trading_alert_quality_summary": lambda **kwargs: { "instrument_id": kwargs["instrument_id"], "recent_alert_count": kwargs["limit"], "recent_visual_review_count": kwargs["visual_limit"], } }, ) self.assertEqual(status, 200) self.assertEqual(payload["instrument_id"], 1) self.assertEqual(payload["recent_alert_count"], 25) self.assertEqual(payload["recent_visual_review_count"], 15) def test_routes_trading_alert_operator_digest(self) -> None: status, payload = route_request( "/trading/alerts/operator-digest", {"instrument_id": ["1"], "limit": ["25"], "visual_limit": ["15"], "top_limit": ["3"]}, dependencies={ "trading_alert_operator_digest": lambda **kwargs: { "instrument_id": kwargs["instrument_id"], "alert_limit": kwargs["limit"], "visual_limit": kwargs["visual_limit"], "top_limit": kwargs["top_limit"], "review_queue": [{"queue_type": "failed_retry"}], } }, ) self.assertEqual(status, 200) self.assertEqual(payload["instrument_id"], 1) self.assertEqual(payload["alert_limit"], 25) self.assertEqual(payload["visual_limit"], 15) self.assertEqual(payload["top_limit"], 3) self.assertEqual(payload["review_queue"][0]["queue_type"], "failed_retry") def test_routes_trading_alert_digest_preferences(self) -> None: get_status, get_payload = route_request( "/trading/alerts/digest-preferences", {"instrument_id": ["1"], "profile_key": ["local_operator"]}, dependencies={ "trading_alert_operator_digest_preferences": lambda **kwargs: { "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "digest_mode": "high_value_only", } }, ) post_status, post_payload = route_request( "/trading/alerts/digest-preferences", {}, method="POST", body={ "instrument_id": 1, "profile_key": "local_operator", "digest_mode": "risk_first", "preferred_history_kind": "reminder", "importance_threshold": 80, }, dependencies={ "trading_alert_operator_digest_preferences_save": lambda **kwargs: { "instrument_id": kwargs["instrument_id"], "digest_mode": kwargs["digest_mode"], "preferred_history_kind": kwargs["preferred_history_kind"], "importance_threshold": kwargs["importance_threshold"], "status": "saved", } }, ) self.assertEqual(get_status, 200) self.assertEqual(get_payload["digest_mode"], "high_value_only") self.assertEqual(post_status, 200) self.assertEqual(post_payload["digest_mode"], "risk_first") self.assertEqual(post_payload["importance_threshold"], 80) def test_routes_archive_baselines(self) -> None: get_status, get_payload = route_request( "/trading/archive-baselines", {"instrument_id": ["1"], "profile_key": ["local_operator"]}, dependencies={ "trading_archive_baselines": lambda **kwargs: { "status": "active", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "digest_snapshot_id": 21, "workflow_snapshot_id": 34, } }, ) post_status, post_payload = route_request( "/trading/archive-baselines", {}, method="POST", body={ "instrument_id": 1, "profile_key": "local_operator", "digest_snapshot_id": 21, "workflow_snapshot_id": 34, }, dependencies={ "trading_archive_baselines_save": lambda **kwargs: { "status": "saved", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "digest_snapshot_id": kwargs["digest_snapshot_id"], "workflow_snapshot_id": kwargs["workflow_snapshot_id"], } }, ) self.assertEqual(get_status, 200) self.assertEqual(get_payload["digest_snapshot_id"], 21) self.assertEqual(post_status, 200) self.assertEqual(post_payload["workflow_snapshot_id"], 34) def test_routes_external_context_surface_history(self) -> None: get_status, get_payload = route_request( "/trading/external-context-surface-history", {"instrument_id": ["1"], "profile_key": ["local_operator"]}, dependencies={ "trading_external_context_surface_history": lambda **kwargs: { "status": "active", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "entry_count": 2, } }, ) post_status, post_payload = route_request( "/trading/external-context-surface-history", {}, method="POST", body={ "instrument_id": 1, "profile_key": "local_operator", "surface_id": "binance_public_chart_reference", "title": "binance_public_chart_reference", "source_group": "visual_reference_card", "selection_kind": "target_url", "target_url": "https://www.binance.com/en/trade/BTC_USDT?type=spot", }, dependencies={ "trading_external_context_surface_history_save": lambda **kwargs: { "status": "saved", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "latest_selection": {"surface_id": kwargs["surface_id"]}, } }, ) ack_status, ack_payload = route_request( "/trading/external-context-surface-history/acknowledge", {}, method="POST", body={ "instrument_id": 1, "profile_key": "local_operator", "surface_id": "binance_public_chart_reference", "selected_ts": "2026-05-12T02:05:30+00:00", "note": "checked and cleared", }, dependencies={ "trading_external_context_surface_history_ack": lambda **kwargs: { "status": "saved", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "latest_acknowledgment": {"surface_id": kwargs["surface_id"], "revisit_ack_note": kwargs["note"]}, } }, ) self.assertEqual(get_status, 200) self.assertEqual(get_payload["entry_count"], 2) self.assertEqual(post_status, 200) self.assertEqual(post_payload["latest_selection"]["surface_id"], "binance_public_chart_reference") self.assertEqual(ack_status, 200) self.assertEqual(ack_payload["latest_acknowledgment"]["surface_id"], "binance_public_chart_reference") self.assertEqual(ack_payload["latest_acknowledgment"]["revisit_ack_note"], "checked and cleared") def test_routes_decision_assistant_archive_and_focus(self) -> None: profiles_status, profiles_payload = route_request( "/trading/alerts/digest-preferences/profiles", {"instrument_id": ["1"]}, dependencies={ "trading_alert_operator_digest_profiles": lambda **kwargs: { "status": "active", "instrument_id": kwargs["instrument_id"], "items": [{"profile_key": "local_operator"}], } }, ) digest_archive_status, digest_archive_payload = route_request( "/trading/alerts/digest-history-archive", {"instrument_id": ["1"], "profile_scope": ["all_profiles"], "query_preset": ["recent_7d"], "limit": ["5"], "offset": ["0"], "preferred_history_kind": ["reminder"], "query_text": ["candidate"]}, dependencies={ "trading_alert_operator_digest_history_archive": lambda **kwargs: { "status": "active", "instrument_id": kwargs["instrument_id"], "profile_scope": kwargs["profile_scope"], "query_preset": kwargs["query_preset"], "limit": kwargs["limit"], "filters": {"preferred_history_kind": kwargs["preferred_history_kind"], "query_text": kwargs["query_text"]}, "long_range_trend_summary": {"status": "active"}, "baseline_pin_summary": {"status": "active"}, "ranked_aggregation_summary": {"status": "active"}, } }, ) digest_export_status, digest_export_payload = route_request( "/trading/alerts/digest-history-archive/export", {"instrument_id": ["1"], "profile_scope": ["all_profiles"], "query_preset": ["recent_7d"], "query_text": ["candidate"], "left_id": ["4"], "right_id": ["3"]}, dependencies={ "trading_alert_operator_digest_history_export": lambda **kwargs: { "status": "active", "export_kind": "digest_archive_bundle", "export_schema_version": "decision_assistant_archive_export_v4", "profile_scope": kwargs["profile_scope"], "query_preset": kwargs["query_preset"], "query_text": kwargs["query_text"], "left_id": kwargs["left_id"], "right_id": kwargs["right_id"], "long_range_trend_summary": {"status": "active"}, "baseline_pin_summary": {"status": "active"}, "ranked_aggregation_summary": {"status": "active"}, "selection_provenance_summary": {"status": "active", "export_scope": "selected_compare"}, "structured_export_summary": {"status": "active"}, "bundle_manifest": {"status": "active", "manifest_version": "decision_assistant_archive_bundle_manifest_v1"}, } }, ) digest_compare_status, digest_compare_payload = route_request( "/trading/alerts/digest-history-archive/compare", {"instrument_id": ["1"], "profile_scope": ["all_profiles"], "query_preset": ["recent_7d"], "left_id": ["4"], "right_id": ["3"], "query_text": ["candidate"]}, dependencies={ "trading_alert_operator_digest_history_compare": lambda **kwargs: { "status": "active", "profile_scope": kwargs["profile_scope"], "query_preset": kwargs["query_preset"], "left_id": kwargs["left_id"], "right_id": kwargs["right_id"], "query_text": kwargs["query_text"], } }, ) workflow_archive_status, workflow_archive_payload = route_request( "/trading/workflow-archive", {"instrument_id": ["1"], "profile_scope": ["all_profiles"], "query_preset": ["intraday_recent"], "current_session": ["LONDON"], "next_phase": ["intraday"], "query_text": ["handoff"]}, dependencies={ "trading_workflow_archive": lambda **kwargs: { "status": "active", "instrument_id": kwargs["instrument_id"], "profile_scope": kwargs["profile_scope"], "query_preset": kwargs["query_preset"], "filters": {"current_session": kwargs["current_session"], "next_phase": kwargs["next_phase"], "query_text": kwargs["query_text"]}, "long_range_trend_summary": {"status": "active"}, "baseline_pin_summary": {"status": "active"}, "ranked_aggregation_summary": {"status": "active"}, } }, ) workflow_export_status, workflow_export_payload = route_request( "/trading/workflow-archive/export", {"instrument_id": ["1"], "profile_scope": ["all_profiles"], "query_preset": ["intraday_recent"], "query_text": ["handoff"], "left_id": ["9"], "right_id": ["8"], "group_date": ["2026-05-16"], "session_group": ["LONDON"]}, dependencies={ "trading_workflow_archive_export": lambda **kwargs: { "status": "active", "export_kind": "workflow_archive_bundle", "export_schema_version": "decision_assistant_archive_export_v4", "profile_scope": kwargs["profile_scope"], "query_preset": kwargs["query_preset"], "query_text": kwargs["query_text"], "left_id": kwargs["left_id"], "right_id": kwargs["right_id"], "group_date": kwargs["group_date"], "session_group": kwargs["session_group"], "long_range_trend_summary": {"status": "active"}, "baseline_pin_summary": {"status": "active"}, "ranked_aggregation_summary": {"status": "active"}, "handoff_bundle_summary": {"status": "active"}, "handoff_packet_starter": {"status": "active"}, "replay_note_bridge_summary": {"status": "active"}, "artifact_bundle_summary": {"status": "active"}, "group_export_summary": {"status": "active"}, "selection_provenance_summary": {"status": "active", "export_scope": "selected_compare"}, "structured_export_summary": {"status": "active"}, "bundle_manifest": {"status": "active", "manifest_version": "decision_assistant_archive_bundle_manifest_v1"}, } }, ) workflow_compare_status, workflow_compare_payload = route_request( "/trading/workflow-archive/compare", {"instrument_id": ["1"], "profile_scope": ["all_profiles"], "query_preset": ["intraday_recent"], "left_id": ["9"], "right_id": ["8"], "query_text": ["handoff"]}, dependencies={ "trading_workflow_archive_compare": lambda **kwargs: { "status": "active", "profile_scope": kwargs["profile_scope"], "query_preset": kwargs["query_preset"], "left_id": kwargs["left_id"], "right_id": kwargs["right_id"], "query_text": kwargs["query_text"], } }, ) workflow_writeback_status, workflow_writeback_payload = route_request( "/trading/workflow-archive/compare-writeback", {}, method="POST", body={"instrument_id": 1, "profile_key": "desk_a", "left_id": 9, "right_id": 8, "query_text": "handoff"}, dependencies={ "trading_workflow_archive_compare_writeback": lambda **kwargs: { "status": "saved", "profile_key": kwargs["profile_key"], "left_id": kwargs["left_id"], "right_id": kwargs["right_id"], "compare_writeback_summary": {"status": "persisted", "note_title": "Workflow replay compare #9 vs #8"}, } }, ) focus_get_status, focus_get_payload = route_request( "/trading/assistant-focus", {"instrument_id": ["1"], "profile_key": ["desk_a"]}, dependencies={ "trading_assistant_focus": lambda **kwargs: { "status": "active", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "focus_candidate_id": 18, } }, ) focus_post_status, focus_post_payload = route_request( "/trading/assistant-focus", {}, method="POST", body={"instrument_id": 1, "profile_key": "desk_a", "candidate_id": 18}, dependencies={ "trading_assistant_focus_save": lambda **kwargs: { "status": "saved", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "focus_candidate_id": kwargs["candidate_id"], } }, ) self.assertEqual(profiles_status, 200) self.assertEqual(profiles_payload["items"][0]["profile_key"], "local_operator") self.assertEqual(digest_archive_status, 200) self.assertEqual(digest_archive_payload["profile_scope"], "all_profiles") self.assertEqual(digest_archive_payload["query_preset"], "recent_7d") self.assertEqual(digest_archive_payload["limit"], 5) self.assertEqual(digest_archive_payload["filters"]["preferred_history_kind"], "reminder") self.assertEqual(digest_archive_payload["filters"]["query_text"], "candidate") self.assertEqual(digest_archive_payload["long_range_trend_summary"]["status"], "active") self.assertEqual(digest_archive_payload["ranked_aggregation_summary"]["status"], "active") self.assertEqual(digest_export_status, 200) self.assertEqual(digest_export_payload["export_kind"], "digest_archive_bundle") self.assertEqual(digest_export_payload["export_schema_version"], "decision_assistant_archive_export_v4") self.assertEqual(digest_export_payload["profile_scope"], "all_profiles") self.assertEqual(digest_export_payload["query_preset"], "recent_7d") self.assertEqual(digest_export_payload["left_id"], 4) self.assertEqual(digest_export_payload["baseline_pin_summary"]["status"], "active") self.assertEqual(digest_export_payload["ranked_aggregation_summary"]["status"], "active") self.assertEqual(digest_export_payload["selection_provenance_summary"]["status"], "active") self.assertEqual(digest_export_payload["structured_export_summary"]["status"], "active") self.assertEqual(digest_export_payload["bundle_manifest"]["manifest_version"], "decision_assistant_archive_bundle_manifest_v1") self.assertEqual(digest_compare_status, 200) self.assertEqual(digest_compare_payload["left_id"], 4) self.assertEqual(digest_compare_payload["profile_scope"], "all_profiles") self.assertEqual(digest_compare_payload["query_preset"], "recent_7d") self.assertEqual(digest_compare_payload["query_text"], "candidate") self.assertEqual(workflow_archive_status, 200) self.assertEqual(workflow_archive_payload["profile_scope"], "all_profiles") self.assertEqual(workflow_archive_payload["query_preset"], "intraday_recent") self.assertEqual(workflow_archive_payload["filters"]["next_phase"], "intraday") self.assertEqual(workflow_archive_payload["filters"]["query_text"], "handoff") self.assertEqual(workflow_archive_payload["baseline_pin_summary"]["status"], "active") self.assertEqual(workflow_archive_payload["ranked_aggregation_summary"]["status"], "active") self.assertEqual(workflow_export_status, 200) self.assertEqual(workflow_export_payload["export_kind"], "workflow_archive_bundle") self.assertEqual(workflow_export_payload["export_schema_version"], "decision_assistant_archive_export_v4") self.assertEqual(workflow_export_payload["profile_scope"], "all_profiles") self.assertEqual(workflow_export_payload["query_preset"], "intraday_recent") self.assertEqual(workflow_export_payload["group_date"], "2026-05-16") self.assertEqual(workflow_export_payload["session_group"], "LONDON") self.assertEqual(workflow_export_payload["long_range_trend_summary"]["status"], "active") self.assertEqual(workflow_export_payload["ranked_aggregation_summary"]["status"], "active") self.assertEqual(workflow_export_payload["handoff_bundle_summary"]["status"], "active") self.assertEqual(workflow_export_payload["handoff_packet_starter"]["status"], "active") self.assertEqual(workflow_export_payload["replay_note_bridge_summary"]["status"], "active") self.assertEqual(workflow_export_payload["artifact_bundle_summary"]["status"], "active") self.assertEqual(workflow_export_payload["group_export_summary"]["status"], "active") self.assertEqual(workflow_export_payload["selection_provenance_summary"]["status"], "active") self.assertEqual(workflow_export_payload["structured_export_summary"]["status"], "active") self.assertEqual(workflow_export_payload["bundle_manifest"]["manifest_version"], "decision_assistant_archive_bundle_manifest_v1") self.assertEqual(workflow_compare_status, 200) self.assertEqual(workflow_compare_payload["profile_scope"], "all_profiles") self.assertEqual(workflow_compare_payload["query_preset"], "intraday_recent") self.assertEqual(workflow_compare_payload["right_id"], 8) self.assertEqual(workflow_compare_payload["query_text"], "handoff") self.assertEqual(workflow_writeback_status, 200) self.assertEqual(workflow_writeback_payload["profile_key"], "desk_a") self.assertEqual(workflow_writeback_payload["compare_writeback_summary"]["status"], "persisted") self.assertEqual(focus_get_status, 200) self.assertEqual(focus_get_payload["focus_candidate_id"], 18) self.assertEqual(focus_post_status, 200) self.assertEqual(focus_post_payload["profile_key"], "desk_a") def test_routes_decision_assistant_profile_management(self) -> None: create_status, create_payload = route_request( "/trading/alerts/digest-preferences/profiles", {}, method="POST", body={ "instrument_id": 1, "profile_key": "desk_b", "profile_label": "Desk B", "digest_mode": "high_value_only", }, dependencies={ "trading_alert_operator_digest_profiles_create": lambda **kwargs: { "status": "created", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "profile_label": kwargs["profile_label"], } }, ) rename_status, rename_payload = route_request( "/trading/alerts/digest-preferences/profiles/desk_b/rename", {}, method="POST", body={ "instrument_id": 1, "profile_label": "Desk Bravo", }, dependencies={ "trading_alert_operator_digest_profiles_rename": lambda **kwargs: { "status": "renamed", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], "profile_label": kwargs["profile_label"], } }, ) delete_status, delete_payload = route_request( "/trading/alerts/digest-preferences/profiles/desk_b/delete", {}, method="POST", body={"instrument_id": 1}, dependencies={ "trading_alert_operator_digest_profiles_delete": lambda **kwargs: { "status": "deleted", "instrument_id": kwargs["instrument_id"], "profile_key": kwargs["profile_key"], } }, ) self.assertEqual(create_status, 200) self.assertEqual(create_payload["status"], "created") self.assertEqual(create_payload["profile_key"], "desk_b") self.assertEqual(rename_status, 200) self.assertEqual(rename_payload["profile_key"], "desk_b") self.assertEqual(rename_payload["profile_label"], "Desk Bravo") self.assertEqual(delete_status, 200) self.assertEqual(delete_payload["status"], "deleted") def test_routes_maintenance_summary(self) -> None: status, payload = route_request( "/trading/maintenance-summary", {"instrument_id": ["1"], "alert_limit": ["25"], "visual_limit": ["15"], "incident_limit": ["5"]}, dependencies={ "maintenance_summary": lambda **kwargs: { "instrument_id": kwargs["instrument_id"], "alert_limit": kwargs["alert_limit"], "visual_limit": kwargs["visual_limit"], "incident_limit": kwargs["incident_limit"], "status": "healthy", } }, ) self.assertEqual(status, 200) self.assertEqual(payload["instrument_id"], 1) self.assertEqual(payload["alert_limit"], 25) self.assertEqual(payload["visual_limit"], 15) self.assertEqual(payload["incident_limit"], 5) def test_routes_trading_alert_silent_window_status(self) -> None: status, payload = route_request( "/trading/alerts/silent-window-status", {}, dependencies={"trading_alert_silent_window_status": lambda: {"active": True, "window_label": "00:00-06:00"}}, ) self.assertEqual(status, 200) self.assertTrue(payload["active"]) self.assertEqual(payload["window_label"], "00:00-06:00") def test_routes_trading_reconcile(self) -> None: status, payload = route_request( "/trading/reconcile", {}, method="POST", body={"instrument_id": 1}, dependencies={"trading_reconciliation_run": lambda **kwargs: {"status": "ok", "imported_execution_ids": [7]}}, ) self.assertEqual(status, 200) self.assertEqual(payload["imported_execution_ids"], [7]) def test_routes_trading_trust_score(self) -> None: status, payload = route_request( "/trading/trust-score", {"instrument_id": ["1"]}, dependencies={"trading_trust_score": lambda **kwargs: {"whitelist": ["OSOK"], "models": [{"model_code": "OSOK"}]}}, ) self.assertEqual(status, 200) self.assertEqual(payload["whitelist"], ["OSOK"]) def test_routes_trading_session_view_operator_notes_and_review_packet(self) -> None: session_status, session_payload = route_request( "/trading/session-view", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={"trading_session_view": lambda **kwargs: {"items": [{"session_code": "LONDON"}]}}, ) notes_status, notes_payload = route_request( "/trading/operator-notes", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={"trading_operator_notes": lambda **kwargs: [{"id": 1, "author": "alice"}]}, ) create_status, create_payload = route_request( "/trading/operator-notes", {}, method="POST", body={"instrument_id": 1, "author": "alice", "body": "note"}, dependencies={"trading_operator_note_create": lambda **kwargs: {"id": 2, **kwargs}}, ) packet_status, packet_payload = route_request( "/trading/review-packet", {"instrument_id": ["1"]}, dependencies={"trading_review_packet": lambda **kwargs: {"instrument_id": 1, "generated_ts": "now"}}, ) self.assertEqual(session_status, 200) self.assertEqual(session_payload["items"][0]["session_code"], "LONDON") self.assertEqual(notes_status, 200) self.assertEqual(notes_payload["items"][0]["author"], "alice") self.assertEqual(create_status, 201) self.assertEqual(create_payload["author"], "alice") self.assertEqual(packet_status, 200) self.assertEqual(packet_payload["instrument_id"], 1) def test_routes_trading_research_council_run_and_queries(self) -> None: status_run, payload_run = route_request( "/trading/research-council/run", {}, method="POST", body={"instrument_id": 1, "timeframe": "5m", "model_code": "SWEEP_MSS_FVG"}, dependencies={"trading_research_council_run": lambda **kwargs: {"id": 41, **kwargs}}, ) status_latest, payload_latest = route_request( "/trading/research-council/latest", {"instrument_id": ["1"]}, dependencies={"trading_research_council_latest": lambda **kwargs: {"id": 41, "instrument_id": kwargs["instrument_id"]}}, ) status_runs, payload_runs = route_request( "/trading/research-council/runs", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={"trading_research_council_runs": lambda **kwargs: [{"id": 41}, {"id": 40}]}, ) status_cycle, payload_cycle = route_request( "/trading/research-council/optimization-cycle", {}, method="POST", body={"instrument_id": 1, "timeframe": "5m", "eval_limit": 5, "lesson_limit": 10}, dependencies={ "trading_research_council_optimization_run": lambda **kwargs: { "id": 51, "status": "completed", **kwargs, } }, ) status_latest_cycle, payload_latest_cycle = route_request( "/trading/research-council/optimization-cycle/latest", {"instrument_id": ["1"]}, dependencies={"trading_research_council_optimization_latest": lambda **kwargs: {"id": 51, "instrument_id": kwargs["instrument_id"]}}, ) status_cycles, payload_cycles = route_request( "/trading/research-council/optimization-cycles", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={"trading_research_council_optimization_cycles": lambda **kwargs: [{"id": 51}]}, ) status_review, payload_review = route_request( "/trading/research-council/optimization-review", {"instrument_id": ["1"], "limit": ["2"], "stale_after_hours": ["12"]}, dependencies={ "trading_research_council_optimization_review": lambda **kwargs: { "instrument_id": kwargs["instrument_id"], "limit": kwargs["limit"], "stale_after_hours": kwargs["stale_after_hours"], "health_state": "healthy", } }, ) status_tick, payload_tick = route_request( "/trading/research-council/governance-tick", {}, method="POST", body={"instrument_id": 1, "limit": 2, "stale_after_hours": 12, "force_cycle": False}, dependencies={ "trading_research_council_governance_tick": lambda **kwargs: { "instrument_id": kwargs["instrument_id"], "limit": kwargs["limit"], "stale_after_hours": kwargs["stale_after_hours"], "force_cycle": kwargs["force_cycle"], "status": "no_op_healthy", } }, ) self.assertEqual(status_run, 200) self.assertEqual(payload_run["id"], 41) self.assertEqual(payload_run["timeframe"], "5m") self.assertEqual(status_latest, 200) self.assertEqual(payload_latest["instrument_id"], 1) self.assertEqual(status_runs, 200) self.assertEqual(len(payload_runs["items"]), 2) self.assertEqual(status_cycle, 200) self.assertEqual(payload_cycle["id"], 51) self.assertEqual(payload_cycle["eval_limit"], 5) self.assertEqual(payload_cycle["lesson_limit"], 10) self.assertEqual(status_latest_cycle, 200) self.assertEqual(payload_latest_cycle["instrument_id"], 1) self.assertEqual(status_cycles, 200) self.assertEqual(payload_cycles["items"], [{"id": 51}]) self.assertEqual(status_review, 200) self.assertEqual(payload_review["health_state"], "healthy") self.assertEqual(payload_review["limit"], 2) self.assertEqual(payload_review["stale_after_hours"], 12.0) self.assertEqual(status_tick, 200) self.assertEqual(payload_tick["status"], "no_op_healthy") self.assertFalse(payload_tick["force_cycle"]) def test_routes_trading_live_readiness(self) -> None: status, payload = route_request( "/trading/live-readiness", {"instrument_id": ["1"], "timeframe": ["5m"]}, dependencies={"trading_live_readiness": lambda **kwargs: {"ready_for_real_live": False, "timeframe": kwargs["timeframe"]}}, ) self.assertEqual(status, 200) self.assertFalse(payload["ready_for_real_live"]) self.assertEqual(payload["timeframe"], "5m") def test_routes_trading_okx_sandbox_preflight(self) -> None: status, payload = route_request( "/trading/okx-sandbox-preflight", {"instrument_id": ["1"], "timeframe": ["5m"]}, dependencies={ "trading_okx_sandbox_preflight": lambda **kwargs: { "status": "sandbox_preflight_ready", "timeframe": kwargs["timeframe"], "next_gate": "sandbox_submit_cancel_reconcile_smoke", } }, ) self.assertEqual(status, 200) self.assertEqual(payload["status"], "sandbox_preflight_ready") self.assertEqual(payload["timeframe"], "5m") self.assertEqual(payload["next_gate"], "sandbox_submit_cancel_reconcile_smoke") def test_routes_trading_launch_readiness(self) -> None: status, payload = route_request( "/trading/launch-readiness", {"instrument_id": ["1"], "timeframe": ["5m"]}, dependencies={"trading_launch_readiness": lambda **kwargs: {"analysis_only_ready": True, "decision_support_ready": False, "real_trade_loop_ready": False, "assisted_trading_ready": True, "timeframe": kwargs["timeframe"]}}, ) self.assertEqual(status, 200) self.assertTrue(payload["assisted_trading_ready"]) self.assertTrue(payload["analysis_only_ready"]) self.assertFalse(payload["decision_support_ready"]) self.assertEqual(payload["timeframe"], "5m") def test_routes_trading_shadow_paper_validation(self) -> None: status, payload = route_request( "/trading/shadow-paper-validation", {"instrument_id": ["1"], "timeframe": ["5m"]}, dependencies={"trading_shadow_paper_validation": lambda **kwargs: {"status": "collecting_paper_samples", "timeframe": kwargs["timeframe"]}}, ) self.assertEqual(status, 200) self.assertEqual(payload["status"], "collecting_paper_samples") self.assertEqual(payload["timeframe"], "5m") def test_routes_trading_shadow_paper_observation_record(self) -> None: status, payload = route_request( "/trading/shadow-paper/observe", {}, method="POST", body={ "instrument_id": 1, "timeframe": "5m", "session_code": "NY_AM", "candidate_id": 7, "paper_ticket_id": 11, "paper_execution_id": 21, "operator_agreement": "agree", "operator_note": "useful paper sample", "created_by": "alice", }, dependencies={ "trading_shadow_paper_observation_record": lambda **kwargs: { "status": "recorded", "instrument_id": kwargs["instrument_id"], "timeframe": kwargs["timeframe"], "candidate_id": kwargs["candidate_id"], "paper_ticket_id": kwargs["paper_ticket_id"], "paper_execution_id": kwargs["paper_execution_id"], "operator_agreement": kwargs["operator_agreement"], "created_by": kwargs["created_by"], } }, ) self.assertEqual(status, 201) self.assertEqual(payload["status"], "recorded") self.assertEqual(payload["timeframe"], "5m") self.assertEqual(payload["candidate_id"], 7) self.assertEqual(payload["paper_ticket_id"], 11) self.assertEqual(payload["paper_execution_id"], 21) self.assertEqual(payload["operator_agreement"], "agree") self.assertEqual(payload["created_by"], "alice") def test_routes_trading_broker_state(self) -> None: status, payload = route_request( "/trading/broker-state", {"instrument_id": ["1"]}, dependencies={"broker_state": lambda **kwargs: {"status": "broker_state_ok", **kwargs}}, ) self.assertEqual(status, 200) self.assertEqual(payload["status"], "broker_state_ok") self.assertEqual(payload["instrument_id"], 1) def test_routes_trading_watch_run(self) -> None: status, payload = route_request( "/trading/watch/run", {}, method="POST", body={"instrument_id": 1, "setup_tf": "15m", "trigger_tf": "5m"}, dependencies={"trading_watch_run": lambda **kwargs: {"id": 4, "status": "executable", **kwargs}}, ) self.assertEqual(status, 200) self.assertEqual(payload["setup_tf"], "15m") def test_routes_trading_candidate_detail_and_confirm_check(self) -> None: status_detail, payload_detail = route_request( "/trading/candidates/4", {}, dependencies={"trading_candidate_by_id": lambda candidate_id: {"id": candidate_id, "status": "executable"}}, ) status_check, payload_check = route_request( "/trading/candidates/4/confirm-check", {}, method="POST", body={}, dependencies={"trading_candidate_confirm_check": lambda candidate_id: {"candidate_id": candidate_id, "executable": True}}, ) self.assertEqual(status_detail, 200) self.assertEqual(payload_detail["id"], 4) self.assertEqual(status_check, 200) self.assertTrue(payload_check["executable"]) def test_routes_trading_ticket_flow(self) -> None: status_build, payload_build = route_request( "/trading/candidates/4/build-ticket", {}, method="POST", body={"execution_mode": "semi_auto", "allow_shadow_paper": False}, dependencies={"trading_candidate_build_ticket": lambda **kwargs: {"ticket_id": 11, **kwargs}}, ) status_get, payload_get = route_request( "/trading/tickets/11", {}, dependencies={"trading_ticket_by_id": lambda ticket_id: {"id": ticket_id, "status": "awaiting_confirmation"}}, ) status_risk, payload_risk = route_request( "/trading/tickets/11/risk-check", {}, method="POST", body={}, dependencies={"trading_ticket_risk_check": lambda *args, **kwargs: {"ticket_id": 11, "allowed": False}}, ) status_confirm, payload_confirm = route_request( "/trading/tickets/11/confirm", {}, method="POST", body={"confirmed_by": "alice"}, dependencies={"trading_ticket_confirm": lambda *args, **kwargs: {"id": 11, "status": "confirmed"}}, ) self.assertEqual(status_build, 200) self.assertEqual(payload_build["ticket_id"], 11) self.assertFalse(payload_build["allow_shadow_paper"]) self.assertEqual(status_get, 200) self.assertEqual(payload_get["status"], "awaiting_confirmation") self.assertEqual(status_risk, 200) self.assertFalse(payload_risk["allowed"]) self.assertEqual(status_confirm, 200) self.assertEqual(payload_confirm["status"], "confirmed") def test_routes_trading_execute_cancel_sync_and_weekly_review(self) -> None: status_paper, payload_paper = route_request( "/trading/tickets/11/paper-execute", {}, method="POST", body={"confirmed_by": "alice"}, dependencies={"trading_ticket_execute": lambda *args, **kwargs: {"status": "open", "mode": kwargs["execution_mode_expected"]}}, ) status_semi, payload_semi = route_request( "/trading/tickets/11/semi-auto-submit", {}, method="POST", body={"confirmed_by": "alice"}, dependencies={"trading_ticket_execute": lambda *args, **kwargs: {"status": "pending", "mode": kwargs["execution_mode_expected"]}}, ) status_cancel, payload_cancel = route_request( "/trading/executions/7/cancel", {}, method="POST", body={}, dependencies={"trading_execution_cancel": lambda execution_id: {"id": execution_id, "status": "cancelled"}}, ) status_close, payload_close = route_request( "/trading/executions/7/close", {}, method="POST", body={}, dependencies={"trading_execution_close": lambda execution_id: {"id": execution_id, "status": "closed"}}, ) status_sync, payload_sync = route_request( "/trading/executions/7/sync", {}, method="POST", body={}, dependencies={"trading_execution_sync": lambda execution_id: {"id": execution_id, "status": "closed"}}, ) status_weekly, payload_weekly = route_request( "/trading/research/weekly-review", {"instrument_id": ["1"]}, dependencies={"trading_weekly_review": lambda **kwargs: {"entry_count": 2, "next_week_actions": ["review_risk_process"]}}, ) self.assertEqual(status_paper, 200) self.assertEqual(payload_paper["mode"], "paper") self.assertEqual(status_semi, 200) self.assertEqual(payload_semi["mode"], "semi_auto") self.assertEqual(status_cancel, 200) self.assertEqual(payload_cancel["status"], "cancelled") self.assertEqual(status_close, 200) self.assertEqual(payload_close["status"], "closed") self.assertEqual(status_sync, 200) self.assertEqual(payload_sync["status"], "closed") self.assertEqual(status_weekly, 200) self.assertEqual(payload_weekly["entry_count"], 2) def test_routes_trading_executions_query(self) -> None: status, payload = route_request( "/trading/executions", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={"trading_executions": lambda **kwargs: [{"execution_id": 1, "execution_status": "open"}]}, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"execution_id": 1, "execution_status": "open"}]) def test_routes_trading_journal_generate(self) -> None: status, payload = route_request( "/trading/journal/generate", {}, method="POST", body={"execution_id": 7}, dependencies={"trading_journal_generate": lambda **kwargs: {"created": 1, **kwargs}}, ) self.assertEqual(status, 200) self.assertEqual(payload["execution_id"], 7) def test_routes_health(self) -> None: status, payload = route_request( "/health", {}, dependencies={ "health": lambda: {"ok": True}, "candles": lambda **kwargs: [], "sessions": lambda **kwargs: [], "signals": lambda **kwargs: [], "backtests": lambda **kwargs: [], }, ) self.assertEqual(status, 200) self.assertEqual(payload, {"ok": True}) def test_routes_trading_plans_query(self) -> None: status, payload = route_request( "/trading/plans", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={ "trading_plans": lambda **kwargs: [{"id": 1}, {"id": 2}], }, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 1}, {"id": 2}]) def test_routes_trading_candidates_query(self) -> None: status, payload = route_request( "/trading/candidates", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={ "trade_plan_candidates": lambda **kwargs: [{"id": 9}], }, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 9}]) def test_routes_trading_executions_query(self) -> None: status, payload = route_request( "/trading/executions", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={ "trading_executions": lambda **kwargs: [{"execution_id": 1, "audit_status": "ok"}], }, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"execution_id": 1, "audit_status": "ok"}]) def test_routes_trading_tickets_query(self) -> None: status, payload = route_request( "/trading/tickets", {"signal_id": ["1"], "limit": ["2"]}, dependencies={ "execution_tickets": lambda **kwargs: [{"id": 1, "status": "awaiting_confirmation"}], }, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 1, "status": "awaiting_confirmation"}]) def test_routes_live_operation_audits_query(self) -> None: status, payload = route_request( "/live-operation-audits", {"execution_ticket_id": ["1"], "limit": ["2"]}, dependencies={ "live_operation_audits": lambda **kwargs: [{"id": 1, "status": "rejected"}], }, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 1, "status": "rejected"}]) def test_routes_trading_journal_query(self) -> None: status, payload = route_request( "/trading/journal", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={ "trading_journal": lambda **kwargs: [{"id": 2, "framework_execution": True}], }, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 2, "framework_execution": True}]) def test_routes_trading_journal_summary_query(self) -> None: status, payload = route_request( "/trading/journal/summary", {"instrument_id": ["1"]}, dependencies={ "trading_journal_summary": lambda **kwargs: {"entry_count": 4, "next_week_actions": ["review_model_context"]}, }, ) self.assertEqual(status, 200) self.assertEqual(payload["entry_count"], 4) def test_routes_health(self) -> None: status, payload = route_request( "/health", {}, dependencies={ "health": lambda: {"ok": True}, "candles": lambda **kwargs: [], "sessions": lambda **kwargs: [], "signals": lambda **kwargs: [], "backtests": lambda **kwargs: [], "executions": lambda **kwargs: [], }, ) self.assertEqual(status, 200) self.assertEqual(payload, {"ok": True}) def test_routes_trading_plans_query_duplicate_guard(self) -> None: status, payload = route_request( "/trading/plans", {"instrument_id": ["1"], "limit": ["2"]}, dependencies={ "trading_plans": lambda **kwargs: [{"id": 1}, {"id": 2}], }, ) self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 1}, {"id": 2}]) def test_rejects_missing_required_query_param(self) -> None: with self.assertRaises(ValueError): route_request( "/trading/plans", {}, dependencies={ "trading_plans": lambda **kwargs: [], }, ) def test_returns_not_found_for_unknown_route(self) -> None: status, payload = route_request( "/unknown", {}, dependencies={ "health": lambda: {"ok": True}, "candles": lambda **kwargs: [], "sessions": lambda **kwargs: [], "signals": lambda **kwargs: [], "backtests": lambda **kwargs: [], "executions": lambda **kwargs: [], }, ) self.assertEqual(status, 404) self.assertEqual(payload["error"], "not_found") class ScriptFallbackTests(unittest.TestCase): def test_rejects_missing_required_query_param(self) -> None: with self.assertRaises(ValueError): route_request( "/trading/plans", {}, dependencies={ "trading_plans": lambda **kwargs: [], }, ) def test_returns_not_found_for_unknown_route(self) -> None: status, payload = route_request( "/unknown", {}, dependencies={ "health": lambda: {"ok": True}, "candles": lambda **kwargs: [], "sessions": lambda **kwargs: [], "signals": lambda **kwargs: [], "backtests": lambda **kwargs: [], }, ) self.assertEqual(status, 404) self.assertEqual(payload["error"], "not_found") class ScriptFallbackTests(unittest.TestCase): def test_run_execution_reports_database_unavailable(self) -> None: output = io.StringIO() with patch("scripts.run_execution.ExecutionService.prepare_execution", side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): exit_code = run_execution.main() self.assertEqual(exit_code, 2) self.assertIn("Database unavailable for execution sample: OperationalError", output.getvalue()) def test_run_execution_reports_database_unavailable_json(self) -> None: output = io.StringIO() with patch("scripts.run_execution.ExecutionService.prepare_execution", side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): exit_code = run_execution.main(env={"EXECUTION_OUTPUT_JSON": "1"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 2) self.assertFalse(payload["passed"]) self.assertEqual(payload["error_type"], "OperationalError") self.assertEqual(payload["execution_mode"], "paper") def test_run_execution_builds_account_risk_state_from_env(self) -> None: state = run_execution.build_account_risk_state_from_env( { "EXECUTION_KILL_SWITCH": "true", "EXECUTION_DAILY_PNL_PCT": "-0.01", "EXECUTION_CONSECUTIVE_LOSSES": "2", "EXECUTION_OPEN_POSITIONS": "1", "EXECUTION_SAME_SIDE_OPEN_POSITIONS": "1", "EXECUTION_MAX_DAILY_LOSS_PCT": "0.015", "EXECUTION_MAX_CONSECUTIVE_LOSSES": "4", "EXECUTION_MAX_OPEN_POSITIONS": "2", "EXECUTION_MAX_SAME_SIDE_POSITIONS": "2", } ) self.assertTrue(state.kill_switch) self.assertEqual(state.daily_pnl_pct, -0.01) self.assertEqual(state.consecutive_losses, 2) self.assertEqual(state.max_daily_loss_pct, 0.015) self.assertEqual(state.max_consecutive_losses, 4) def test_run_execution_empty_env_ignores_ambient_risk_state(self) -> None: with patch.dict( os.environ, { "EXECUTION_KILL_SWITCH": "true", "EXECUTION_CONSECUTIVE_LOSSES": "9", "EXECUTION_OPEN_POSITIONS": "7", }, clear=False, ): state = run_execution.build_account_risk_state_from_env({}) self.assertFalse(state.kill_switch) self.assertEqual(state.consecutive_losses, 0) self.assertEqual(state.open_positions, 0) def test_run_execution_settlement_reports_database_unavailable(self) -> None: output = io.StringIO() with patch("scripts.run_execution_settlement.ExecutionService.settle_latest_execution_for_signal", side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): exit_code = run_execution_settlement.main() self.assertEqual(exit_code, 2) self.assertIn("Database unavailable for execution settlement sample: OperationalError", output.getvalue()) def test_run_execution_settlement_reports_database_unavailable_json(self) -> None: output = io.StringIO() with patch("scripts.run_execution_settlement.ExecutionService.settle_latest_execution_for_signal", side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): exit_code = run_execution_settlement.main(env={"EXECUTION_SETTLEMENT_OUTPUT_JSON": "1"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 2) self.assertFalse(payload["passed"]) self.assertEqual(payload["error_type"], "OperationalError") self.assertEqual(payload["execution_mode"], "paper") def test_run_pipeline_reports_database_unavailable(self) -> None: output = io.StringIO() with patch("scripts.run_pipeline.run_closed_loop", side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): exit_code = run_pipeline.main() self.assertEqual(exit_code, 2) self.assertIn("Database unavailable for pipeline sample: OperationalError", output.getvalue()) def test_run_pipeline_reports_database_unavailable_json(self) -> None: output = io.StringIO() with patch("scripts.run_pipeline.run_closed_loop", side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): exit_code = run_pipeline.main(env={"PIPELINE_OUTPUT_JSON": "1"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 2) self.assertFalse(payload["passed"]) self.assertEqual(payload["error_type"], "OperationalError") def test_run_pipeline_passes_backtest_cost_config(self) -> None: captured = {} def fake_run_closed_loop(**kwargs): captured.update(kwargs) return {"ok": True} output = io.StringIO() env = { "BACKTEST_FEE_RATE": "0.0005", "BACKTEST_SLIPPAGE_RATE": "0.0007", "BACKTEST_FIXED_COST_R": "0.05", } with patch("scripts.run_pipeline.run_closed_loop", side_effect=fake_run_closed_loop), patch.dict(os.environ, env, clear=False), contextlib.redirect_stdout(output): run_pipeline.main() cost_config = captured["backtest_cost_config"] self.assertEqual(cost_config.fee_rate, 0.0005) self.assertEqual(cost_config.slippage_rate, 0.0007) self.assertEqual(cost_config.fixed_cost_r, 0.05) def test_run_pipeline_passes_backtest_quality_gate_config(self) -> None: captured = {} def fake_run_closed_loop(**kwargs): captured.update(kwargs) return {"ok": True, "backtest_quality_gate": {"passed": True}} output = io.StringIO() env = { "BACKTEST_MIN_TOTAL_SIGNALS": "50", "BACKTEST_MIN_EXECUTED_SIGNALS": "20", "BACKTEST_MIN_EXPECTANCY_R": "0.1", "BACKTEST_MIN_PROFIT_FACTOR_R": "1.5", "BACKTEST_MAX_DRAWDOWN_R": "6.0", "BACKTEST_MAX_CONSECUTIVE_LOSSES": "3", } with patch("scripts.run_pipeline.run_closed_loop", side_effect=fake_run_closed_loop), patch.dict(os.environ, env, clear=False), contextlib.redirect_stdout(output): run_pipeline.main() quality_gate = captured["backtest_quality_gate"] self.assertEqual(quality_gate.min_total_signals, 50) self.assertEqual(quality_gate.min_executed_signals, 20) self.assertEqual(quality_gate.min_expectancy_r, 0.1) self.assertEqual(quality_gate.min_profit_factor_r, 1.5) self.assertEqual(quality_gate.max_drawdown_r, 6.0) self.assertEqual(quality_gate.max_consecutive_losses, 3) def test_run_pipeline_passes_backtest_replay_mode_and_config(self) -> None: captured = {} def fake_run_closed_loop(**kwargs): captured.update(kwargs) return {"ok": True, "backtest_quality_gate": {"passed": True}} output = io.StringIO() env = { "BACKTEST_MODE": "replay_no_lookahead", "BACKTEST_REPLAY_LEFT_BARS": "1", "BACKTEST_REPLAY_RIGHT_BARS": "3", } with patch("scripts.run_pipeline.run_closed_loop", side_effect=fake_run_closed_loop), patch.dict(os.environ, env, clear=False), contextlib.redirect_stdout(output): run_pipeline.main() self.assertEqual(captured["backtest_mode"], "replay_no_lookahead") self.assertEqual(captured["backtest_replay_config"].left_bars, 1) self.assertEqual(captured["backtest_replay_config"].right_bars, 3) def test_run_pipeline_passes_signal_evidence_mode(self) -> None: captured = {} def fake_run_closed_loop(**kwargs): captured.update(kwargs) return {"ok": True, "backtest_quality_gate": {"passed": True}} output = io.StringIO() with patch("scripts.run_pipeline.run_closed_loop", side_effect=fake_run_closed_loop), patch.dict(os.environ, {"SIGNAL_EVIDENCE_MODE": "strict"}, clear=False), contextlib.redirect_stdout(output): run_pipeline.main() self.assertIn("sweep_id", captured["signal_required_evidence_keys"]) self.assertIn("liquidity_target_id", captured["signal_required_evidence_keys"]) def test_run_pipeline_passes_execution_account_state(self) -> None: captured = {} def fake_run_closed_loop(**kwargs): captured.update(kwargs) return {"ok": True} output = io.StringIO() env = { "EXECUTION_DAILY_PNL_PCT": "-0.01", "EXECUTION_CONSECUTIVE_LOSSES": "2", } with patch("scripts.run_pipeline.run_closed_loop", side_effect=fake_run_closed_loop), patch.dict(os.environ, env, clear=False), contextlib.redirect_stdout(output): run_pipeline.main() account_state = captured["execution_account_state"] self.assertEqual(account_state.daily_pnl_pct, -0.01) self.assertEqual(account_state.consecutive_losses, 2) def test_run_pipeline_blocks_when_required_quality_gate_fails(self) -> None: output = io.StringIO() result = {"backtest_quality_gate": {"passed": False, "reasons": ["sample_size_too_small"]}} with patch("scripts.run_pipeline.run_closed_loop", return_value=result), patch.dict(os.environ, {"BACKTEST_REQUIRE_QUALITY_GATE": "1"}, clear=False), contextlib.redirect_stdout(output): exit_code = run_pipeline.main() self.assertEqual(exit_code, 2) self.assertIn('"backtest_quality_gate"', output.getvalue()) class OfflineChecksScriptTests(unittest.TestCase): def test_build_offline_check_commands_adds_replay_file_when_configured(self) -> None: commands = run_offline_checks.build_check_commands(env={"OFFLINE_REPLAY_FILE": "sample.jsonl"}) self.assertEqual(commands[-1][0], "replay_file") self.assertEqual(commands[-1][1][-1], "sample.jsonl") def test_run_offline_checks_returns_zero_when_all_checks_pass(self) -> None: calls = [] def fake_runner(command, **kwargs): calls.append(command) return SimpleNamespace(returncode=0, stdout="ok\n", stderr="") output = io.StringIO() commands = [("unit_tests", ["python", "-m", "unittest"]), ("dashboard_js_syntax", ["node", "--check", "src/web/app.js"])] with contextlib.redirect_stdout(output): exit_code = run_offline_checks.run_checks(commands=commands, runner=fake_runner, node_available=True, env={}) self.assertEqual(exit_code, 0) self.assertEqual(len(calls), 2) self.assertIn("Offline check passed: dashboard_js_syntax", output.getvalue()) def test_run_offline_checks_can_include_optional_replay_file(self) -> None: calls = [] def fake_runner(command, **kwargs): calls.append(command) return SimpleNamespace(returncode=0, stdout="", stderr="") output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_offline_checks.run_checks(runner=fake_runner, node_available=True, env={"OFFLINE_REPLAY_FILE": "sample.csv"}) self.assertEqual(exit_code, 0) self.assertEqual(calls[-1][-1], "sample.csv") self.assertIn("Offline check passed: replay_file", output.getvalue()) def test_run_offline_checks_stops_on_failure(self) -> None: calls = [] def fake_runner(command, **kwargs): calls.append(command) return SimpleNamespace(returncode=7, stdout="", stderr="failed\n") output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_offline_checks.run_checks(commands=[("unit_tests", ["python", "-m", "unittest"])], runner=fake_runner, env={}) self.assertEqual(exit_code, 7) self.assertEqual(len(calls), 1) self.assertIn("Offline check failed: unit_tests exit_code=7", output.getvalue()) def test_run_offline_checks_skips_node_when_unavailable(self) -> None: calls = [] def fake_runner(command, **kwargs): calls.append(command) return SimpleNamespace(returncode=0, stdout="", stderr="") output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_offline_checks.run_checks(commands=[("dashboard_js_syntax", ["node", "--check", "src/web/app.js"])], runner=fake_runner, node_available=False, env={}) self.assertEqual(exit_code, 0) self.assertEqual(calls, []) self.assertIn("Offline check skipped: dashboard_js_syntax", output.getvalue()) def test_run_offline_checks_can_emit_json_summary(self) -> None: def fake_runner(command, **kwargs): return SimpleNamespace(returncode=0, stdout="ok\n", stderr="") output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_offline_checks.run_checks( commands=[("unit_tests", ["python", "-m", "unittest"])], runner=fake_runner, env={"OFFLINE_CHECKS_OUTPUT_JSON": "1"}, ) summary = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(summary["passed"]) self.assertEqual(summary["checks"][0]["name"], "unit_tests") self.assertEqual(summary["checks"][0]["status"], "passed") def test_run_offline_checks_passes_merged_env_to_subprocesses(self) -> None: seen_env = {} def fake_runner(command, **kwargs): seen_env.update(kwargs["env"]) return SimpleNamespace(returncode=0, stdout="", stderr="") with contextlib.redirect_stdout(io.StringIO()): exit_code = run_offline_checks.run_checks( commands=[("unit_tests", ["python", "-m", "unittest"])], runner=fake_runner, env={"CUSTOM_OFFLINE_FLAG": "1"}, ) self.assertEqual(exit_code, 0) self.assertEqual(seen_env["CUSTOM_OFFLINE_FLAG"], "1") self.assertIn("PATH", seen_env) if __name__ == "__main__": unittest.main()