Codex
124 lines
5.3 KiB
Python
124 lines
5.3 KiB
Python
import unittest
|
|
from datetime import date, datetime, timezone
|
|
|
|
from src.services.readiness.shadow_paper_observation_backfill_service import ShadowPaperObservationBackfillService
|
|
|
|
|
|
class ShadowPaperObservationBackfillServiceTests(unittest.TestCase):
|
|
def test_dry_run_finds_candidate_days_without_writing(self) -> None:
|
|
repository = _BackfillRepository()
|
|
service = ShadowPaperObservationBackfillService(repository=repository)
|
|
|
|
payload = service.build_backfill_audit(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
now=datetime(2026, 5, 18, 12, 0, tzinfo=timezone.utc),
|
|
apply=False,
|
|
)
|
|
|
|
self.assertEqual(payload["status"], "dry_run")
|
|
self.assertEqual(payload["eligible_backfill_dates"], ["2026-05-15", "2026-05-16"])
|
|
self.assertEqual(payload["created_observation_count"], 0)
|
|
self.assertEqual(repository.insert_calls, [])
|
|
self.assertFalse(payload["paper_backfill_allowed"])
|
|
self.assertFalse(payload["operator_feedback_backfill_allowed"])
|
|
self.assertIn("paper_ticket_backfill_not_allowed_without_executable_candidate", payload["blocked_backfill_items"])
|
|
|
|
def test_apply_creates_only_shadow_observations(self) -> None:
|
|
repository = _BackfillRepository()
|
|
service = ShadowPaperObservationBackfillService(repository=repository)
|
|
|
|
payload = service.build_backfill_audit(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
now=datetime(2026, 5, 18, 12, 0, tzinfo=timezone.utc),
|
|
apply=True,
|
|
created_by="test_backfill",
|
|
)
|
|
|
|
self.assertEqual(payload["status"], "applied")
|
|
self.assertEqual(payload["created_observation_count"], 2)
|
|
self.assertEqual([call["observation_date"].isoformat() for call in repository.insert_calls], ["2026-05-15", "2026-05-16"])
|
|
self.assertTrue(all(call["candidate"]["status"] == "blocked" for call in repository.insert_calls))
|
|
self.assertTrue(all(call["created_by"] == "test_backfill" for call in repository.insert_calls))
|
|
self.assertTrue(all(row["paper_ticket_id"] is None for row in payload["created_observations"]))
|
|
self.assertTrue(all(row["paper_execution_id"] is None for row in payload["created_observations"]))
|
|
self.assertTrue(all(row["operator_agreement"] == "pending" for row in payload["created_observations"]))
|
|
|
|
def test_no_candidate_days_has_no_eligible_backfill(self) -> None:
|
|
repository = _BackfillRepository(candidate_rows=[])
|
|
service = ShadowPaperObservationBackfillService(repository=repository)
|
|
|
|
payload = service.build_backfill_audit(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
now=datetime(2026, 5, 18, 12, 0, tzinfo=timezone.utc),
|
|
apply=True,
|
|
)
|
|
|
|
self.assertEqual(payload["eligible_backfill_dates"], [])
|
|
self.assertEqual(payload["created_observation_count"], 0)
|
|
self.assertEqual(repository.insert_calls, [])
|
|
|
|
|
|
class _BackfillRepository:
|
|
def __init__(self, *, candidate_rows=None) -> None:
|
|
self.candidate_rows = candidate_rows
|
|
self.insert_calls = []
|
|
|
|
def fetch_candidate_day_summaries(self, **kwargs):
|
|
if self.candidate_rows is not None:
|
|
return self.candidate_rows
|
|
return [
|
|
{"day": date(2026, 5, 15), "status": "blocked", "count": 3},
|
|
{"day": date(2026, 5, 16), "status": "blocked", "count": 4},
|
|
{"day": date(2026, 5, 17), "status": "blocked", "count": 5},
|
|
]
|
|
|
|
def fetch_observation_day_summaries(self, **kwargs):
|
|
return [{"day": date(2026, 5, 17), "count": 1, "feedback_count": 1}]
|
|
|
|
def fetch_latest_candidate_for_day(self, **kwargs):
|
|
return {
|
|
"id": int(kwargs["day_start"].strftime("%d")),
|
|
"instrument_id": kwargs["instrument_id"],
|
|
"model_code": "SWEEP_MSS_FVG",
|
|
"status": "blocked",
|
|
"session_code": "ALL",
|
|
"created_ts": kwargs["day_start"],
|
|
"quality_score": 0.5,
|
|
}
|
|
|
|
def fetch_latest_decision_packet_for_day(self, **kwargs):
|
|
return {
|
|
"id": int(kwargs["day_start"].strftime("%d")) + 100,
|
|
"instrument_id": kwargs["instrument_id"],
|
|
"candidate_id": kwargs["candidate_id"],
|
|
"status": "blocked",
|
|
"action_recommendation": "stand_down",
|
|
"packet_ts": kwargs["day_start"],
|
|
}
|
|
|
|
def insert_shadow_observation(self, **kwargs):
|
|
self.insert_calls.append(kwargs)
|
|
return {
|
|
"id": len(self.insert_calls),
|
|
"instrument_id": kwargs["instrument_id"],
|
|
"observation_date": kwargs["observation_date"],
|
|
"timeframe": kwargs["timeframe"],
|
|
"session_code": kwargs["session_code"],
|
|
"candidate_id": kwargs["candidate"]["id"],
|
|
"decision_packet_id": kwargs["decision_packet"]["id"],
|
|
"paper_ticket_id": None,
|
|
"paper_execution_id": None,
|
|
"observation_status": "candidate_observed",
|
|
"system_recommendation": "stand_down_blocked_candidate",
|
|
"operator_agreement": "pending",
|
|
"reason_codes": ["shadow_observation_backfill"],
|
|
"observed_ts": kwargs["candidate"]["created_ts"],
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|