|
|
"""
|
|
|
降级场景验证测试
|
|
|
|
|
|
覆盖：
|
|
|
1. Redis 不可用但 MySQL 可用时，自动降级到 MySQL
|
|
|
2. Redis 和 MySQL 均不可用时，自动降级到 SQLite
|
|
|
3. 各降级场景下接口正常返回，无 500 错误
|
|
|
4. 恢复 Redis 服务后，等待惰性检测间隔验证自动恢复
|
|
|
|
|
|
实现说明：
|
|
|
- 使用 FastAPI TestClient 验证接口层行为。
|
|
|
- 使用内存 SQLite 引擎作为 MySQL 替身（StorageManager 基于 SQLAlchemy 通用接口）。
|
|
|
- 通过 patch `app.services.cache.get_storage_manager` 注入已配置好的 StorageManager，
|
|
|
从而绕过当前 `app/main.py` 启动时尚未将初始化后的 Redis/MySQL 客户端注入 StorageManager
|
|
|
全局单例的问题，专注于验证 cache / API 层的降级与恢复逻辑。
|
|
|
"""
|
|
|
import json
|
|
|
import logging
|
|
|
import time
|
|
|
from datetime import datetime
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
import pytest
|
|
|
from fastapi.testclient import TestClient
|
|
|
from sqlalchemy import create_engine, text
|
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
|
|
from app.models import Base, MarketData
|
|
|
from app.storage_manager import StorageManager
|
|
|
|
|
|
# test_main_lifespan.py 中的 fixture 会 patch app.main.Base.metadata.create_all 且未正确恢复，
|
|
|
# 导致后续依赖 create_all 的测试失败。这里保留原始方法引用，必要时恢复。
|
|
|
_original_create_all = Base.metadata.create_all
|
|
|
|
|
|
|
|
|
# ===== Fixtures =====
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
def sqlite_engine():
|
|
|
"""应用主数据库使用的内存 SQLite 引擎。"""
|
|
|
# 防御：修复上游 fixture 对 Base.metadata.create_all 的 patch 泄漏
|
|
|
if isinstance(Base.metadata.create_all, MagicMock):
|
|
|
Base.metadata.create_all = _original_create_all
|
|
|
|
|
|
engine = create_engine(
|
|
|
"sqlite:///:memory:",
|
|
|
connect_args={"check_same_thread": False},
|
|
|
poolclass=StaticPool,
|
|
|
)
|
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
# 确认表已创建
|
|
|
with engine.connect() as conn:
|
|
|
result = conn.execute(
|
|
|
text("SELECT name FROM sqlite_master WHERE type='table' AND name='market_data'")
|
|
|
)
|
|
|
assert result.fetchone() is not None
|
|
|
|
|
|
yield engine
|
|
|
engine.dispose()
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
def mysql_engine():
|
|
|
"""作为 MySQL 替身的内存 SQLite 引擎（StorageManager 使用 SQLAlchemy 通用接口）。"""
|
|
|
# 防御：修复上游 fixture 对 Base.metadata.create_all 的 patch 泄漏
|
|
|
if isinstance(Base.metadata.create_all, MagicMock):
|
|
|
Base.metadata.create_all = _original_create_all
|
|
|
|
|
|
engine = create_engine(
|
|
|
"sqlite:///:memory:",
|
|
|
connect_args={"check_same_thread": False},
|
|
|
poolclass=StaticPool,
|
|
|
)
|
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
# 确认表已创建
|
|
|
with engine.connect() as conn:
|
|
|
result = conn.execute(
|
|
|
text("SELECT name FROM sqlite_master WHERE type='table' AND name='market_data'")
|
|
|
)
|
|
|
assert result.fetchone() is not None
|
|
|
|
|
|
yield engine
|
|
|
engine.dispose()
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
def app_with_db(sqlite_engine):
|
|
|
"""返回替换 get_db 依赖为内存 SQLite 的 FastAPI app。"""
|
|
|
from app.main import app
|
|
|
from app.database import get_db as original_get_db
|
|
|
|
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=sqlite_engine)
|
|
|
|
|
|
def override_get_db():
|
|
|
db = SessionLocal()
|
|
|
try:
|
|
|
yield db
|
|
|
finally:
|
|
|
db.close()
|
|
|
|
|
|
app.dependency_overrides[original_get_db] = override_get_db
|
|
|
yield app
|
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
def patch_lifespan_deps():
|
|
|
"""Patch lifespan 依赖，避免真实连接数据库/Redis/MySQL 和启动调度器。"""
|
|
|
mock_db = MagicMock()
|
|
|
mock_session_local = MagicMock(return_value=mock_db)
|
|
|
|
|
|
patches = [
|
|
|
patch("app.main.Base.metadata.create_all"),
|
|
|
patch("app.main.UserBase.metadata.create_all"),
|
|
|
patch("app.analysis_db.init_analysis_db"),
|
|
|
patch("app.database.SessionLocal", mock_session_local),
|
|
|
patch("app.auth_service.create_default_admin"),
|
|
|
patch("app.main.start_scheduler"),
|
|
|
patch("app.services.cache.list_tasks", return_value=[]),
|
|
|
patch("app.services.scheduler.add_job"),
|
|
|
patch("app.main.stop_scheduler"),
|
|
|
patch("app.redis_client.init_redis"),
|
|
|
patch("app.mysql_database.init_mysql"),
|
|
|
]
|
|
|
started = [p.start() for p in patches]
|
|
|
yield {
|
|
|
"Base_create_all": started[0],
|
|
|
"UserBase_create_all": started[1],
|
|
|
"init_analysis_db": started[2],
|
|
|
"SessionLocal": started[3],
|
|
|
"create_default_admin": started[4],
|
|
|
"start_scheduler": started[5],
|
|
|
"list_tasks": started[6],
|
|
|
"add_job": started[7],
|
|
|
"stop_scheduler": started[8],
|
|
|
"init_redis": started[9],
|
|
|
"init_mysql": started[10],
|
|
|
}
|
|
|
for p in patches:
|
|
|
p.stop()
|
|
|
|
|
|
|
|
|
class FakeRedis:
|
|
|
"""可切换可用状态的轻量级 Redis 替身。"""
|
|
|
|
|
|
def __init__(self, available=True):
|
|
|
self._store = {}
|
|
|
self._available = available
|
|
|
|
|
|
def set_available(self, available):
|
|
|
self._available = available
|
|
|
|
|
|
def ping(self):
|
|
|
if not self._available:
|
|
|
raise Exception("redis down")
|
|
|
return True
|
|
|
|
|
|
def get(self, key):
|
|
|
return self._store.get(key)
|
|
|
|
|
|
def setex(self, key, ttl, value):
|
|
|
self._store[key] = value
|
|
|
|
|
|
def delete(self, *keys):
|
|
|
for key in keys:
|
|
|
self._store.pop(key, None)
|
|
|
|
|
|
def keys(self, pattern):
|
|
|
prefix = pattern.rstrip("*")
|
|
|
return [k for k in self._store if k.startswith(prefix)]
|
|
|
|
|
|
|
|
|
# ===== Helpers =====
|
|
|
|
|
|
|
|
|
def _seed_market_data(
|
|
|
engine,
|
|
|
symbol="AG2606",
|
|
|
data_type="futures",
|
|
|
period="5min",
|
|
|
current_price=123.45,
|
|
|
):
|
|
|
"""在指定引擎中写入一条行情数据。"""
|
|
|
Session = sessionmaker(bind=engine)
|
|
|
with Session() as db:
|
|
|
db.add(
|
|
|
MarketData(
|
|
|
symbol=symbol,
|
|
|
data_type=data_type,
|
|
|
period=period,
|
|
|
candles_json=json.dumps(
|
|
|
[
|
|
|
{
|
|
|
"datetime": "2026-07-04T09:55:00",
|
|
|
"open": 100,
|
|
|
"high": 115,
|
|
|
"low": 95,
|
|
|
"close": 110,
|
|
|
"volume": 1000,
|
|
|
}
|
|
|
]
|
|
|
),
|
|
|
current_price=current_price,
|
|
|
fetched_at=datetime(2026, 7, 4, 10, 0, 0),
|
|
|
candle_count=1,
|
|
|
)
|
|
|
)
|
|
|
db.commit()
|
|
|
|
|
|
|
|
|
# ===== Lifespan 日志测试 =====
|
|
|
|
|
|
|
|
|
class TestLifespanFallbackLogs:
|
|
|
"""启动时降级检测日志测试。"""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
async def test_logs_mysql_mode_when_redis_unavailable(
|
|
|
self, patch_lifespan_deps, caplog
|
|
|
):
|
|
|
"""Redis 不可用但 MySQL 可用时，lifespan 应输出 MySQL 降级模式日志。"""
|
|
|
from app.main import lifespan, app
|
|
|
|
|
|
patch_lifespan_deps["init_redis"].return_value = None
|
|
|
patch_lifespan_deps["init_mysql"].return_value = MagicMock()
|
|
|
|
|
|
with caplog.at_level(logging.WARNING, logger="app.main"):
|
|
|
async with lifespan(app):
|
|
|
pass
|
|
|
|
|
|
assert "存储模式: MySQL (Redis 不可用)" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
async def test_logs_sqlite_mode_when_both_unavailable(
|
|
|
self, patch_lifespan_deps, caplog
|
|
|
):
|
|
|
"""Redis 和 MySQL 均不可用时，lifespan 应输出 SQLite 降级模式日志。"""
|
|
|
from app.main import lifespan, app
|
|
|
|
|
|
patch_lifespan_deps["init_redis"].return_value = None
|
|
|
patch_lifespan_deps["init_mysql"].return_value = None
|
|
|
|
|
|
with caplog.at_level(logging.ERROR, logger="app.main"):
|
|
|
async with lifespan(app):
|
|
|
pass
|
|
|
|
|
|
assert "存储模式: SQLite (Redis 和 MySQL 均不可用)" in caplog.text
|
|
|
|
|
|
|
|
|
# ===== StorageManager 运行时降级与恢复测试 =====
|
|
|
|
|
|
|
|
|
class TestStorageManagerFallback:
|
|
|
"""StorageManager 运行时降级与恢复测试。"""
|
|
|
|
|
|
def test_falls_back_to_mysql_when_redis_unavailable(self, mysql_engine):
|
|
|
"""Redis 不可用但 MySQL 可用时，从 MySQL 读取数据。"""
|
|
|
_seed_market_data(mysql_engine, current_price=123.45)
|
|
|
|
|
|
fake_redis = FakeRedis(available=False)
|
|
|
manager = StorageManager(redis_client=fake_redis, mysql_engine=mysql_engine)
|
|
|
|
|
|
result = manager.get_market_data("AG2606", "futures", "5min")
|
|
|
|
|
|
assert result is not None
|
|
|
assert result["current_price"] == 123.45
|
|
|
|
|
|
def test_returns_none_when_both_unavailable(self):
|
|
|
"""Redis 和 MySQL 均不可用时，StorageManager 返回 None（由调用方降级到 SQLite）。"""
|
|
|
fake_redis = FakeRedis(available=False)
|
|
|
manager = StorageManager(redis_client=fake_redis, mysql_engine=None)
|
|
|
|
|
|
result = manager.get_market_data("AG2606", "futures", "5min")
|
|
|
|
|
|
assert result is None
|
|
|
|
|
|
def test_recovers_to_redis_after_interval(self, mysql_engine):
|
|
|
"""Redis 恢复后，超过惰性检测间隔应重新使用 Redis 并回填缓存。"""
|
|
|
_seed_market_data(mysql_engine, current_price=123.45)
|
|
|
|
|
|
fake_redis = FakeRedis(available=False)
|
|
|
manager = StorageManager(
|
|
|
redis_client=fake_redis, mysql_engine=mysql_engine, check_interval=0.1
|
|
|
)
|
|
|
|
|
|
# Redis 不可用，首次读取应来自 MySQL
|
|
|
result = manager.get_market_data("AG2606", "futures", "5min")
|
|
|
assert result["current_price"] == 123.45
|
|
|
# Redis 不可用，不应有回填
|
|
|
assert fake_redis.get("market_data:AG2606:5min") is None
|
|
|
|
|
|
# 恢复 Redis
|
|
|
fake_redis.set_available(True)
|
|
|
# 等待检测间隔过期
|
|
|
time.sleep(0.15)
|
|
|
|
|
|
# 再次读取：Redis ping 通过，get 返回 None，回源 MySQL 后回填 Redis
|
|
|
result = manager.get_market_data("AG2606", "futures", "5min")
|
|
|
assert result["current_price"] == 123.45
|
|
|
assert fake_redis.get("market_data:AG2606:5min") is not None
|
|
|
|
|
|
|
|
|
# ===== API 接口降级测试 =====
|
|
|
|
|
|
|
|
|
class TestApiFallbackResponses:
|
|
|
"""API 接口在不同存储模式下的响应测试。"""
|
|
|
|
|
|
def test_api_returns_mysql_data_when_redis_unavailable(
|
|
|
self, patch_lifespan_deps, app_with_db, sqlite_engine, mysql_engine
|
|
|
):
|
|
|
"""Redis 不可用但 MySQL 可用时，接口从 MySQL 返回数据且无 500 错误。"""
|
|
|
_seed_market_data(mysql_engine, current_price=123.45)
|
|
|
_seed_market_data(sqlite_engine, current_price=456.78)
|
|
|
|
|
|
fake_redis = FakeRedis(available=False)
|
|
|
storage_manager = StorageManager(
|
|
|
redis_client=fake_redis, mysql_engine=mysql_engine, check_interval=0.1
|
|
|
)
|
|
|
|
|
|
patch_lifespan_deps["init_redis"].return_value = None
|
|
|
patch_lifespan_deps["init_mysql"].return_value = mysql_engine
|
|
|
|
|
|
with patch("app.mysql_database.mysql_engine", mysql_engine):
|
|
|
with patch("app.services.cache.get_storage_manager", return_value=storage_manager):
|
|
|
with TestClient(app_with_db) as client:
|
|
|
response = client.get("/api/v1/data/latest/AG2606?period=5min")
|
|
|
|
|
|
assert response.status_code == 200
|
|
|
data = response.json()
|
|
|
assert data["current_price"] == 123.45
|
|
|
|
|
|
def test_api_returns_sqlite_data_when_both_unavailable(
|
|
|
self, patch_lifespan_deps, app_with_db, sqlite_engine
|
|
|
):
|
|
|
"""Redis 和 MySQL 均不可用时，接口降级到 SQLite 返回数据且无 500 错误。"""
|
|
|
_seed_market_data(sqlite_engine, current_price=456.78)
|
|
|
|
|
|
fake_redis = FakeRedis(available=False)
|
|
|
storage_manager = StorageManager(
|
|
|
redis_client=fake_redis, mysql_engine=None, check_interval=0.1
|
|
|
)
|
|
|
|
|
|
patch_lifespan_deps["init_redis"].return_value = None
|
|
|
patch_lifespan_deps["init_mysql"].return_value = None
|
|
|
|
|
|
with patch("app.services.cache.get_storage_manager", return_value=storage_manager):
|
|
|
with TestClient(app_with_db) as client:
|
|
|
response = client.get("/api/v1/data/latest/AG2606?period=5min")
|
|
|
|
|
|
assert response.status_code == 200
|
|
|
data = response.json()
|
|
|
assert data["current_price"] == 456.78
|
|
|
|
|
|
def test_api_returns_404_not_500_when_no_data_anywhere(
|
|
|
self, patch_lifespan_deps, app_with_db
|
|
|
):
|
|
|
"""任何存储层均无数据时，接口应返回 404 而非 500。"""
|
|
|
fake_redis = FakeRedis(available=False)
|
|
|
storage_manager = StorageManager(
|
|
|
redis_client=fake_redis, mysql_engine=None, check_interval=0.1
|
|
|
)
|
|
|
|
|
|
patch_lifespan_deps["init_redis"].return_value = None
|
|
|
patch_lifespan_deps["init_mysql"].return_value = None
|
|
|
|
|
|
with patch("app.services.cache.get_storage_manager", return_value=storage_manager):
|
|
|
with TestClient(app_with_db) as client:
|
|
|
response = client.get("/api/v1/data/latest/UNKNOWN?period=5min")
|
|
|
|
|
|
assert response.status_code == 404
|
