Codex
59 lines
2.6 KiB
Python
59 lines
2.6 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from src.services.runtime import TradingRuntimeScheduler
|
|
from tests.unit.test_trading_runtime_service import (
|
|
FakeAlertDispatcherService,
|
|
FakeBrokerReadModelService,
|
|
FakeDecisionPacketService,
|
|
FakeMarketStateService,
|
|
FakeReconciliationService,
|
|
FakeResearchCouncilService,
|
|
FakeShadowPaperSampleService,
|
|
FakeSetupWatcherService,
|
|
FakeTradingPlanService,
|
|
)
|
|
|
|
|
|
class ResearchCouncilRuntimeIntegrationTests(unittest.TestCase):
|
|
def _build_scheduler(self, research_council_service):
|
|
return TradingRuntimeScheduler(
|
|
market_state_service=FakeMarketStateService(),
|
|
trading_plan_service=FakeTradingPlanService(),
|
|
setup_watcher_service=FakeSetupWatcherService(),
|
|
broker_read_model_service=FakeBrokerReadModelService(),
|
|
reconciliation_service=FakeReconciliationService(),
|
|
decision_packet_service=FakeDecisionPacketService(),
|
|
alert_dispatcher_service=FakeAlertDispatcherService(),
|
|
research_council_service=research_council_service,
|
|
shadow_paper_sample_service=FakeShadowPaperSampleService(),
|
|
)
|
|
|
|
def test_council_task_runs_after_watch_before_decision_packet(self) -> None:
|
|
council_service = FakeResearchCouncilService()
|
|
scheduler = self._build_scheduler(council_service)
|
|
|
|
with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):
|
|
payload = scheduler.run_cycle()
|
|
|
|
self.assertEqual(payload["tasks"]["watch"]["last_status"], "ok")
|
|
self.assertEqual(payload["tasks"]["research_council"]["last_status"], "ok")
|
|
self.assertEqual(payload["tasks"]["research_council"]["details"]["run_id"], 601)
|
|
self.assertEqual(payload["tasks"]["decision_packet"]["last_status"], "ok")
|
|
self.assertEqual(council_service.calls[0]["timeframe"], scheduler.trigger_tf)
|
|
|
|
def test_council_failure_degrades_without_blocking_decision_packet(self) -> None:
|
|
scheduler = self._build_scheduler(FakeResearchCouncilService(should_fail=True))
|
|
|
|
with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):
|
|
payload = scheduler.run_cycle()
|
|
|
|
self.assertEqual(payload["tasks"]["research_council"]["last_status"], "error")
|
|
self.assertEqual(payload["tasks"]["decision_packet"]["last_status"], "ok")
|
|
self.assertIn("research_council_task_failed:BTC-USDT", payload["current_warnings"])
|
|
self.assertNotIn("research_council_task_failed:BTC-USDT", payload["current_blockers"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|