parent
5db0546e5f
commit
7ee6794095
@ -0,0 +1,286 @@
|
||||
"""
|
||||
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.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(2026, 7, 4, 10, 0, 0)
|
||||
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.assert_called_once_with("AG2606", sample_data)
|
||||
storage_manager_mock.save_symbol_timestamp.assert_called_once()
|
||||
assert result is None
|
||||
|
||||
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.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_called_once()
|
||||
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.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)
|
||||
|
||||
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()
|
||||
|
||||
def mock_get_market_data(symbol, data_type, period):
|
||||
return {
|
||||
"current_price": 123.45,
|
||||
"timestamp": fetched_at.isoformat(),
|
||||
"candles": [{"datetime": f"2026-07-04T09:{50 if period == '5min' else 45}:00", "open": 100, "close": 110}],
|
||||
}
|
||||
|
||||
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", "15min"])
|
||||
|
||||
assert storage_manager_mock.get_market_data.call_count == 2
|
||||
storage_manager_mock.get_market_data.assert_any_call("AG2606", "futures", "5min")
|
||||
storage_manager_mock.get_market_data.assert_any_call("AG2606", "futures", "15min")
|
||||
|
||||
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
|
||||
|
||||
|
||||
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)
|
||||
Loading…
Reference in new issue