119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
from server import DashboardService
|
|
|
|
|
|
class DetailDatabaseStub:
|
|
@staticmethod
|
|
def list_watchlist(user_id):
|
|
return []
|
|
|
|
@staticmethod
|
|
def list_notes(user_id, code=""):
|
|
return []
|
|
|
|
|
|
class RealtimeClientStub:
|
|
quote_calls = 0
|
|
|
|
def __init__(self, token):
|
|
self.token = token
|
|
|
|
@staticmethod
|
|
def resolve_trade_context(requested_date):
|
|
return requested_date, requested_date
|
|
|
|
@classmethod
|
|
def realtime_stock_quote(cls, ts_code, reference_date=""):
|
|
cls.quote_calls += 1
|
|
return {
|
|
"name": "测试股票",
|
|
"sector": "测试行业",
|
|
"price": 9.8,
|
|
"change": -2.0,
|
|
"open": 10.1,
|
|
"high": 10.2,
|
|
"low": 9.7,
|
|
"volume": 123400,
|
|
"amount_billion": 1.25,
|
|
"turnover_rate": 3.5,
|
|
}
|
|
|
|
|
|
class FixedMarketDatetime(datetime):
|
|
fixed_now = datetime.now().astimezone().replace(hour=10, minute=30, second=0, microsecond=0)
|
|
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return cls.fixed_now
|
|
|
|
|
|
class StockDetailRealtimeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.service = DashboardService.__new__(DashboardService)
|
|
self.service._system_credentials = {"tushare_token": "test-token"}
|
|
self.service.database = DetailDatabaseStub()
|
|
self.service._request_context = threading.local()
|
|
self.service._request_context.user_id = 1
|
|
RealtimeClientStub.quote_calls = 0
|
|
|
|
def test_today_detail_merges_rt_quote_without_mutating_daily_cache(self):
|
|
today = FixedMarketDatetime.fixed_now.strftime("%Y%m%d")
|
|
yesterday = (FixedMarketDatetime.fixed_now - timedelta(days=1)).strftime("%Y-%m-%d")
|
|
cached = {
|
|
"meta": {"trade_date": today, "source": "tushare"},
|
|
"stock": {"code": "002141", "name": "旧名称", "price": 10, "change": 7.1},
|
|
"prices": [
|
|
{
|
|
"trade_date": yesterday,
|
|
"open": 9.5,
|
|
"high": 10.1,
|
|
"low": 9.4,
|
|
"close": 10,
|
|
"change": 7.1,
|
|
"volume": 100,
|
|
}
|
|
],
|
|
"moneyflow": {},
|
|
}
|
|
|
|
with patch("server.datetime", FixedMarketDatetime), patch(
|
|
"server.TushareClient", RealtimeClientStub
|
|
):
|
|
result = self.service._prepare_stock_detail(cached, "002141", today)
|
|
|
|
self.assertEqual(result["meta"]["trade_date"], FixedMarketDatetime.fixed_now.strftime("%Y-%m-%d"))
|
|
self.assertTrue(result["meta"]["realtime"])
|
|
self.assertEqual(result["stock"]["price"], 9.8)
|
|
self.assertEqual(result["stock"]["change"], -2.0)
|
|
self.assertEqual(result["prices"][-1]["change"], -2.0)
|
|
self.assertEqual(result["prices"][-1]["trade_date"], FixedMarketDatetime.fixed_now.strftime("%Y-%m-%d"))
|
|
self.assertEqual(cached["stock"]["change"], 7.1)
|
|
self.assertEqual(len(cached["prices"]), 1)
|
|
self.assertEqual(RealtimeClientStub.quote_calls, 1)
|
|
|
|
def test_historical_detail_never_requests_realtime_quote(self):
|
|
historical = (FixedMarketDatetime.fixed_now - timedelta(days=5)).strftime("%Y%m%d")
|
|
payload = {
|
|
"meta": {"trade_date": historical, "source": "tushare"},
|
|
"stock": {"code": "002141", "price": 10, "change": 1.2},
|
|
"prices": [{"trade_date": historical, "close": 10, "change": 1.2}],
|
|
}
|
|
with patch("server.datetime", FixedMarketDatetime), patch(
|
|
"server.TushareClient", RealtimeClientStub
|
|
):
|
|
result = self.service._prepare_stock_detail(payload, "002141", historical)
|
|
|
|
self.assertEqual(result["stock"]["change"], 1.2)
|
|
self.assertNotIn("realtime", result["meta"])
|
|
self.assertEqual(RealtimeClientStub.quote_calls, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|