Codex
493 lines
23 KiB
Python
493 lines
23 KiB
Python
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()
|