ai-exchange/tests/unit/test_runtime_script_json.py

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_session_generator,
    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_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)


if __name__ == "__main__":
    unittest.main()