diff --git a/app/main.py b/app/main.py index a34bd37..a0994c9 100644 --- a/app/main.py +++ b/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 diff --git a/tests/test_main_lifespan.py b/tests/test_main_lifespan.py new file mode 100644 index 0000000..ac54156 --- /dev/null +++ b/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