feat(app): initialize Redis and MySQL connections in lifespan

app/main.py

@@ -35,7 +35,28 @@ async def lifespan(app: FastAPI):
     from app.analysis_db import init_analysis_db
     init_analysis_db()
     logger.info("期货智析数据库初始化完成")
-    
+
+    # 初始化 Redis 和 MySQL
+    from app.redis_client import init_redis
+    from app.mysql_database import init_mysql
+
+    try:
+        redis_ok = init_redis() is not None
+    except Exception:
+        redis_ok = False
+
+    try:
+        mysql_ok = init_mysql() is not None
+    except Exception:
+        mysql_ok = False
+
+    if redis_ok and mysql_ok:
+        logger.info("存储模式: Redis + MySQL")
+    elif mysql_ok:
+        logger.warning("存储模式: MySQL (Redis 不可用)")
+    else:
+        logger.error("存储模式: SQLite (Redis 和 MySQL 均不可用)")
+
     # 创建默认管理员账户
     from app.database import SessionLocal
     from app import auth_service

tests/test_main_lifespan.py

@@ -0,0 +1,117 @@
+import logging
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def patch_lifespan_deps():
+    """Patch all dependencies around lifespan so only storage init is exercised."""
+    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()
+
+
+@pytest.mark.asyncio
+async def test_lifespan_calls_redis_and_mysql_init(patch_lifespan_deps, caplog):
+    """ lifespan 应调用 Redis 与 MySQL 初始化函数。 """
+    from app.main import lifespan, app
+
+    patch_lifespan_deps["init_redis"].return_value = MagicMock()
+    patch_lifespan_deps["init_mysql"].return_value = MagicMock()
+
+    with caplog.at_level(logging.INFO, logger="app.main"):
+        async with lifespan(app):
+            pass
+
+    patch_lifespan_deps["init_redis"].assert_called_once()
+    patch_lifespan_deps["init_mysql"].assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_lifespan_storage_mode_redis_and_mysql(patch_lifespan_deps, caplog):
+    """ Redis 与 MySQL 均可用时，日志显示 Redis + MySQL 模式。 """
+    from app.main import lifespan, app
+
+    patch_lifespan_deps["init_redis"].return_value = MagicMock()
+    patch_lifespan_deps["init_mysql"].return_value = MagicMock()
+
+    with caplog.at_level(logging.INFO, logger="app.main"):
+        async with lifespan(app):
+            pass
+
+    assert "存储模式: Redis + MySQL" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_lifespan_storage_mode_mysql_only(patch_lifespan_deps, caplog):
+    """ 仅 MySQL 可用时，日志显示 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_lifespan_storage_mode_sqlite_fallback(patch_lifespan_deps, caplog):
+    """ Redis 与 MySQL 均不可用时，日志显示 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
+
+
+@pytest.mark.asyncio
+async def test_lifespan_storage_init_failure_does_not_raise(patch_lifespan_deps, caplog):
+    """ 初始化 Redis/MySQL 失败抛出异常时，应用仍可正常启动。 """
+    from app.main import lifespan, app
+
+    patch_lifespan_deps["init_redis"].side_effect = Exception("redis down")
+    patch_lifespan_deps["init_mysql"].side_effect = Exception("mysql down")
+
+    with caplog.at_level(logging.ERROR, logger="app.main"):
+        async with lifespan(app):
+            pass
+
+    assert "存储模式: SQLite (Redis 和 MySQL 均不可用)" in caplog.text