From 59e461a163bc97e6cf5ee0705d3a225c94b7d42d Mon Sep 17 00:00:00 2001 From: Lxy Date: Sat, 4 Jul 2026 10:27:50 +0800 Subject: [PATCH] fix(storage): use input timestamp in save and expand test coverage --- app/storage_manager.py | 19 ++- tests/test_storage_manager.py | 256 ++++++++++++++++++++++++++++++++++ 2 files changed, 274 insertions(+), 1 deletion(-) diff --git a/app/storage_manager.py b/app/storage_manager.py index 57d783e..d690935 100644 --- a/app/storage_manager.py +++ b/app/storage_manager.py @@ -1,3 +1,4 @@ +import datetime as _datetime import json import logging import time @@ -64,6 +65,22 @@ class StorageManager: self.last_mysql_check = now return self.mysql_available + def _parse_fetched_at(self, timestamp_raw): + """优先使用输入 timestamp，解析失败时回退到当前时间。""" + if not timestamp_raw: + return _datetime.datetime.now() + + if isinstance(timestamp_raw, str): + try: + return _datetime.datetime.fromisoformat(timestamp_raw) + except Exception: + return _datetime.datetime.now() + + if hasattr(timestamp_raw, "isoformat"): + return timestamp_raw + + return _datetime.datetime.now() + def is_available(self): """判断 Redis 或 MySQL 至少一个可用。""" return self.check_redis() or self.check_mysql() @@ -225,7 +242,7 @@ class StorageManager: data_type = data.get("type", "futures") timeframes = data.get("timeframes", {}) current_price = data.get("current_price") - fetched_at = datetime.now() + fetched_at = self._parse_fetched_at(data.get("timestamp")) cache_keys = [f"market_data:{symbol}:{period}" for period in timeframes.keys()] diff --git a/tests/test_storage_manager.py b/tests/test_storage_manager.py index c9df767..49bb640 100644 --- a/tests/test_storage_manager.py +++ b/tests/test_storage_manager.py @@ -237,6 +237,62 @@ class TestStorageManagerReadLogic: assert result is None + def test_get_market_data_falls_back_to_mysql_when_redis_read_fails(self): + """Redis 读取失败但 MySQL 可用时，应降级到 MySQL 并回填 Redis。""" + redis_client = MagicMock() + redis_client.get.side_effect = Exception("redis read failed") + mysql_engine = MagicMock() + manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) + + fetched_at = datetime(2026, 7, 4, 10, 0, 0) + record = MagicMock() + record.current_price = 123.45 + record.fetched_at = fetched_at + record.candles_json = json.dumps([{"open": 100, "close": 110}]) + + mock_query = MagicMock() + mock_query.filter_by.return_value.first.return_value = record + mock_session = MagicMock() + mock_session.return_value = mock_session + mock_session.query.return_value = mock_query + mock_session.__enter__ = MagicMock(return_value=mock_session) + mock_session.__exit__ = MagicMock(return_value=False) + + mock_sessionmaker = MagicMock(return_value=mock_session) + + with patch("app.storage_manager.sessionmaker", mock_sessionmaker): + result = manager.get_market_data("AG2606", "futures", "5min") + + expected = { + "current_price": 123.45, + "timestamp": "2026-07-04T10:00:00", + "candles": [{"open": 100, "close": 110}], + } + assert result == expected + redis_client.setex.assert_called_once_with( + "market_data:AG2606:5min", 2592000, json.dumps(expected, ensure_ascii=False) + ) + + def test_get_market_data_returns_none_when_mysql_read_fails(self): + """Redis 不可用且 MySQL 读取失败时返回 None。""" + redis_client = MagicMock() + redis_client.ping.side_effect = Exception("redis down") + mysql_engine = MagicMock() + manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) + + mock_session = MagicMock() + mock_session.return_value = mock_session + mock_session.query.side_effect = Exception("mysql read failed") + mock_session.__enter__ = MagicMock(return_value=mock_session) + mock_session.__exit__ = MagicMock(return_value=False) + + mock_sessionmaker = MagicMock(return_value=mock_session) + + with patch("app.storage_manager.sessionmaker", mock_sessionmaker): + result = manager.get_market_data("AG2606", "futures", "5min") + + assert result is None + def test_get_symbol_timestamp_returns_redis_value_on_cache_hit(self): """Redis 命中时直接返回合约时间戳数据，不访问 MySQL。""" redis_client = MagicMock() @@ -442,6 +498,206 @@ class TestStorageManagerWriteLogic: mock_session.commit.assert_called_once() assert "Redis 更新失败" in caplog.text + def test_save_market_data_uses_input_timestamp(self): + """save_market_data 应优先使用输入数据中的 timestamp，而非 datetime.now()。""" + redis_client = MagicMock() + mysql_engine = MagicMock() + manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) + + input_timestamp = datetime(2026, 1, 15, 8, 30, 0) + data = { + "type": "futures", + "current_price": 123.45, + "timestamp": input_timestamp.isoformat(), + "timeframes": {"5min": [{"open": 100, "close": 110}]}, + } + + records = [] + + def mock_filter_by(symbol=None, data_type=None, period=None): + existing = next( + ( + r + for r in records + if r.symbol == symbol and r.data_type == data_type and r.period == period + ), + None, + ) + fb_query = MagicMock() + fb_query.first.return_value = existing + return fb_query + + mock_query = MagicMock() + mock_query.filter_by.side_effect = mock_filter_by + + mock_session = MagicMock() + mock_session.return_value = mock_session + mock_session.query.return_value = mock_query + mock_session.add.side_effect = records.append + mock_session.commit = MagicMock() + mock_session.rollback = MagicMock() + mock_session.__enter__ = MagicMock(return_value=mock_session) + mock_session.__exit__ = MagicMock(return_value=False) + + mock_sessionmaker = MagicMock(return_value=mock_session) + + with patch("app.storage_manager.sessionmaker", mock_sessionmaker): + result = manager.save_market_data("AG2606", data) + + assert result is True + assert len(records) == 1 + assert records[0].fetched_at == input_timestamp + redis_client.setex.assert_called_once_with( + "market_data:AG2606:5min", + 2592000, + json.dumps( + { + "current_price": 123.45, + "timestamp": input_timestamp.isoformat(), + "candles": [{"open": 100, "close": 110}], + }, + ensure_ascii=False, + ), + ) + + def test_save_market_data_falls_back_to_now_on_invalid_timestamp(self): + """timestamp 无法解析时应回退到当前时间。""" + redis_client = MagicMock() + mysql_engine = MagicMock() + manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) + + fallback_time = datetime(2026, 3, 1, 12, 0, 0) + data = { + "type": "futures", + "current_price": 123.45, + "timestamp": "not-a-valid-timestamp", + "timeframes": {"5min": [{"open": 100, "close": 110}]}, + } + + records = [] + + def mock_filter_by(symbol=None, data_type=None, period=None): + existing = next( + ( + r + for r in records + if r.symbol == symbol and r.data_type == data_type and r.period == period + ), + None, + ) + fb_query = MagicMock() + fb_query.first.return_value = existing + return fb_query + + mock_query = MagicMock() + mock_query.filter_by.side_effect = mock_filter_by + + mock_session = MagicMock() + mock_session.return_value = mock_session + mock_session.query.return_value = mock_query + mock_session.add.side_effect = records.append + mock_session.commit = MagicMock() + mock_session.rollback = MagicMock() + mock_session.__enter__ = MagicMock(return_value=mock_session) + mock_session.__exit__ = MagicMock(return_value=False) + + mock_sessionmaker = MagicMock(return_value=mock_session) + + with patch("app.storage_manager.sessionmaker", mock_sessionmaker): + with patch("app.storage_manager._datetime.datetime") as mock_datetime: + mock_datetime.now.return_value = fallback_time + mock_datetime.fromisoformat.side_effect = ValueError( + "Invalid isoformat string" + ) + result = manager.save_market_data("AG2606", data) + + assert result is True + assert len(records) == 1 + assert records[0].fetched_at == fallback_time + + def test_save_market_data_writes_mysql_when_redis_unavailable(self): + """Redis 不可用但 MySQL 可用时，save_market_data 应直接写入 MySQL 并返回 True。""" + redis_client = MagicMock() + redis_client.ping.side_effect = Exception("redis down") + mysql_engine = MagicMock() + manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) + + fetched_at = datetime(2026, 7, 4, 10, 0, 0) + data = { + "type": "futures", + "current_price": 123.45, + "timestamp": fetched_at.isoformat(), + "timeframes": {"5min": [{"open": 100, "close": 110}]}, + } + + records = [] + + def mock_filter_by(symbol=None, data_type=None, period=None): + existing = next( + ( + r + for r in records + if r.symbol == symbol and r.data_type == data_type and r.period == period + ), + None, + ) + fb_query = MagicMock() + fb_query.first.return_value = existing + return fb_query + + mock_query = MagicMock() + mock_query.filter_by.side_effect = mock_filter_by + + mock_session = MagicMock() + mock_session.return_value = mock_session + mock_session.query.return_value = mock_query + mock_session.add.side_effect = records.append + mock_session.commit = MagicMock() + mock_session.rollback = MagicMock() + mock_session.__enter__ = MagicMock(return_value=mock_session) + mock_session.__exit__ = MagicMock(return_value=False) + + mock_sessionmaker = MagicMock(return_value=mock_session) + + with patch("app.storage_manager.sessionmaker", mock_sessionmaker): + result = manager.save_market_data("AG2606", data) + + assert result is True + assert len(records) == 1 + assert records[0].symbol == "AG2606" + redis_client.delete.assert_not_called() + redis_client.setex.assert_not_called() + + def test_save_market_data_returns_true_when_timeframes_empty(self): + """timeframes 为空时，save_market_data 应直接返回 True 且不操作缓存。""" + redis_client = MagicMock() + mysql_engine = MagicMock() + manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) + + data = { + "type": "futures", + "current_price": 123.45, + "timestamp": datetime(2026, 7, 4, 10, 0, 0).isoformat(), + "timeframes": {}, + } + + mock_session = MagicMock() + mock_session.return_value = mock_session + mock_session.commit = MagicMock() + mock_session.rollback = MagicMock() + mock_session.__enter__ = MagicMock(return_value=mock_session) + mock_session.__exit__ = MagicMock(return_value=False) + + mock_sessionmaker = MagicMock(return_value=mock_session) + + with patch("app.storage_manager.sessionmaker", mock_sessionmaker): + result = manager.save_market_data("AG2606", data) + + assert result is True + mock_session.commit.assert_called_once() + redis_client.delete.assert_not_called() + redis_client.setex.assert_not_called() + def test_save_symbol_timestamp_deletes_cache_writes_mysql_and_updates_redis(self): """合约时间戳双写成功时，应先删缓存、再写 MySQL、最后回填 Redis。""" redis_client = MagicMock()