from __future__ import annotations import threading import unittest from datetime import datetime from unittest.mock import patch from backend.data.providers.tushare_client import TushareClient from backend.data.providers.tushare_transport import TushareError from server import DashboardService class FixedDatetime(datetime): fixed_now = datetime(2026, 8, 28, 15, 4).astimezone() @classmethod def now(cls, tz=None): return cls.fixed_now class WindowClient(TushareClient): def __init__(self, token: str = "test-token"): super().__init__(token) self.calls: list[str] = [] self.daily_rows: list[dict] = [] self.rt_k_error: Exception | None = None def query(self, api_name, params=None, fields=""): self.calls.append(api_name) params = params or {} if api_name == "trade_cal": return [ { "cal_date": "20260828", "is_open": 1, "pretrade_date": "20260827", } ] if api_name == "daily": return list(self.daily_rows) if api_name == "rt_k": if self.rt_k_error is not None: raise self.rt_k_error raise AssertionError("rt_k should not be called in this scenario") if api_name in {"limit_list_d", "stock_basic", "stk_limit", "daily_basic"}: return [] raise AssertionError(f"Unexpected API call: {api_name} {params}") def resolve_trade_context(self, requested_date: str): return requested_date, "20260827" class SyncDatabaseStub: def __init__(self, latest=None): self.latest = latest self.snapshots: dict[str, dict] = {} self.sync_runs: list[dict] = [] self._sync_id = 0 def start_sync(self, trade_date: str, source: str) -> int: self._sync_id += 1 self.sync_runs.append( { "id": self._sync_id, "trade_date": trade_date, "source": source, "status": "running", } ) return self._sync_id def finish_sync( self, sync_id: int, status: str, record_count: int = 0, message: str = "", source: str | None = None, ) -> None: for row in self.sync_runs: if row["id"] == sync_id: row.update( { "status": status, "record_count": record_count, "message": message, "source": source or row["source"], } ) return raise AssertionError(f"unknown sync_id {sync_id}") def save_snapshot(self, trade_date: str, source: str, payload: dict) -> None: self.snapshots[trade_date] = {"source": source, "payload": payload} def save_data_snapshot(self, kind: str, cache_key: str, source: str, payload: dict) -> None: return None def get_latest_real_snapshot(self, _trade_date: str, strictly_before: bool = False): return self.latest def reason_overrides(self, _trade_date: str): return {} class DashboardRefreshWindowTests(unittest.TestCase): def setUp(self) -> None: TushareClient._realtime_reference_cache.clear() TushareClient._capital_cache.clear() TushareClient._latest_realtime_market.clear() TushareClient._stock_activity_cache.clear() def _service(self, client: WindowClient, latest=None) -> DashboardService: service = object.__new__(DashboardService) service._system_credentials = {"tushare_token": "test-token"} service.sync_lock = threading.Lock() service.database = SyncDatabaseStub(latest=latest) service.data_gateway = None service._tushare_client = lambda: client service._enrich_dashboard_sentiment = lambda dashboard, _date: dashboard service._apply_reason_overrides = lambda dashboard: dashboard return service def test_should_use_realtime_at_1504(self) -> None: FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 4).astimezone() with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime): self.assertTrue(TushareClient.should_use_realtime("20260828", "20260828")) def test_should_use_daily_at_1505(self) -> None: FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 5).astimezone() with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime): self.assertFalse(TushareClient.should_use_realtime("20260828", "20260828")) def test_after_close_empty_daily_does_not_call_rt_k(self) -> None: FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone() client = WindowClient() client.daily_rows = [] with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime): with self.assertRaises(TushareError): client.dashboard("20260828") self.assertIn("daily", client.calls) self.assertNotIn("rt_k", client.calls) def test_after_close_uses_daily_when_ready(self) -> None: FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone() client = WindowClient() client.daily_rows = [ { "ts_code": "000001.SZ", "trade_date": "20260828", "open": 10, "high": 11, "low": 9.5, "close": 10.5, "pre_close": 10, "pct_chg": 5, "vol": 1000, "amount": 1_000_000, } ] def load_limit_lists(_trade_date): return [] def load_limit_type(_trade_date, _limit_type): return [] client._load_limit_lists = load_limit_lists # type: ignore[method-assign] client._load_limit_type = load_limit_type # type: ignore[method-assign] client._derive_limits = lambda *args, **kwargs: [] # type: ignore[method-assign] with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime): dashboard = client.dashboard("20260828") self.assertIn("daily", client.calls) self.assertNotIn("rt_k", client.calls) self.assertFalse(dashboard["meta"].get("realtime")) self.assertEqual(dashboard["meta"]["trade_date"], "2026-08-28") def test_fallback_old_snapshot_marks_sync_and_job_status_failed(self) -> None: FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone() client = WindowClient() client.daily_rows = [] latest = { "meta": {"source": "tushare", "trade_date": "2026-08-27"}, "overview": {}, "limits": [], "broken": [], "down_limits": [], "yesterday_limits": [], } service = self._service(client, latest=latest) with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime): result = service.sync_dashboard("20260828") self.assertEqual(result["status"], "failed") self.assertTrue(result["meta"]["carried_forward"]) self.assertEqual(result["meta"]["trade_date"], "2026-08-27") self.assertEqual(service.database.sync_runs[-1]["status"], "failed") self.assertNotIn("rt_k", client.calls) def test_closing_snapshot_due_after_1505_when_today_missing(self) -> None: FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone() service = object.__new__(DashboardService) service._system_credentials = {"tushare_token": "test-token"} with patch("backend.features.market.service.datetime", FixedDatetime), patch( "backend.features.market.service.date" ) as fake_date: fake_date.today.return_value = FixedDatetime.fixed_now.date() self.assertTrue(service._closing_snapshot_due("20260828", {})) self.assertFalse( service._closing_snapshot_due( "20260828", { "meta": { "trade_date": "2026-08-28", "realtime": False, } }, ) ) if __name__ == "__main__": unittest.main()