import json import logging import time from datetime import datetime from unittest.mock import MagicMock, patch import pytest from app.storage_manager import StorageManager, get_storage_manager class TestStorageManagerSkeleton: """StorageManager 骨架与降级检测测试。""" def test_init_holds_client_and_engine_references(self): """__init__ 应持有 redis_client / mysql_engine 引用。""" redis_client = MagicMock() mysql_engine = MagicMock() manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) assert manager.redis_client is redis_client assert manager.mysql_engine is mysql_engine def test_init_sets_default_flags(self): """初始化时应将可用性标志设为 False。""" manager = StorageManager(redis_client=None, mysql_engine=None) assert manager.redis_available is False assert manager.mysql_available is False def test_check_redis_returns_cached_result_within_interval(self): """30 秒间隔内应返回缓存的 Redis 可用性结果，不重新探测。""" redis_client = MagicMock() redis_client.ping.side_effect = Exception("redis down") manager = StorageManager(redis_client=redis_client, mysql_engine=None) first = manager.check_redis() redis_client.ping.side_effect = None second = manager.check_redis() assert first is False assert second is False assert redis_client.ping.call_count == 1 def test_check_redis_recovers_after_interval(self): """超过检测间隔后，Redis 恢复应返回 True。""" redis_client = MagicMock() redis_client.ping.side_effect = Exception("redis down") manager = StorageManager( redis_client=redis_client, mysql_engine=None, check_interval=0.1 ) assert manager.check_redis() is False redis_client.ping.side_effect = None time.sleep(0.15) assert manager.check_redis() is True def test_check_redis_false_when_client_is_none(self): """redis_client 为 None 时 check_redis 返回 False。""" manager = StorageManager(redis_client=None, mysql_engine=None) assert manager.check_redis() is False def test_check_mysql_returns_cached_result_within_interval(self): """30 秒间隔内应返回缓存的 MySQL 可用性结果，不重新探测。""" mysql_engine = MagicMock() mysql_engine.connect.side_effect = Exception("mysql down") manager = StorageManager(redis_client=None, mysql_engine=mysql_engine) first = manager.check_mysql() mysql_engine.connect.side_effect = None second = manager.check_mysql() assert first is False assert second is False assert mysql_engine.connect.call_count == 1 def test_check_mysql_recovers_after_interval(self): """超过检测间隔后，MySQL 恢复应返回 True。""" mysql_engine = MagicMock() mysql_engine.connect.side_effect = Exception("mysql down") manager = StorageManager( redis_client=None, mysql_engine=mysql_engine, check_interval=0.1 ) assert manager.check_mysql() is False mysql_engine.connect.side_effect = None time.sleep(0.15) assert manager.check_mysql() is True def test_check_mysql_false_when_engine_is_none(self): """mysql_engine 为 None 时 check_mysql 返回 False。""" manager = StorageManager(redis_client=None, mysql_engine=None) assert manager.check_mysql() is False def test_is_available_true_when_redis_available(self): """Redis 可用时 is_available 返回 True。""" redis_client = MagicMock() manager = StorageManager(redis_client=redis_client, mysql_engine=None) assert manager.is_available() is True def test_is_available_true_when_mysql_available(self): """MySQL 可用时 is_available 返回 True。""" mysql_engine = MagicMock() manager = StorageManager(redis_client=None, mysql_engine=mysql_engine) assert manager.is_available() is True def test_is_available_false_when_both_unavailable(self): """Redis 与 MySQL 均不可用时 is_available 返回 False。""" redis_client = MagicMock() redis_client.ping.side_effect = Exception("redis down") mysql_engine = MagicMock() mysql_engine.connect.side_effect = Exception("mysql down") manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) assert manager.is_available() is False def test_initialize_checks_backends_and_logs(self, caplog): """initialize 应检测后端可用性并输出日志。""" redis_client = MagicMock() mysql_engine = MagicMock() manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) with caplog.at_level(logging.INFO): manager.initialize() assert "Redis 可用" in caplog.text assert "MySQL 可用" in caplog.text def test_initialize_logs_fallback_when_unavailable(self, caplog): """后端不可用时 initialize 应输出降级日志。""" manager = StorageManager(redis_client=None, mysql_engine=None) with caplog.at_level(logging.WARNING): manager.initialize() assert "Redis 不可用" in caplog.text assert "MySQL 不可用" in caplog.text class TestStorageManagerSingleton: """全局单例工厂函数测试。""" def test_get_storage_manager_returns_same_instance(self): """多次调用 get_storage_manager 应返回同一实例。""" first = get_storage_manager() second = get_storage_manager() assert first is second def test_get_storage_manager_returns_storage_manager(self): """get_storage_manager 应返回 StorageManager 实例。""" manager = get_storage_manager() assert isinstance(manager, StorageManager) class TestStorageManagerReadLogic: """StorageManager Redis → MySQL 读取逻辑测试。""" def test_get_market_data_returns_redis_value_on_cache_hit(self): """Redis 命中时直接返回反序列化数据，不访问 MySQL。""" redis_client = MagicMock() mysql_engine = MagicMock() expected = { "current_price": 123.45, "timestamp": "2026-07-04T10:00:00", "candles": [{"open": 100, "close": 110}], } redis_client.get.return_value = json.dumps(expected) manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) result = manager.get_market_data("AG2606", "futures", "5min") assert result == expected redis_client.get.assert_called_once_with("market_data:AG2606:5min") mysql_engine.connect.assert_not_called() def test_get_market_data_falls_back_to_mysql_and_backfills_redis(self): """Redis 未命中时从 MySQL 读取并回填 Redis（TTL 30 天）。""" redis_client = MagicMock() redis_client.get.return_value = None 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 mock_sessionmaker.assert_called_once_with(bind=mysql_engine) mock_query.filter_by.assert_called_once_with( symbol="AG2606", data_type="futures", period="5min" ) 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_unavailable(self): """Redis 和 MySQL 均不可用时返回 None，由调用方降级到 SQLite。""" redis_client = MagicMock() redis_client.ping.side_effect = Exception("redis down") mysql_engine = MagicMock() mysql_engine.connect.side_effect = Exception("mysql down") manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) result = manager.get_market_data("AG2606", "futures", "5min") 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() mysql_engine = MagicMock() expected = { "last_refresh_at": "2026-07-04T10:00:00", "refresh_count": 42, } redis_client.get.return_value = json.dumps(expected) manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) result = manager.get_symbol_timestamp("AG2606", "futures") assert result == expected redis_client.get.assert_called_once_with("symbol_timestamps:AG2606") mysql_engine.connect.assert_not_called() def test_get_symbol_timestamp_falls_back_to_mysql_and_backfills_redis(self): """Redis 未命中时从 MySQL 读取时间戳并回填 Redis（TTL 30 天）。""" redis_client = MagicMock() redis_client.get.return_value = None mysql_engine = MagicMock() manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) last_refresh = datetime(2026, 7, 4, 10, 0, 0) record = MagicMock() record.last_refresh_at = last_refresh record.refresh_count = 42 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_symbol_timestamp("AG2606", "futures") expected = { "last_refresh_at": "2026-07-04T10:00:00", "refresh_count": 42, } assert result == expected mock_sessionmaker.assert_called_once_with(bind=mysql_engine) mock_query.filter_by.assert_called_once_with(symbol="AG2606", data_type="futures") redis_client.setex.assert_called_once_with( "symbol_timestamps:AG2606", 2592000, json.dumps(expected, ensure_ascii=False) ) class TestStorageManagerWriteLogic: """StorageManager 双写一致性测试。""" def test_save_market_data_deletes_cache_writes_mysql_and_updates_redis(self): """MySQL 写入成功时，应先删缓存、再写 MySQL、最后回填 Redis。""" redis_client = MagicMock() 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}], "15min": [{"open": 105, "close": 115}], }, } 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") as mock_datetime: mock_datetime.now.return_value = fetched_at result = manager.save_market_data("AG2606", data) assert result is True assert len(records) == 2 assert records[0].symbol == "AG2606" assert records[0].data_type == "futures" assert records[0].period == "5min" assert records[0].current_price == 123.45 assert records[0].candle_count == 1 assert records[0].fetched_at == fetched_at redis_client.delete.assert_called_once_with( "market_data:AG2606:5min", "market_data:AG2606:15min" ) redis_client.setex.assert_any_call( "market_data:AG2606:5min", 2592000, json.dumps( { "current_price": 123.45, "timestamp": fetched_at.isoformat(), "candles": [{"open": 100, "close": 110}], }, ensure_ascii=False, ), ) redis_client.setex.assert_any_call( "market_data:AG2606:15min", 2592000, json.dumps( { "current_price": 123.45, "timestamp": fetched_at.isoformat(), "candles": [{"open": 105, "close": 115}], }, ensure_ascii=False, ), ) def test_save_market_data_returns_false_and_skips_redis_on_mysql_failure(self): """MySQL 写入失败时应回滚事务，不更新 Redis，并返回 False。""" 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": {"5min": [{"open": 100, "close": 110}]}, } mock_session = MagicMock() mock_session.return_value = mock_session mock_session.commit.side_effect = Exception("mysql write failed") 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 False redis_client.delete.assert_called_once_with("market_data:AG2606:5min") redis_client.setex.assert_not_called() mock_session.rollback.assert_called_once() def test_save_market_data_returns_true_when_redis_update_fails(self, caplog): """MySQL 成功但 Redis 回填失败时，应返回 True 并记录警告日志。""" redis_client = MagicMock() redis_client.setex.side_effect = Exception("redis write failed") 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}]}, } mock_session = MagicMock() mock_session.return_value = mock_session 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") as mock_datetime: mock_datetime.now.return_value = fetched_at with caplog.at_level(logging.WARNING): result = manager.save_market_data("AG2606", data) assert result is True 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() mysql_engine = MagicMock() manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) refresh_time = datetime(2026, 7, 4, 10, 0, 0) records = [] def mock_filter_by(symbol=None, data_type=None): existing = next( (r for r in records if r.symbol == symbol and r.data_type == data_type), 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_symbol_timestamp("AG2606", "futures", refresh_time) assert result is True assert len(records) == 1 assert records[0].symbol == "AG2606" assert records[0].data_type == "futures" assert records[0].last_refresh_at == refresh_time assert records[0].refresh_count == 1 redis_client.delete.assert_called_once_with("symbol_timestamps:AG2606") redis_client.setex.assert_called_once_with( "symbol_timestamps:AG2606", 2592000, json.dumps( {"last_refresh_at": refresh_time.isoformat(), "refresh_count": 1}, ensure_ascii=False, ), ) def test_save_symbol_timestamp_returns_false_on_mysql_failure(self): """合约时间戳 MySQL 写入失败时应返回 False 且不更新 Redis。""" redis_client = MagicMock() mysql_engine = MagicMock() manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine) mock_session = MagicMock() mock_session.return_value = mock_session mock_session.commit.side_effect = Exception("mysql write failed") 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_symbol_timestamp( "AG2606", "futures", datetime(2026, 7, 4, 10, 0, 0) ) assert result is False redis_client.delete.assert_called_once_with("symbol_timestamps:AG2606") redis_client.setex.assert_not_called() mock_session.rollback.assert_called_once() def test_delete_cache_removes_market_data_keys(self): """delete_cache 应批量删除指定周期的 Redis 缓存键。""" redis_client = MagicMock() manager = StorageManager(redis_client=redis_client, mysql_engine=None) manager.delete_cache("AG2606", ["5min", "15min"]) redis_client.delete.assert_called_once_with( "market_data:AG2606:5min", "market_data:AG2606:15min" )