From 7cea089d48ccc0797da59dd2287b94b9240d87f1 Mon Sep 17 00:00:00 2001 From: Lxy Date: Sat, 4 Jul 2026 11:38:18 +0800 Subject: [PATCH] test(integration): verify fallback to MySQL and SQLite scenarios --- tests/test_fallback_scenarios.py | 375 +++++++++++++++++++++++++++++++ 1 file changed, 375 insertions(+) create mode 100644 tests/test_fallback_scenarios.py diff --git a/tests/test_fallback_scenarios.py b/tests/test_fallback_scenarios.py new file mode 100644 index 0000000..d6fb230 --- /dev/null +++ b/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