|
|
"""
|
|
|
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_update_symbol_timestamp_falls_back_to_sqlite_when_storage_returns_false(self, db, storage_manager_mock):
|
|
|
"""StorageManager 可用但返回 False 时降级到 SQLite 写入合约时间戳。"""
|
|
|
storage_manager_mock.save_symbol_timestamp.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_called_once_with("AG2606", "futures", refresh_time)
|
|
|
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_update_symbol_timestamp_falls_back_to_sqlite_when_storage_raises(self, db, storage_manager_mock, caplog):
|
|
|
"""StorageManager 可用但写入异常时降级到 SQLite 写入合约时间戳并记录警告。"""
|
|
|
storage_manager_mock.save_symbol_timestamp.side_effect = Exception("mysql down")
|
|
|
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
|
|
|
|
|
|
with caplog.at_level(logging.WARNING, logger="app.services.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 not None
|
|
|
assert sqlite_record.last_refresh_at == refresh_time
|
|
|
assert sqlite_record.refresh_count == 1
|
|
|
assert "StorageManager 时间戳写入失败" in caplog.text
|
|
|
|
|
|
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)
|
