117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import unittest
|
|
|
|
from server import DashboardService
|
|
|
|
|
|
class SnapshotDatabase:
|
|
def __init__(self, snapshot, latest=None):
|
|
self.snapshot = snapshot
|
|
self.latest = latest
|
|
self.aliases = {}
|
|
|
|
def get_snapshot(self, _trade_date):
|
|
return copy.deepcopy(self.snapshot)
|
|
|
|
def reason_overrides(self, _trade_date):
|
|
return {}
|
|
|
|
def get_data_snapshot(self, kind, cache_key):
|
|
return copy.deepcopy(self.aliases.get((kind, cache_key)))
|
|
|
|
def save_data_snapshot(self, kind, cache_key, _source, payload):
|
|
self.aliases[(kind, cache_key)] = copy.deepcopy(payload)
|
|
|
|
def save_snapshot(self, _trade_date, _source, payload):
|
|
self.snapshot = copy.deepcopy(payload)
|
|
|
|
def get_latest_real_snapshot(self, _trade_date, strictly_before=False):
|
|
return copy.deepcopy(self.latest)
|
|
|
|
|
|
class DashboardCacheTests(unittest.TestCase):
|
|
def service(self, snapshot):
|
|
service = object.__new__(DashboardService)
|
|
service.database = SnapshotDatabase(snapshot)
|
|
return service
|
|
|
|
def test_cached_dashboard_skips_sentiment_rebuild_when_fields_are_complete(self):
|
|
snapshot = {
|
|
"meta": {"source": "tushare", "trade_date": "2026-07-22"},
|
|
"overview": {
|
|
"sentiment_score": 32,
|
|
"sentiment_label": "weak",
|
|
"sentiment_phase": "retreat",
|
|
"sentiment_direction": "cooling",
|
|
"sentiment_components": {},
|
|
"sentiment_engine_version": 2,
|
|
},
|
|
}
|
|
service = self.service(snapshot)
|
|
service._enrich_dashboard_sentiment = lambda *_args: self.fail(
|
|
"complete cached sentiment must not be rebuilt"
|
|
)
|
|
|
|
payload = service.get_dashboard("2026-07-22")
|
|
|
|
self.assertTrue(payload["meta"]["cached"])
|
|
self.assertEqual(payload["overview"]["sentiment_score"], 32)
|
|
|
|
def test_cached_dashboard_rebuilds_legacy_snapshot_missing_sentiment(self):
|
|
snapshot = {
|
|
"meta": {"source": "tushare", "trade_date": "2026-07-22"},
|
|
"overview": {"limit_up_count": 20},
|
|
}
|
|
service = self.service(snapshot)
|
|
calls = []
|
|
|
|
def enrich(payload, trade_date):
|
|
calls.append(trade_date)
|
|
payload["overview"].update({
|
|
"sentiment_score": 20,
|
|
"sentiment_label": "weak",
|
|
"sentiment_phase": "ice",
|
|
"sentiment_direction": "cooling",
|
|
"sentiment_components": {},
|
|
"sentiment_engine_version": 2,
|
|
})
|
|
return payload
|
|
|
|
service._enrich_dashboard_sentiment = enrich
|
|
|
|
payload = service.get_dashboard("2026-07-22")
|
|
|
|
self.assertEqual(calls, ["20260722"])
|
|
self.assertEqual(payload["overview"]["sentiment_phase"], "ice")
|
|
|
|
def test_weekend_dashboard_reuses_latest_close_without_external_sync(self):
|
|
latest = {
|
|
"meta": {"source": "tushare", "trade_date": "2026-07-24"},
|
|
"overview": {
|
|
"sentiment_score": 32,
|
|
"sentiment_label": "weak",
|
|
"sentiment_phase": "retreat",
|
|
"sentiment_direction": "cooling",
|
|
"sentiment_components": {},
|
|
},
|
|
}
|
|
service = object.__new__(DashboardService)
|
|
service.database = SnapshotDatabase(None, latest)
|
|
service.sync_dashboard = lambda *_args: self.fail(
|
|
"weekend refresh must not call the external synchronization path"
|
|
)
|
|
|
|
first = service.get_dashboard("2026-07-25")
|
|
service.database.latest = None
|
|
second = service.get_dashboard("2026-07-25")
|
|
|
|
self.assertTrue(first["meta"]["carried_forward"])
|
|
self.assertEqual(first["meta"]["trade_date"], "2026-07-24")
|
|
self.assertEqual(second["meta"]["requested_date"], "2026-07-25")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|