Codex
107 lines
4.6 KiB
Python
107 lines
4.6 KiB
Python
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from src.services.readiness.shadow_paper_sample_collection_service import ShadowPaperSampleCollectionService
|
|
|
|
|
|
class ShadowPaperSampleCollectionServiceTests(unittest.TestCase):
|
|
def test_records_daily_observation_from_latest_candidate_and_packet(self) -> None:
|
|
now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
|
|
repository = _ObservationRepository()
|
|
service = ShadowPaperSampleCollectionService(repository=repository)
|
|
|
|
payload = service.record_daily_observation(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
operator_agreement="agree",
|
|
operator_note="aligned with operator review",
|
|
created_by="alice",
|
|
runtime_status={"current_blockers": []},
|
|
now=now,
|
|
)
|
|
|
|
self.assertEqual(payload["status"], "recorded")
|
|
self.assertEqual(payload["observation_status"], "operator_feedback_recorded")
|
|
self.assertEqual(payload["system_recommendation"], "wait_for_operator_confirmation")
|
|
self.assertEqual(payload["operator_agreement"], "agree")
|
|
self.assertEqual(payload["candidate_id"], 7)
|
|
self.assertEqual(payload["decision_packet_id"], 9)
|
|
self.assertIn("keep_daily_shadow_observation_running", payload["next_collection_actions"])
|
|
self.assertIn("build_shadow_paper_ticket_for_latest_candidate", payload["next_collection_actions"])
|
|
self.assertEqual(repository.upsert_call["created_by"], "alice")
|
|
self.assertEqual(repository.upsert_call["reason_codes"][0], "operator_feedback_recorded")
|
|
|
|
def test_runtime_blocker_records_blocked_observation_without_sample_credit_action(self) -> None:
|
|
now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
|
|
service = ShadowPaperSampleCollectionService(repository=_ObservationRepository())
|
|
|
|
payload = service.record_daily_observation(
|
|
instrument_id=1,
|
|
runtime_status={"current_blockers": ["broker_sync_degraded:BTC-USDT"]},
|
|
now=now,
|
|
)
|
|
|
|
self.assertEqual(payload["observation_status"], "runtime_blocked")
|
|
self.assertEqual(payload["system_recommendation"], "pause_runtime_blocked")
|
|
self.assertEqual(payload["next_collection_actions"], ["clear_runtime_blockers_before_counting_shadow_day"])
|
|
|
|
def test_blocked_candidate_waits_before_paper_ticket_action(self) -> None:
|
|
now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
|
|
service = ShadowPaperSampleCollectionService(repository=_ObservationRepository(candidate_status="blocked"))
|
|
|
|
payload = service.record_daily_observation(
|
|
instrument_id=1,
|
|
runtime_status={"current_blockers": []},
|
|
now=now,
|
|
)
|
|
|
|
self.assertIn("wait_for_executable_candidate_before_paper_ticket", payload["next_collection_actions"])
|
|
self.assertNotIn("build_shadow_paper_ticket_for_latest_candidate", payload["next_collection_actions"])
|
|
|
|
|
|
class _ObservationRepository:
|
|
def __init__(self, *, candidate_status: str = "executable") -> None:
|
|
self.upsert_call = None
|
|
self.candidate_status = candidate_status
|
|
|
|
def fetch_latest_candidate(self, **kwargs):
|
|
return {
|
|
"id": 7,
|
|
"model_code": "SWEEP_MSS_FVG",
|
|
"status": self.candidate_status,
|
|
"session_code": "NY_AM",
|
|
"confirmation_state": {"reason_codes": ["mss_confirmed"]},
|
|
"created_ts": kwargs["since"],
|
|
}
|
|
|
|
def fetch_latest_decision_packet(self, **kwargs):
|
|
return {
|
|
"id": 9,
|
|
"status": "actionable",
|
|
"action_recommendation": "confirm_ticket",
|
|
"payload": {"decision": {"reason_codes": ["candidate_actionable"]}},
|
|
"packet_ts": kwargs["since"],
|
|
}
|
|
|
|
def upsert_daily_observation(self, **kwargs):
|
|
self.upsert_call = kwargs
|
|
return {
|
|
"id": 12,
|
|
"observation_date": kwargs["observation_date"],
|
|
"timeframe": kwargs["timeframe"],
|
|
"session_code": kwargs["session_code"],
|
|
"observation_status": kwargs["observation_status"],
|
|
"system_recommendation": kwargs["system_recommendation"],
|
|
"operator_agreement": kwargs["operator_agreement"],
|
|
"candidate_id": kwargs["candidate_id"],
|
|
"decision_packet_id": kwargs["decision_packet_id"],
|
|
"paper_ticket_id": kwargs["paper_ticket_id"],
|
|
"paper_execution_id": kwargs["paper_execution_id"],
|
|
"reason_codes": kwargs["reason_codes"],
|
|
"observed_ts": kwargs["observed_ts"],
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|