|
|
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_triggers_migration_when_mysql_ok(patch_lifespan_deps):
|
|
|
""" MySQL 可用时，lifespan 应触发 SQLite 到 MySQL 的迁移。 """
|
|
|
from app.main import lifespan, app
|
|
|
|
|
|
patch_lifespan_deps["init_redis"].return_value = None
|
|
|
patch_lifespan_deps["init_mysql"].return_value = MagicMock()
|
|
|
|
|
|
with patch("app.migration.migrate_sqlite_to_mysql") as mock_migrate:
|
|
|
async with lifespan(app):
|
|
|
pass
|
|
|
|
|
|
mock_migrate.assert_called_once()
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
async def test_lifespan_skips_migration_when_mysql_unavailable(patch_lifespan_deps):
|
|
|
""" MySQL 不可用时，lifespan 不应调用迁移函数。 """
|
|
|
from app.main import lifespan, app
|
|
|
|
|
|
patch_lifespan_deps["init_redis"].return_value = None
|
|
|
patch_lifespan_deps["init_mysql"].return_value = None
|
|
|
|
|
|
with patch("app.migration.migrate_sqlite_to_mysql") as mock_migrate:
|
|
|
async with lifespan(app):
|
|
|
pass
|
|
|
|
|
|
mock_migrate.assert_not_called()
|
|
|
|
|
|
|
|
|
@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
|
