import contextlib import io import json import os import unittest 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 scripts import ( apply_migrations, check_db, check_db_connection, run_historical_compatibility_repair, run_api_server, run_dev, run_displacement, run_fvg, run_liquidity, run_broker_reconciliation, run_mss, run_ranges, run_decision_packet, run_launch_readiness, run_research_council_arena, run_research_council_backfill_lessons, run_research_council, run_research_council_eval, run_research_council_governance_tick, run_research_council_optimization_loop, run_research_council_optimization_review, run_research_council_replay, run_review_packet, run_session_generator, run_live_readiness, run_trading_runtime, run_sweeps, run_swings, run_trust_score, show_models, ) class RuntimeScriptJsonTests(unittest.TestCase): def test_structure_scripts_report_database_unavailable_json(self) -> None: cases = [ (run_session_generator, "SESSION_GENERATOR_OUTPUT_JSON", "scripts.run_session_generator.SessionGenerator.generate_for_date"), (run_swings, "SWINGS_OUTPUT_JSON", "scripts.run_swings.SwingService.detect_swings"), (run_ranges, "RANGES_OUTPUT_JSON", "scripts.run_ranges.DealingRangeService.build_active_range"), (run_liquidity, "LIQUIDITY_OUTPUT_JSON", "scripts.run_liquidity.LiquidityPoolService.build_candidates"), (run_sweeps, "SWEEPS_OUTPUT_JSON", "scripts.run_sweeps.SweepService.detect_sweeps"), (run_displacement, "DISPLACEMENT_OUTPUT_JSON", "scripts.run_displacement.DisplacementService.detect_displacement"), (run_mss, "MSS_OUTPUT_JSON", "scripts.run_mss.MSSService.detect_mss"), (run_fvg, "FVG_OUTPUT_JSON", "scripts.run_fvg.FVGService.detect_fvg"), ] for module, env_key, patch_target in cases: with self.subTest(script=module.__name__): output = io.StringIO() with patch(patch_target, side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): if module.__name__ == run_historical_compatibility_repair.__name__: exit_code = module.main(argv=[], env={env_key: "1"}) else: exit_code = module.main(env={env_key: "1"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 2) self.assertFalse(payload["passed"]) self.assertEqual(payload["error_type"], "OperationalError") self.assertEqual(payload["instrument_id"], 1) self.assertEqual(payload["timeframe"], "1m") def test_database_check_scripts_report_unavailable_json(self) -> None: cases = [ (check_db, "CHECK_DB_OUTPUT_JSON", "scripts.check_db.engine.connect"), (check_db_connection, "CHECK_DB_CONNECTION_OUTPUT_JSON", "scripts.check_db_connection.engine.connect"), (run_historical_compatibility_repair, "HISTORICAL_REPAIR_OUTPUT_JSON", "scripts.run_historical_compatibility_repair.engine.begin"), ] for module, env_key, patch_target in cases: with self.subTest(script=module.__name__): output = io.StringIO() with patch(patch_target, side_effect=OperationalError("stmt", {}, Exception())), contextlib.redirect_stdout(output): if module.__name__ == run_historical_compatibility_repair.__name__: exit_code = module.main(argv=[], env={env_key: "1"}) else: exit_code = module.main(env={env_key: "1"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 2) self.assertFalse(payload["passed"]) self.assertEqual(payload["error_type"], "OperationalError") if module in {check_db, check_db_connection}: self.assertFalse(payload["available"]) def test_show_models_outputs_json(self) -> None: output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = show_models.main(env={"SHOW_MODELS_OUTPUT_JSON": "1"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertGreater(payload["model_count"], 0) self.assertIn("candle", payload["models"]) def test_apply_migrations_delegates_to_safe_runner(self) -> None: with patch("scripts.apply_migrations.run_migrations_main", return_value=2) as run_migrations: exit_code = apply_migrations.main(env={"MIGRATIONS_OUTPUT_JSON": "1"}) self.assertEqual(exit_code, 2) run_migrations.assert_called_once_with(env={"MIGRATIONS_OUTPUT_JSON": "1"}) def test_run_dev_uses_env_and_returns_when_server_stops(self) -> None: with patch("scripts.run_dev.serve_api") as serve_api: exit_code = run_dev.main(env={"AI_ICT_API_HOST": "127.0.0.1", "AI_ICT_API_PORT": "8124"}) self.assertEqual(exit_code, 0) serve_api.assert_called_once_with(host="127.0.0.1", port=8124) def test_run_api_server_uses_env_and_returns_when_server_stops(self) -> None: with patch("scripts.run_api_server.serve_api") as serve_api: exit_code = run_api_server.main(env={"AI_ICT_API_HOST": "127.0.0.1", "AI_ICT_API_PORT": "8123"}) self.assertEqual(exit_code, 0) serve_api.assert_called_once_with(host="127.0.0.1", port=8123) def test_run_trading_runtime_outputs_json(self) -> None: service = type("Service", (), {"run_once": lambda self: {"scheduler_state": "running", "current_blockers": []}})() output = io.StringIO() with patch("scripts.run_trading_runtime.get_trading_runtime_service", return_value=service), contextlib.redirect_stdout(output): exit_code = run_trading_runtime.main(env={"TRADING_RUNTIME_OUTPUT_JSON": "1"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["scheduler_state"], "running") def test_run_decision_packet_outputs_json(self) -> None: service = type("Service", (), {"build_persist_and_alert": lambda self, instrument_id: {"id": 1, "status": "actionable", "instrument_id": instrument_id}})() output = io.StringIO() with patch("scripts.run_decision_packet.DecisionPacketService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_decision_packet.main(env={"DECISION_PACKET_OUTPUT_JSON": "1", "DECISION_PACKET_INSTRUMENT_ID": "2"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["instrument_id"], 2) def test_run_decision_packet_json_handles_nonprimitive_values(self) -> None: service = type( "Service", (), { "build_persist_and_alert": lambda self, instrument_id: { "id": 1, "status": "actionable", "instrument_id": instrument_id, "packet_ts": object(), } }, )() output = io.StringIO() with patch("scripts.run_decision_packet.DecisionPacketService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_decision_packet.main(env={"DECISION_PACKET_OUTPUT_JSON": "1", "DECISION_PACKET_INSTRUMENT_ID": "2"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["instrument_id"], 2) def test_run_broker_reconciliation_outputs_json(self) -> None: service = type("Service", (), {"reconcile_instrument": lambda self, instrument_id: {"status": "ok", "instrument_id": instrument_id, "matched_execution_ids": [1]}})() output = io.StringIO() with patch("scripts.run_broker_reconciliation.BrokerExecutionReconciliationService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_broker_reconciliation.main( env={"BROKER_RECONCILIATION_OUTPUT_JSON": "1", "BROKER_RECONCILIATION_INSTRUMENT_ID": "2"} ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["instrument_id"], 2) def test_run_trust_score_outputs_json(self) -> None: service = type("Service", (), {"build_summary": lambda self, instrument_id: {"instrument_id": instrument_id, "whitelist": ["OSOK"], "models": []}})() output = io.StringIO() with patch("scripts.run_trust_score.ModelTrustService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_trust_score.main(env={"TRUST_SCORE_OUTPUT_JSON": "1", "TRUST_SCORE_INSTRUMENT_ID": "2"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["instrument_id"], 2) def test_run_review_packet_outputs_json(self) -> None: service = type("Service", (), {"build_packet": lambda self, instrument_id: {"instrument_id": instrument_id, "generated_ts": "now"}})() output = io.StringIO() with patch("scripts.run_review_packet.ReviewPacketService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_review_packet.main(env={"REVIEW_PACKET_OUTPUT_JSON": "1", "REVIEW_PACKET_INSTRUMENT_ID": "2"}) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["instrument_id"], 2) def test_run_research_council_outputs_json(self) -> None: service = type( "Service", (), { "build_and_persist": lambda self, **kwargs: {"id": 5, **kwargs}, "get_resume_state": lambda self, instrument_id: {"instrument_id": instrument_id, "can_resume": False}, }, )() output = io.StringIO() with patch("scripts.run_research_council.ResearchCouncilService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council.main( env={ "RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2", "RESEARCH_COUNCIL_TIMEFRAME": "5m", } ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["instrument_id"], 2) self.assertEqual(payload["timeframe"], "5m") def test_run_research_council_json_handles_nonprimitive_values(self) -> None: service = type( "Service", (), { "build_and_persist": lambda self, **kwargs: {"id": 5, "generated_ts": object(), **kwargs}, "get_resume_state": lambda self, instrument_id: {"instrument_id": instrument_id, "can_resume": False}, }, )() output = io.StringIO() with patch("scripts.run_research_council.ResearchCouncilService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council.main( env={ "RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2", } ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) def test_run_research_council_outputs_checkpoint_status_json(self) -> None: service = type( "Service", (), { "get_resume_state": lambda self, instrument_id: { "instrument_id": instrument_id, "can_resume": True, "latest_checkpoint": {"stage": "run_persisted"}, } }, )() output = io.StringIO() with patch("scripts.run_research_council.ResearchCouncilService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council.main( env={ "RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_CHECKPOINT_STATUS": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2", } ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertTrue(payload["can_resume"]) self.assertEqual(payload["latest_checkpoint"]["stage"], "run_persisted") def test_run_research_council_backfill_lessons_outputs_json(self) -> None: service = type("Service", (), {"backfill_lessons": lambda self, **kwargs: {"instrument_id": kwargs["instrument_id"], "totals": {"requested": 1, "created": 1, "updated": 0}}})() output = io.StringIO() with patch("scripts.run_research_council_backfill_lessons.ResearchCouncilService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council_backfill_lessons.main( env={"RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2"} ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["totals"]["created"], 1) def test_run_research_council_replay_outputs_json(self) -> None: service = type( "Service", (), { "get_resume_state": lambda self, instrument_id: {"instrument_id": instrument_id, "can_resume": False}, "build_preview": lambda self, **kwargs: {"decision": {"overall_stance": "observe_only"}, "context": {"compact_lessons": []}}, }, )() output = io.StringIO() with patch("scripts.run_research_council_replay.ResearchCouncilService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council_replay.main( env={"RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2"} ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["preview"]["decision"]["overall_stance"], "observe_only") def test_run_research_council_eval_outputs_json(self) -> None: service = type( "Service", (), { "get_recent_runs": lambda self, **kwargs: [ { "payload": { "context": { "latest_candidate": {"session_code": "LONDON", "model_code": "SWEEP_MSS_FVG", "missing_confirmations": ["mss"]}, "trust_policy": {"blocking_reasons": ["trust_blocked"]}, "risk_result": {"reasons": ["risk_blocked"]}, "compact_lessons": [{"lesson_key": "journal:1"}], }, "decision": { "follow_up_checks": ["mss", "risk_blocked"], "operator_questions": ["question"], "disagreement_points": ["debate"], "committee_confidence": 0.6, }, } } ] }, )() output = io.StringIO() with patch("scripts.run_research_council_eval.ResearchCouncilService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council_eval.main( env={"RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2"} ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["run_count"], 1) def test_run_research_council_arena_outputs_json(self) -> None: service = type( "Service", (), { "build_preview": lambda self, **kwargs: { "decision": {"overall_stance": "observe_only", "committee_confidence": 0.6, "follow_up_checks": ["x"], "disagreement_points": []}, "context": {"compact_lessons": [{"lesson_key": "journal:1"}]}, } }, )() output = io.StringIO() with patch("scripts.run_research_council_arena.ResearchCouncilService", return_value=service), patch( "scripts.run_research_council_arena.ModelTrustService", return_value=type("TrustService", (), {"build_summary": lambda self, instrument_id: {"models": [{"model_code": "SWEEP_MSS_FVG"}]}})(), ), contextlib.redirect_stdout(output): exit_code = run_research_council_arena.main( env={"RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2"} ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["rows"][0]["lesson_count"], 1) def test_run_research_council_optimization_loop_outputs_json(self) -> None: service = type( "Service", (), { "run_cycle": lambda self, **kwargs: { "id": 9, "status": "completed", "passed": True, "instrument_id": kwargs["instrument_id"], "timeframe": kwargs["timeframe"], "next_actions": ["rerun_optimization_cycle_after_next_market_sample"], } }, )() output = io.StringIO() with patch("scripts.run_research_council_optimization_loop.CouncilOptimizationService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council_optimization_loop.main( env={ "RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2", "RESEARCH_COUNCIL_TIMEFRAME": "5m", "RESEARCH_COUNCIL_REPLAY_FILE": "tests/fixtures/offline_replay_sample.jsonl", } ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["instrument_id"], 2) self.assertEqual(payload["timeframe"], "5m") def test_run_research_council_optimization_review_outputs_json(self) -> None: service = type( "Service", (), { "build_review": lambda self, **kwargs: { "present": True, "passed": True, "instrument_id": kwargs["instrument_id"], "health_state": "healthy", "recommended_actions": ["continue_periodic_optimization_cycle"], } }, )() output = io.StringIO() with patch("scripts.run_research_council_optimization_review.CouncilOptimizationReviewService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council_optimization_review.main( env={ "RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2", "RESEARCH_COUNCIL_REVIEW_LIMIT": "3", } ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["health_state"], "healthy") def test_run_research_council_governance_tick_outputs_json(self) -> None: service = type( "Service", (), { "run_tick": lambda self, **kwargs: { "status": "no_op_healthy", "passed": True, "instrument_id": kwargs["instrument_id"], "effective_health_state": "healthy", "recommended_actions": ["continue_periodic_optimization_cycle"], } }, )() output = io.StringIO() with patch("scripts.run_research_council_governance_tick.CouncilGovernanceTickService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_research_council_governance_tick.main( env={ "RESEARCH_COUNCIL_OUTPUT_JSON": "1", "RESEARCH_COUNCIL_INSTRUMENT_ID": "2", "RESEARCH_COUNCIL_GOVERNANCE_FORCE_CYCLE": "0", } ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertTrue(payload["passed"]) self.assertEqual(payload["status"], "no_op_healthy") self.assertEqual(payload["instrument_id"], 2) def test_run_live_readiness_outputs_json(self) -> None: service = type("Service", (), {"build_live_readiness": lambda self, instrument_id, timeframe: {"instrument_id": instrument_id, "timeframe": timeframe, "ready_for_real_live": False, "blocking_reasons": ["adapter_real_live_not_ready"]}})() output = io.StringIO() with patch("scripts.run_live_readiness.LiveBrokerService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_live_readiness.main( env={"LIVE_READINESS_OUTPUT_JSON": "1", "LIVE_READINESS_INSTRUMENT_ID": "2", "LIVE_READINESS_TIMEFRAME": "5m"} ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 2) self.assertEqual(payload["instrument_id"], 2) self.assertEqual(payload["timeframe"], "5m") def test_run_launch_readiness_outputs_json(self) -> None: service = type("Service", (), {"build_summary": lambda self, instrument_id, timeframe: {"instrument_id": instrument_id, "timeframe": timeframe, "analysis_only_ready": True, "decision_support_ready": False, "real_trade_loop_ready": False, "assisted_trading_ready": True, "real_live_ready": False, "blocking_reasons": [], "warning_reasons": ["real_trade_sample_verified"]}})() output = io.StringIO() with patch("scripts.run_launch_readiness.LaunchReadinessService", return_value=service), contextlib.redirect_stdout(output): exit_code = run_launch_readiness.main( env={"LAUNCH_READINESS_OUTPUT_JSON": "1", "LAUNCH_READINESS_INSTRUMENT_ID": "2", "LAUNCH_READINESS_TIMEFRAME": "5m"} ) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["instrument_id"], 2) self.assertEqual(payload["timeframe"], "5m") if __name__ == "__main__": unittest.main()