""" cache.py StorageManager 集成与 SQLite 降级逻辑测试 """ import json import logging from datetime import datetime from unittest.mock import MagicMock, patch import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.models import Base, MarketData, SymbolTimestamp @pytest.fixture def db(): """提供已创建表结构的内存 SQLite 会话。""" engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(bind=engine) Session = sessionmaker(bind=engine) session = Session() yield session session.close() @pytest.fixture def sample_data(): """构造与采集脚本格式一致的测试数据。""" return { "type": "futures", "current_price": 123.45, "timestamp": datetime(2026, 7, 4, 10, 0, 0).isoformat(), "timeframes": { "5min": [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}], "15min": [{"datetime": "2026-07-04T09:45:00", "open": 105, "close": 115}], }, } @pytest.fixture def storage_manager_mock(): """构造默认可用的 StorageManager mock。""" manager = MagicMock() manager.is_available.return_value = True manager.save_market_data_with_timestamp.return_value = True manager.save_market_data.return_value = True manager.save_symbol_timestamp.return_value = True manager.get_market_data.return_value = None manager.get_symbol_timestamp.return_value = None return manager def _seed_sqlite_market_data(db, symbol="AG2606", data_type="futures"): """在 SQLite 中写入行情数据（使用当前时间确保缓存未过期）。""" fetched_at = datetime.now() db.add_all( [ MarketData( symbol=symbol, data_type=data_type, period="5min", candles_json=json.dumps( [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}] ), current_price=123.45, fetched_at=fetched_at, candle_count=1, ), MarketData( symbol=symbol, data_type=data_type, period="15min", candles_json=json.dumps( [{"datetime": "2026-07-04T09:45:00", "open": 105, "close": 115}] ), current_price=123.45, fetched_at=fetched_at, candle_count=1, ), ] ) db.commit() def _seed_sqlite_symbol_timestamp(db, symbol="AG2606", data_type="futures"): """在 SQLite 中写入合约时间戳。""" db.add( SymbolTimestamp( symbol=symbol, data_type=data_type, last_refresh_at=datetime(2026, 7, 4, 10, 0, 0), refresh_count=3, ) ) db.commit() class TestSaveMarketData: """save_market_data 双写与降级测试。""" def test_uses_storage_manager_when_available(self, db, sample_data, storage_manager_mock): """StorageManager 可用时通过原子方法写入行情数据与时间戳，并返回代表记录。""" with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.save_market_data(db, "AG2606", sample_data) storage_manager_mock.is_available.assert_called_once() storage_manager_mock.save_market_data_with_timestamp.assert_called_once() call_args = storage_manager_mock.save_market_data_with_timestamp.call_args[0] assert call_args[0] == "AG2606" assert call_args[1] == sample_data assert isinstance(call_args[2], datetime) storage_manager_mock.save_market_data.assert_not_called() storage_manager_mock.save_symbol_timestamp.assert_not_called() assert isinstance(result, MarketData) assert result.symbol == "AG2606" assert result.data_type == "futures" assert result.current_price == 123.45 sqlite_records = db.query(MarketData).filter_by(symbol="AG2606").all() assert len(sqlite_records) == 0 def test_falls_back_to_sqlite_when_storage_unavailable(self, db, sample_data, storage_manager_mock): """StorageManager 不可用时降级到 SQLite 写入。""" storage_manager_mock.is_available.return_value = False with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.save_market_data(db, "AG2606", sample_data) storage_manager_mock.save_market_data.assert_not_called() storage_manager_mock.save_symbol_timestamp.assert_not_called() assert isinstance(result, MarketData) sqlite_records = db.query(MarketData).filter_by(symbol="AG2606").all() assert len(sqlite_records) == 2 periods = {r.period for r in sqlite_records} assert periods == {"5min", "15min"} def test_falls_back_to_sqlite_when_storage_save_fails(self, db, sample_data, storage_manager_mock): """StorageManager 写入失败时降级到 SQLite 写入。""" storage_manager_mock.save_market_data_with_timestamp.return_value = False with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.save_market_data(db, "AG2606", sample_data) storage_manager_mock.save_market_data_with_timestamp.assert_called_once() storage_manager_mock.save_market_data.assert_not_called() storage_manager_mock.save_symbol_timestamp.assert_not_called() assert isinstance(result, MarketData) assert db.query(MarketData).filter_by(symbol="AG2606").count() == 2 def test_falls_back_to_sqlite_when_storage_raises(self, db, sample_data, storage_manager_mock, caplog): """StorageManager 写入异常时降级到 SQLite 写入并记录警告。""" storage_manager_mock.save_market_data_with_timestamp.side_effect = Exception("storage down") with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache with caplog.at_level(logging.WARNING, logger="app.services.cache"): result = cache.save_market_data(db, "AG2606", sample_data) storage_manager_mock.save_market_data_with_timestamp.assert_called_once() storage_manager_mock.save_market_data.assert_not_called() storage_manager_mock.save_symbol_timestamp.assert_not_called() assert isinstance(result, MarketData) assert db.query(MarketData).filter_by(symbol="AG2606").count() == 2 assert "StorageManager 写入失败" in caplog.text class TestGetCachedData: """get_cached_data 读取与降级测试。""" def test_uses_storage_manager_and_merges_periods(self, db, storage_manager_mock): """StorageManager 返回合并后的多周期数据。""" fetched_at = datetime.now() storage_manager_mock.get_market_data.return_value = { "symbol": "AG2606", "type": "futures", "current_price": 123.45, "timestamp": fetched_at.isoformat(), "fetched_at": fetched_at.isoformat(), "timeframes": { "5min": [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}], "15min": [{"datetime": "2026-07-04T09:45:00", "open": 105, "close": 115}], }, "is_fresh": True, } with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.get_cached_data(db, "AG2606", "futures", ["5min", "15min"]) storage_manager_mock.get_market_data.assert_called_once_with( "AG2606", "futures", ["5min", "15min"], end_time=None, max_candles=100 ) assert result is not None assert result["symbol"] == "AG2606" assert result["type"] == "futures" assert result["current_price"] == 123.45 assert result["timestamp"] == fetched_at.isoformat() assert result["fetched_at"] == fetched_at.isoformat() assert "5min" in result["timeframes"] assert "15min" in result["timeframes"] assert len(result["timeframes"]["5min"]) == 1 assert result["is_fresh"] is True def test_falls_back_to_sqlite_when_storage_returns_none(self, db, storage_manager_mock): """StorageManager 未命中时降级到 SQLite 读取。""" storage_manager_mock.get_market_data.return_value = None _seed_sqlite_market_data(db) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.get_cached_data(db, "AG2606", "futures", ["5min", "15min"]) assert result is not None assert result["symbol"] == "AG2606" assert "5min" in result["timeframes"] assert "15min" in result["timeframes"] def test_falls_back_to_sqlite_when_storage_raises(self, db, storage_manager_mock, caplog): """StorageManager 读取异常时降级到 SQLite 读取并记录警告。""" storage_manager_mock.get_market_data.side_effect = Exception("storage down") _seed_sqlite_market_data(db) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache with caplog.at_level(logging.WARNING, logger="app.services.cache"): result = cache.get_cached_data(db, "AG2606", "futures", ["5min"]) assert result is not None assert "StorageManager 读取失败" in caplog.text def test_passes_end_time_and_max_candles_to_storage(self, db, storage_manager_mock): """get_cached_data 将 end_time / max_candles 透传给 StorageManager。""" end_time = datetime(2026, 7, 4, 10, 0, 0) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache cache.get_cached_data(db, "AG2606", "futures", ["5min"], end_time=end_time, max_candles=50) storage_manager_mock.get_market_data.assert_called_once_with( "AG2606", "futures", ["5min"], end_time=end_time, max_candles=50 ) def test_does_not_refilter_storage_manager_results(self, db, storage_manager_mock): """cache.py 不再对 StorageManager 返回的数据做二次过滤，直接原样返回。""" fetched_at = datetime.now() def mock_get_market_data(symbol, data_type, periods, end_time=None, max_candles=100): return { "symbol": symbol, "type": data_type, "current_price": 123.45, "timestamp": fetched_at.isoformat(), "fetched_at": fetched_at.isoformat(), "timeframes": { "5min": [ {"datetime": "2026-07-04T09:50:00", "open": 100, "close": 101}, {"datetime": "2026-07-04T09:55:00", "open": 101, "close": 102}, {"datetime": "2026-07-04T10:05:00", "open": 102, "close": 103}, ] }, "is_fresh": True, } storage_manager_mock.get_market_data.side_effect = mock_get_market_data with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.get_cached_data( db, "AG2606", "futures", ["5min"], end_time=datetime(2026, 7, 4, 10, 0, 0), max_candles=1, ) assert result is not None candles = result["timeframes"]["5min"] # StorageManager 未做过滤，cache.py 也不应再次过滤 assert len(candles) == 3 def test_does_not_mutate_storage_manager_result(self, db, storage_manager_mock): """get_cached_data 返回 StorageManager 结果时不会原地修改原对象。""" fetched_at = datetime.now() original = { "symbol": "AG2606", "type": "futures", "current_price": 123.45, "timestamp": fetched_at.isoformat(), "fetched_at": fetched_at.isoformat(), "timeframes": { "5min": [ {"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110} ] }, "is_fresh": True, } storage_manager_mock.get_market_data.return_value = original with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.get_cached_data(db, "AG2606", "futures", ["5min"]) assert result is not None assert result is not original assert result["timeframes"]["5min"] == original["timeframes"]["5min"] # 修改返回结果不应影响 StorageManager 返回的原对象 result["timeframes"]["5min"].append( {"datetime": "2026-07-04T10:00:00", "open": 111, "close": 112} ) assert len(original["timeframes"]["5min"]) == 1 def test_queries_all_periods_when_periods_is_none(self, db, storage_manager_mock): """periods=None 时优先走 StorageManager 查询全部周期。""" fetched_at = datetime.now() def mock_get_market_data(symbol, data_type, periods=None, end_time=None, max_candles=100): return { "symbol": symbol, "type": data_type, "current_price": 123.45, "timestamp": fetched_at.isoformat(), "fetched_at": fetched_at.isoformat(), "timeframes": {"5min": [], "15min": []}, "is_fresh": True, } storage_manager_mock.get_market_data.side_effect = mock_get_market_data with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.get_cached_data(db, "AG2606", "futures", periods=None) storage_manager_mock.get_market_data.assert_called_once_with( "AG2606", "futures", None, end_time=None, max_candles=100 ) assert result is not None assert "5min" in result["timeframes"] assert "15min" in result["timeframes"] class TestCacheStatus: """is_cache_valid / check_cache_status / get_latest_cached 存储层兼容测试。""" def _make_merged_result(self, symbol="AG2606", data_type="futures", periods=None, age_seconds=0): fetched_at = datetime.now() - __import__("datetime").timedelta(seconds=age_seconds) timeframes = {} for p in (periods or ["5min", "15min"]): timeframes[p] = [{"datetime": fetched_at.isoformat(), "open": 100, "close": 110}] return { "symbol": symbol, "type": data_type, "current_price": 123.45, "timestamp": fetched_at.isoformat(), "fetched_at": fetched_at.isoformat(), "timeframes": timeframes, "is_fresh": True, } def test_is_cache_valid_uses_storage_when_available(self, db, storage_manager_mock): """StorageManager 可用时优先判断缓存有效性。""" storage_manager_mock.get_market_data.return_value = self._make_merged_result(age_seconds=10) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache assert cache.is_cache_valid(db, "AG2606", "futures", "5min") is True storage_manager_mock.get_market_data.assert_called_once_with( "AG2606", "futures", ["5min"] ) def test_is_cache_valid_falls_back_to_sqlite(self, db, storage_manager_mock): """StorageManager 无数据时降级到 SQLite 判断缓存有效性。""" storage_manager_mock.get_market_data.return_value = None _seed_sqlite_market_data(db) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache assert cache.is_cache_valid(db, "AG2606", "futures", "5min") is True def test_check_cache_status_uses_storage_when_available(self, db, storage_manager_mock): """StorageManager 可用时优先检查多周期缓存状态。""" storage_manager_mock.get_market_data.return_value = self._make_merged_result() with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache status = cache.check_cache_status(db, "AG2606", "futures", ["5min", "15min"]) assert status["all_valid"] is True assert set(status["valid_periods"]) == {"5min", "15min"} assert status["missing_periods"] == [] def test_check_cache_status_falls_back_to_sqlite(self, db, storage_manager_mock): """StorageManager 无数据时降级到 SQLite 检查多周期缓存状态。""" storage_manager_mock.get_market_data.return_value = None _seed_sqlite_market_data(db) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache status = cache.check_cache_status(db, "AG2606", "futures", ["5min", "15min"]) assert status["all_valid"] is True assert set(status["valid_periods"]) == {"5min", "15min"} def test_get_latest_cached_uses_storage_when_available(self, db, storage_manager_mock): """StorageManager 可用时优先获取最新缓存记录。""" storage_manager_mock.get_market_data.return_value = self._make_merged_result() with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache records = cache.get_latest_cached(db, "AG2606", "futures") assert len(records) == 2 periods = {r.period for r in records} assert periods == {"5min", "15min"} assert all(isinstance(r, MarketData) for r in records) def test_get_latest_cached_falls_back_to_sqlite(self, db, storage_manager_mock): """StorageManager 无数据时降级到 SQLite 获取最新缓存记录。""" storage_manager_mock.get_market_data.return_value = None _seed_sqlite_market_data(db) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache records = cache.get_latest_cached(db, "AG2606", "futures") assert len(records) == 2 periods = {r.period for r in records} assert periods == {"5min", "15min"} class TestSymbolTimestamp: """合约时间戳函数封装测试。""" def test_update_symbol_timestamp_uses_storage_when_available(self, db, storage_manager_mock): """StorageManager 可用时优先写入合约时间戳。""" refresh_time = datetime(2026, 7, 4, 10, 0, 0) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache cache.update_symbol_timestamp(db, "AG2606", "futures", refresh_time) storage_manager_mock.save_symbol_timestamp.assert_called_once_with("AG2606", "futures", refresh_time) sqlite_record = db.query(SymbolTimestamp).filter_by(symbol="AG2606").first() assert sqlite_record is None def test_update_symbol_timestamp_falls_back_to_sqlite(self, db, storage_manager_mock): """StorageManager 不可用时降级到 SQLite 写入合约时间戳。""" storage_manager_mock.is_available.return_value = False refresh_time = datetime(2026, 7, 4, 10, 0, 0) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache cache.update_symbol_timestamp(db, "AG2606", "futures", refresh_time) storage_manager_mock.save_symbol_timestamp.assert_not_called() sqlite_record = db.query(SymbolTimestamp).filter_by(symbol="AG2606").first() assert sqlite_record is not None assert sqlite_record.last_refresh_at == refresh_time assert sqlite_record.refresh_count == 1 def test_get_symbol_timestamp_uses_storage_when_available(self, db, storage_manager_mock): """StorageManager 可用时优先读取合约时间戳。""" storage_manager_mock.get_symbol_timestamp.return_value = { "last_refresh_at": datetime(2026, 7, 4, 10, 0, 0).isoformat(), "refresh_count": 42, } with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.get_symbol_timestamp(db, "AG2606", "futures") storage_manager_mock.get_symbol_timestamp.assert_called_once_with("AG2606", "futures") assert result == datetime(2026, 7, 4, 10, 0, 0) def test_get_symbol_timestamp_falls_back_to_sqlite(self, db, storage_manager_mock): """StorageManager 未命中时降级到 SQLite 读取合约时间戳。""" storage_manager_mock.get_symbol_timestamp.return_value = None _seed_sqlite_symbol_timestamp(db) with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): from app.services import cache result = cache.get_symbol_timestamp(db, "AG2606", "futures") assert result == datetime(2026, 7, 4, 10, 0, 0)