test(integration): verify fallback to MySQL and SQLite scenarios

tests/test_fallback_scenarios.py

@@ -0,0 +1,375 @@
+"""
+降级场景验证测试
+
+覆盖：
+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