Codex
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from src.services.broker.live_readiness_service import LiveReadinessService
|
|
|
|
|
|
class LiveReadinessServiceTests(unittest.TestCase):
|
|
def test_build_summary_reports_blocking_reasons(self) -> None:
|
|
service = LiveReadinessService(
|
|
broker_read_model_service=_BrokerReadService(),
|
|
market_state_service=_MarketStateService(),
|
|
execution_repository=_ExecutionRepo(),
|
|
decision_packet_service=_DecisionPacketService(),
|
|
trust_service=_TrustService(),
|
|
)
|
|
|
|
with patch("src.services.broker.live_readiness_service.get_trading_runtime_service", return_value=_RuntimeService()):
|
|
payload = service.build_summary(instrument_id=1, timeframe="1m")
|
|
|
|
self.assertFalse(payload["ready_for_real_live"])
|
|
self.assertIn("adapter_real_live_not_ready", payload["blocking_reasons"])
|
|
self.assertEqual(payload["operator_confirmation_protocol"]["protocol_version"], "live-confirmation-v1")
|
|
self.assertEqual(payload["live_order_schema"]["schema_version"], "live-order-v1")
|
|
|
|
|
|
class _BrokerState:
|
|
status = "broker_state_ok"
|
|
|
|
def to_dict(self):
|
|
return {"status": "broker_state_ok"}
|
|
|
|
|
|
class _BrokerReadService:
|
|
def fetch_snapshot(self, **kwargs):
|
|
return _BrokerState()
|
|
|
|
|
|
class _MarketState:
|
|
freshness = type("Freshness", (), {"code": "market_data_fresh"})()
|
|
stream_state = type("Stream", (), {"code": "market_stream_connected"})()
|
|
data_gap_state = type("Gap", (), {"code": "market_data_gap_clear"})()
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"freshness": {"code": "market_data_fresh"},
|
|
"stream_state": {"code": "market_stream_connected"},
|
|
"data_gap_state": {"code": "market_data_gap_clear"},
|
|
}
|
|
|
|
|
|
class _MarketStateService:
|
|
def build_market_state(self, **kwargs):
|
|
return _MarketState()
|
|
|
|
|
|
class _ExecutionRepo:
|
|
def fetch_recent_execution_ticket_rows(self, **kwargs):
|
|
return []
|
|
|
|
def fetch_recent_live_operation_audits(self, **kwargs):
|
|
return []
|
|
|
|
|
|
class _DecisionPacketService:
|
|
def build_preview(self, **kwargs):
|
|
return {"action_recommendation": "confirm_ticket"}
|
|
|
|
|
|
class _TrustService:
|
|
def build_summary(self, **kwargs):
|
|
return {"models": [{"model_code": "OSOK"}], "whitelist": ["OSOK"], "blacklist": []}
|
|
|
|
|
|
class _RuntimeService:
|
|
def status(self):
|
|
return {"scheduler_state": "running", "current_blockers": []}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|