mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-02 22:17:08 +08:00
feat: 数据库迁移前自动备份 (#6360)
This commit is contained in:
InfinityPacer
@@ -90,6 +90,7 @@ _LOG_DOWNLOAD_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
|
||||
_DATABASE_BACKUP_SETTING_KEYS = {
|
||||
"DB_BACKUP_ENABLE",
|
||||
"DB_BACKUP_CRON",
|
||||
"DB_BACKUP_ON_UPGRADE",
|
||||
"DB_BACKUP_PATH",
|
||||
"DB_BACKUP_RETENTION_DAYS",
|
||||
"DB_BACKUP_MAX_COUNT",
|
||||
|
||||
+2
-13
@@ -38,7 +38,6 @@ from uvicorn import Config
|
||||
from app.adapters.system.stdio import configure_rotating_stdio
|
||||
from app.adapters.system.host import SystemUtils
|
||||
|
||||
# 禁用输出
|
||||
stdio_log_file = os.getenv("MOVIEPILOT_STDIO_LOG_FILE")
|
||||
if stdio_log_file:
|
||||
# 本地 CLI 会把 stdout/stderr 切到滚动日志，避免无限追加单独的大文件。
|
||||
@@ -56,9 +55,8 @@ elif SystemUtils.is_frozen():
|
||||
|
||||
from app.factory import app
|
||||
from app.runtime.config import global_vars, settings
|
||||
from app.startup.database_initializer import init_db, update_db
|
||||
from app.startup.database_initializer import prepare_database
|
||||
|
||||
# 设置进程名
|
||||
setproctitle.setproctitle(settings.PROJECT_NAME)
|
||||
|
||||
|
||||
@@ -70,7 +68,6 @@ class MoviePilotServer(uvicorn.Server):
|
||||
super().handle_exit(sig, frame)
|
||||
|
||||
|
||||
# uvicorn服务
|
||||
Server = MoviePilotServer(Config(app, host=settings.HOST, port=settings.PORT,
|
||||
reload=settings.DEV, workers=settings.API_WORKERS,
|
||||
timeout_graceful_shutdown=60))
|
||||
@@ -109,7 +106,6 @@ def start_tray():
|
||||
|
||||
import pystray
|
||||
|
||||
# 托盘图标
|
||||
TrayIcon = pystray.Icon(
|
||||
settings.PROJECT_NAME,
|
||||
icon=Image.open(settings.ROOT_PATH / 'app.ico'),
|
||||
@@ -124,7 +120,6 @@ def start_tray():
|
||||
)
|
||||
)
|
||||
)
|
||||
# 启动托盘图标
|
||||
threading.Thread(target=TrayIcon.run, daemon=True).start()
|
||||
|
||||
|
||||
@@ -138,17 +133,11 @@ def signal_handler(signum, frame):
|
||||
|
||||
def run_application() -> None:
|
||||
"""初始化进程并启动 API 服务"""
|
||||
# 注册信号处理器
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
|
||||
# 启动托盘
|
||||
start_tray()
|
||||
# 初始化数据库
|
||||
init_db()
|
||||
# 更新数据库
|
||||
update_db()
|
||||
# 启动API服务
|
||||
prepare_database()
|
||||
Server.run()
|
||||
|
||||
|
||||
|
||||
@@ -194,6 +194,8 @@ class ConfigModel(BaseModel):
|
||||
DB_BACKUP_ENABLE: bool = False
|
||||
# 定时备份的 Cron 表达式，留空时不注册定时任务
|
||||
DB_BACKUP_CRON: str = "0 3 * * *"
|
||||
# 检测到现有数据库需要迁移时，在结构变更前创建恢复点
|
||||
DB_BACKUP_ON_UPGRADE: bool = True
|
||||
# 备份根目录；未配置时使用 CONFIG_PATH/database_backup
|
||||
DB_BACKUP_PATH: Optional[str] = None
|
||||
# 本地备份的保留天数，0 表示不按时间清理
|
||||
|
||||
@@ -348,13 +348,11 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
|
||||
},
|
||||
}
|
||||
|
||||
# 创建定时服务
|
||||
self._scheduler = BackgroundScheduler(
|
||||
timezone=settings.TZ,
|
||||
executors={"default": ThreadPoolExecutor(settings.CONF.scheduler)},
|
||||
)
|
||||
|
||||
# 数据库备份复用宿主调度器，不创建独立定时线程。
|
||||
self._register_database_backup_job()
|
||||
|
||||
# CookieCloud定时同步
|
||||
|
||||
@@ -1,24 +1,130 @@
|
||||
from collections.abc import Callable
|
||||
from configparser import ConfigParser as _ConfigParser
|
||||
import traceback
|
||||
|
||||
from alembic.command import upgrade
|
||||
from alembic.config import Config
|
||||
from alembic.migration import MigrationContext
|
||||
from alembic.script import ScriptDirectory
|
||||
from alembic.util import CommandError
|
||||
from sqlalchemy import inspect
|
||||
from sqlalchemy.engine import Engine
|
||||
|
||||
from app.runtime.config import settings
|
||||
from app.db import Base
|
||||
from app.db.engine import get_engine
|
||||
from app.db.models import load_all_models
|
||||
from app.runtime.log import logger
|
||||
from app.startup.database import build_database_governance
|
||||
|
||||
|
||||
def _build_alembic_config(engine: Engine | None = None) -> Config:
|
||||
"""构造与应用活动数据库一致的 Alembic 配置。"""
|
||||
engine = engine or get_engine()
|
||||
alembic_cfg = Config()
|
||||
alembic_cfg.file_config = _ConfigParser(interpolation=None)
|
||||
alembic_cfg.set_main_option(
|
||||
'script_location',
|
||||
str(settings.ROOT_PATH / 'database'),
|
||||
)
|
||||
alembic_cfg.set_main_option(
|
||||
'sqlalchemy.url',
|
||||
engine.url.render_as_string(hide_password=False),
|
||||
)
|
||||
return alembic_cfg
|
||||
|
||||
|
||||
def _migration_state(
|
||||
engine: Engine,
|
||||
alembic_cfg: Config,
|
||||
) -> tuple[bool, tuple[str, ...], tuple[str, ...]]:
|
||||
"""读取数据库迁移状态，并在结构写入前校验版本链。"""
|
||||
script = ScriptDirectory.from_config(alembic_cfg)
|
||||
target_heads = tuple(script.get_heads())
|
||||
with engine.connect() as connection:
|
||||
table_names = set(inspect(connection).get_table_names())
|
||||
current_heads = tuple(
|
||||
MigrationContext.configure(connection).get_current_heads()
|
||||
)
|
||||
has_existing_database = bool(table_names - {'alembic_version'})
|
||||
_validate_migration_lineage(script, current_heads, target_heads)
|
||||
return has_existing_database, current_heads, target_heads
|
||||
|
||||
|
||||
def _validate_migration_lineage(
|
||||
script: ScriptDirectory,
|
||||
current_heads: tuple[str, ...],
|
||||
target_heads: tuple[str, ...],
|
||||
) -> None:
|
||||
"""拒绝无法沿当前迁移链安全升级的数据库版本。"""
|
||||
if len(target_heads) != 1:
|
||||
raise RuntimeError(
|
||||
f"数据库迁移脚本必须只有一个 head，当前为 {target_heads}"
|
||||
)
|
||||
if len(current_heads) > 1:
|
||||
raise RuntimeError(
|
||||
f"数据库存在多个 current revision，无法自动迁移：{current_heads}"
|
||||
)
|
||||
if not current_heads:
|
||||
return
|
||||
|
||||
current = current_heads[0]
|
||||
target = target_heads[0]
|
||||
try:
|
||||
script.get_revision(current)
|
||||
except CommandError as error:
|
||||
raise RuntimeError(
|
||||
f"当前 MoviePilot 无法识别数据库 revision：{current}"
|
||||
) from error
|
||||
if current == target:
|
||||
return
|
||||
|
||||
ancestors = {
|
||||
revision.revision
|
||||
for revision in script.walk_revisions(base='base', head=target)
|
||||
}
|
||||
if current not in ancestors:
|
||||
raise RuntimeError(
|
||||
f"数据库 revision {current} 不是当前 head {target} 的可升级祖先"
|
||||
)
|
||||
|
||||
|
||||
def prepare_database(*, before_alembic: Callable[[], None] | None = None) -> None:
|
||||
"""在建表或迁移前完成版本校验及可选备份。"""
|
||||
engine = get_engine()
|
||||
alembic_cfg = _build_alembic_config(engine)
|
||||
has_existing_database, current_heads, target_heads = _migration_state(
|
||||
engine,
|
||||
alembic_cfg,
|
||||
)
|
||||
requires_migration = (
|
||||
has_existing_database
|
||||
and set(current_heads) != set(target_heads)
|
||||
)
|
||||
if (
|
||||
requires_migration
|
||||
and settings.DB_BACKUP_ENABLE
|
||||
and settings.DB_BACKUP_ON_UPGRADE
|
||||
):
|
||||
current_version = current_heads[0] if current_heads else "未标记"
|
||||
target_version = target_heads[0]
|
||||
logger.info(
|
||||
f"数据库需要从版本 {current_version} 升级到 {target_version}，"
|
||||
"正在创建迁移前备份"
|
||||
)
|
||||
build_database_governance().create_backup()
|
||||
|
||||
init_db()
|
||||
if before_alembic:
|
||||
# 首次初始化需要先建立用户表，再把管理员密码交给 Alembic 基础迁移消费。
|
||||
before_alembic()
|
||||
update_db(alembic_cfg)
|
||||
|
||||
|
||||
def init_db():
|
||||
"""
|
||||
初始化数据库
|
||||
"""
|
||||
# 函数内导入而非模块级：写成模块级会让 import 本模块的一方也被迫拉起引擎模块。
|
||||
# 引擎一律用 get_engine() 取——旧名字 `app.db.Engine` 只为仓库外插件保留，且它一经
|
||||
# 属性访问就把引擎建出来，模块级写法会使本模块反过来依赖「数据库已在别处初始化完成」。
|
||||
from app.db.engine import get_engine
|
||||
|
||||
# 确保所有模型都已注册到 Base.metadata 中
|
||||
load_all_models()
|
||||
|
||||
@@ -26,25 +132,15 @@ def init_db():
|
||||
Base.metadata.create_all(bind=get_engine())
|
||||
|
||||
|
||||
def update_db():
|
||||
def update_db(alembic_cfg: Config | None = None):
|
||||
"""
|
||||
更新数据库
|
||||
"""
|
||||
script_location = settings.ROOT_PATH / 'database'
|
||||
try:
|
||||
alembic_cfg = Config()
|
||||
alembic_cfg.file_config = _ConfigParser(interpolation=None)
|
||||
alembic_cfg.set_main_option('script_location', str(script_location))
|
||||
|
||||
# 与引擎构建使用同一套 URL 推导：两处各自拼接会在配置变更时悄悄漂移，
|
||||
# 导致迁移连到与应用不同的库上
|
||||
db_url = settings.DB_SQLITE_URL() if settings.DB_TYPE.lower() != "postgresql" \
|
||||
else settings.DB_POSTGRESQL_URL()
|
||||
|
||||
alembic_cfg.set_main_option('sqlalchemy.url', db_url)
|
||||
alembic_cfg = alembic_cfg or _build_alembic_config()
|
||||
upgrade(alembic_cfg, 'head')
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f'数据库更新失败：{str(error)} - {traceback.format_exc()}'
|
||||
f"数据库更新失败：{error}\n{traceback.format_exc()}"
|
||||
)
|
||||
raise
|
||||
|
||||
+18
-11
@@ -2408,7 +2408,7 @@ def _apply_local_system_config_inner(config_payload: dict[str, Any]) -> None:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
try:
|
||||
from app.startup.database_initializer import init_db, update_db
|
||||
from app.startup.database_initializer import prepare_database
|
||||
from app.db.oper.systemconfig import SystemConfigOper
|
||||
from app.schemas.types import SystemConfigKey
|
||||
except ModuleNotFoundError as exc:
|
||||
@@ -2416,9 +2416,13 @@ def _apply_local_system_config_inner(config_payload: dict[str, Any]) -> None:
|
||||
"当前环境尚未安装 MoviePilot 运行依赖，请先执行 moviepilot install deps 或 moviepilot setup"
|
||||
) from exc
|
||||
|
||||
init_db()
|
||||
generated_password = _prepare_superuser_password_for_bootstrap()
|
||||
update_db()
|
||||
generated_password = None
|
||||
|
||||
def prepare_superuser_password() -> None:
|
||||
nonlocal generated_password
|
||||
generated_password = _prepare_superuser_password_for_bootstrap()
|
||||
|
||||
prepare_database(before_alembic=prepare_superuser_password)
|
||||
_ensure_superuser_account_inner()
|
||||
if generated_password:
|
||||
print_step(f"超级管理员初始密码：{generated_password}")
|
||||
@@ -2571,15 +2575,19 @@ def _sync_superuser_account_inner() -> None:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
try:
|
||||
from app.startup.database_initializer import init_db, update_db
|
||||
from app.startup.database_initializer import prepare_database
|
||||
except ModuleNotFoundError as exc:
|
||||
raise RuntimeError(
|
||||
"当前环境尚未安装 MoviePilot 运行依赖，请先执行 moviepilot install deps 或 moviepilot setup"
|
||||
) from exc
|
||||
|
||||
init_db()
|
||||
generated_password = _prepare_superuser_password_for_bootstrap()
|
||||
update_db()
|
||||
generated_password = None
|
||||
|
||||
def prepare_superuser_password() -> None:
|
||||
nonlocal generated_password
|
||||
generated_password = _prepare_superuser_password_for_bootstrap()
|
||||
|
||||
prepare_database(before_alembic=prepare_superuser_password)
|
||||
_ensure_superuser_account_inner()
|
||||
if generated_password:
|
||||
print_step(f"超级管理员初始密码：{generated_password}")
|
||||
@@ -3671,7 +3679,7 @@ def run_agent_request(
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
try:
|
||||
from app.startup.database_initializer import init_db, update_db
|
||||
from app.startup.database_initializer import prepare_database
|
||||
from app.agent import MoviePilotAgent
|
||||
from app.runtime.config import settings
|
||||
except ModuleNotFoundError as exc:
|
||||
@@ -3682,8 +3690,7 @@ def run_agent_request(
|
||||
if not settings.AI_AGENT_ENABLE:
|
||||
raise RuntimeError("MoviePilot 智能体未启用，请先在配置中打开 AI_AGENT_ENABLE")
|
||||
|
||||
init_db()
|
||||
update_db()
|
||||
prepare_database()
|
||||
|
||||
session = (session_id or "").strip()
|
||||
if new_session or not session:
|
||||
|
||||
+4
-2
@@ -13,8 +13,8 @@
|
||||
"runtime_to_db": [],
|
||||
"workflow_to_db": []
|
||||
},
|
||||
"edge_count": 6059,
|
||||
"edge_sha256": "85ab4dd2f01f48bae417f0401f272763f15ba7256b58ca0ab5a8256f4946c2f9",
|
||||
"edge_count": 6061,
|
||||
"edge_sha256": "022ba984b711776539b4dc120d0f2e92645f61b28b59dd6be3c32a64499d08ce",
|
||||
"edges": [
|
||||
"app -> app.runtime",
|
||||
"app -> app.runtime.compat",
|
||||
@@ -5705,6 +5705,8 @@
|
||||
"app.startup.database_initializer -> app.runtime",
|
||||
"app.startup.database_initializer -> app.runtime.config",
|
||||
"app.startup.database_initializer -> app.runtime.log",
|
||||
"app.startup.database_initializer -> app.startup",
|
||||
"app.startup.database_initializer -> app.startup.database",
|
||||
"app.startup.domain_initializer -> app.adapters",
|
||||
"app.startup.domain_initializer -> app.adapters.system",
|
||||
"app.startup.domain_initializer -> app.adapters.system.rust",
|
||||
|
||||
@@ -1210,19 +1210,19 @@
|
||||
},
|
||||
{
|
||||
"caller": "app.api.endpoints.system",
|
||||
"line": 859
|
||||
"line": 860
|
||||
},
|
||||
{
|
||||
"caller": "app.api.endpoints.system",
|
||||
"line": 973
|
||||
"line": 974
|
||||
},
|
||||
{
|
||||
"caller": "app.api.endpoints.system",
|
||||
"line": 1027
|
||||
"line": 1028
|
||||
},
|
||||
{
|
||||
"caller": "app.api.endpoints.system",
|
||||
"line": 1041
|
||||
"line": 1042
|
||||
},
|
||||
{
|
||||
"caller": "app.chain._messaging",
|
||||
@@ -1652,7 +1652,7 @@
|
||||
"consumers": [
|
||||
{
|
||||
"caller": "app.scheduler",
|
||||
"line": 1150
|
||||
"line": 1148
|
||||
}
|
||||
],
|
||||
"producers": []
|
||||
@@ -1813,7 +1813,7 @@
|
||||
},
|
||||
{
|
||||
"caller": "app.scheduler",
|
||||
"line": 770
|
||||
"line": 768
|
||||
}
|
||||
]
|
||||
},
|
||||
|
||||
@@ -27,6 +27,7 @@ def test_database_backup_schedule_only_watches_job_shape() -> None:
|
||||
assert Scheduler.CONFIG_WATCH.intersection({
|
||||
"DB_BACKUP_ENABLE",
|
||||
"DB_BACKUP_CRON",
|
||||
"DB_BACKUP_ON_UPGRADE",
|
||||
"DB_BACKUP_PATH",
|
||||
"DB_BACKUP_RETENTION_DAYS",
|
||||
"DB_BACKUP_MAX_COUNT",
|
||||
|
||||
@@ -1,11 +1,17 @@
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from alembic.util import CommandError
|
||||
from sqlalchemy import Column, Integer, MetaData, Table, create_engine, inspect, text
|
||||
from sqlalchemy.engine.url import make_url
|
||||
|
||||
from app.startup import database_initializer as db_init
|
||||
from app.startup import database as startup_database
|
||||
|
||||
|
||||
LOCAL_SETUP_PATH = (
|
||||
@@ -43,6 +49,334 @@ def test_update_db_preserves_migration_error_and_traceback(monkeypatch) -> None:
|
||||
assert "RuntimeError: migration failed" in logged_errors[0]
|
||||
|
||||
|
||||
def test_prepare_database_creates_backup_before_schema_changes(monkeypatch) -> None:
|
||||
"""既有数据库待迁移时，恢复点必须早于所有结构写入。"""
|
||||
calls: list[str] = []
|
||||
logged_messages: list[str] = []
|
||||
governance = Mock()
|
||||
governance.create_backup.side_effect = lambda: calls.append("backup")
|
||||
monkeypatch.setattr(db_init, "get_engine", lambda: object())
|
||||
monkeypatch.setattr(db_init, "_build_alembic_config", lambda _engine: object())
|
||||
monkeypatch.setattr(
|
||||
db_init,
|
||||
"_migration_state",
|
||||
lambda *_: (True, ("old",), ("head",)),
|
||||
)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ENABLE", True)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ON_UPGRADE", True)
|
||||
monkeypatch.setattr(
|
||||
db_init,
|
||||
"build_database_governance",
|
||||
lambda: governance,
|
||||
)
|
||||
monkeypatch.setattr(db_init.logger, "info", logged_messages.append)
|
||||
monkeypatch.setattr(db_init, "init_db", lambda: calls.append("create_all"))
|
||||
monkeypatch.setattr(
|
||||
db_init,
|
||||
"update_db",
|
||||
lambda _config: calls.append("alembic"),
|
||||
)
|
||||
|
||||
db_init.prepare_database(before_alembic=lambda: calls.append("before_alembic"))
|
||||
|
||||
assert calls == ["backup", "create_all", "before_alembic", "alembic"]
|
||||
assert logged_messages == [
|
||||
"数据库需要从版本 old 升级到 head，正在创建迁移前备份"
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("has_existing_database", "current_heads", "backup_enabled", "upgrade_enabled"),
|
||||
(
|
||||
(False, (), True, True),
|
||||
(True, ("head",), True, True),
|
||||
(True, ("old",), False, True),
|
||||
(True, ("old",), True, False),
|
||||
),
|
||||
)
|
||||
def test_prepare_database_skips_backup_outside_enabled_pending_migration(
|
||||
monkeypatch,
|
||||
has_existing_database: bool,
|
||||
current_heads: tuple[str, ...],
|
||||
backup_enabled: bool,
|
||||
upgrade_enabled: bool,
|
||||
) -> None:
|
||||
"""全新库、已到 head 或关闭保护时不创建自动恢复点。"""
|
||||
governance = Mock()
|
||||
monkeypatch.setattr(db_init, "get_engine", lambda: object())
|
||||
monkeypatch.setattr(db_init, "_build_alembic_config", lambda _engine: object())
|
||||
monkeypatch.setattr(
|
||||
db_init,
|
||||
"_migration_state",
|
||||
lambda *_: (has_existing_database, current_heads, ("head",)),
|
||||
)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ENABLE", backup_enabled)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ON_UPGRADE", upgrade_enabled)
|
||||
monkeypatch.setattr(
|
||||
db_init,
|
||||
"build_database_governance",
|
||||
lambda: governance,
|
||||
)
|
||||
monkeypatch.setattr(db_init, "init_db", lambda: None)
|
||||
monkeypatch.setattr(db_init, "update_db", lambda _config: None)
|
||||
|
||||
db_init.prepare_database()
|
||||
|
||||
governance.create_backup.assert_not_called()
|
||||
|
||||
|
||||
def test_prepare_database_stops_before_schema_changes_when_backup_fails(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
"""迁移保护失败时不得继续执行 create_all 或 Alembic。"""
|
||||
backup_error = RuntimeError("backup failed")
|
||||
init_calls: list[None] = []
|
||||
governance = Mock()
|
||||
governance.create_backup.side_effect = backup_error
|
||||
monkeypatch.setattr(db_init, "get_engine", lambda: object())
|
||||
monkeypatch.setattr(db_init, "_build_alembic_config", lambda _engine: object())
|
||||
monkeypatch.setattr(
|
||||
db_init,
|
||||
"_migration_state",
|
||||
lambda *_: (True, ("old",), ("head",)),
|
||||
)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ENABLE", True)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ON_UPGRADE", True)
|
||||
monkeypatch.setattr(
|
||||
db_init,
|
||||
"build_database_governance",
|
||||
lambda: governance,
|
||||
)
|
||||
monkeypatch.setattr(db_init, "init_db", lambda: init_calls.append(None))
|
||||
monkeypatch.setattr(db_init, "update_db", lambda _config: init_calls.append(None))
|
||||
|
||||
with pytest.raises(RuntimeError) as raised:
|
||||
db_init.prepare_database(
|
||||
before_alembic=lambda: init_calls.append(None),
|
||||
)
|
||||
|
||||
assert raised.value is backup_error
|
||||
assert init_calls == []
|
||||
|
||||
|
||||
def test_alembic_config_uses_active_engine_url_without_hiding_password() -> None:
|
||||
"""Alembic 必须连接活动引擎目标，内部配置不得把密码替换为星号。"""
|
||||
engine = SimpleNamespace(
|
||||
url=make_url("postgresql://moviepilot:secret@database/moviepilot")
|
||||
)
|
||||
|
||||
config = db_init._build_alembic_config(engine)
|
||||
|
||||
assert config.get_main_option("sqlalchemy.url") == (
|
||||
"postgresql://moviepilot:secret@database/moviepilot"
|
||||
)
|
||||
|
||||
|
||||
def test_migration_state_distinguishes_fresh_legacy_and_current_sqlite(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""SQLite 空库不备份，已有业务表且无 revision 时识别为待迁移。"""
|
||||
engine = create_engine(f"sqlite:///{tmp_path / 'migration.db'}")
|
||||
config = db_init._build_alembic_config(engine)
|
||||
target_heads = tuple(db_init.ScriptDirectory.from_config(config).get_heads())
|
||||
|
||||
assert db_init._migration_state(engine, config) == (False, (), target_heads)
|
||||
|
||||
with engine.begin() as connection:
|
||||
connection.execute(text("CREATE TABLE legacy_data (id INTEGER PRIMARY KEY)"))
|
||||
|
||||
assert db_init._migration_state(engine, config) == (True, (), target_heads)
|
||||
|
||||
with engine.begin() as connection:
|
||||
connection.execute(
|
||||
text("CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)")
|
||||
)
|
||||
connection.execute(
|
||||
text("INSERT INTO alembic_version (version_num) VALUES (:head)"),
|
||||
{"head": target_heads[0]},
|
||||
)
|
||||
|
||||
assert db_init._migration_state(engine, config) == (
|
||||
True,
|
||||
target_heads,
|
||||
target_heads,
|
||||
)
|
||||
|
||||
|
||||
def test_migration_state_rejects_unknown_revision_before_schema_writes(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""未知或更高版本 revision 不得被误判为可执行升级。"""
|
||||
engine = create_engine(f"sqlite:///{tmp_path / 'future.db'}")
|
||||
config = db_init._build_alembic_config(engine)
|
||||
with engine.begin() as connection:
|
||||
connection.execute(text("CREATE TABLE legacy_data (id INTEGER PRIMARY KEY)"))
|
||||
connection.execute(
|
||||
text("CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)")
|
||||
)
|
||||
connection.execute(
|
||||
text("INSERT INTO alembic_version (version_num) VALUES ('future')")
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError, match="无法识别数据库 revision：future"):
|
||||
db_init._migration_state(engine, config)
|
||||
|
||||
assert set(db_init.inspect(engine).get_table_names()) == {
|
||||
"alembic_version",
|
||||
"legacy_data",
|
||||
}
|
||||
|
||||
|
||||
def test_migration_lineage_rejects_multiple_heads() -> None:
|
||||
"""当前迁移执行器仅接受仓库和数据库均保持单一 head。"""
|
||||
script = Mock()
|
||||
|
||||
with pytest.raises(RuntimeError, match="迁移脚本必须只有一个 head"):
|
||||
db_init._validate_migration_lineage(script, (), ("head-a", "head-b"))
|
||||
|
||||
with pytest.raises(RuntimeError, match="多个 current revision"):
|
||||
db_init._validate_migration_lineage(
|
||||
script,
|
||||
("current-a", "current-b"),
|
||||
("head",),
|
||||
)
|
||||
|
||||
|
||||
def test_migration_lineage_rejects_known_divergent_revision() -> None:
|
||||
"""可识别但不在目标祖先链上的 revision 不得继续自动迁移。"""
|
||||
script = Mock()
|
||||
script.walk_revisions.return_value = (
|
||||
SimpleNamespace(revision="head"),
|
||||
SimpleNamespace(revision="base"),
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError, match="不是当前 head head 的可升级祖先"):
|
||||
db_init._validate_migration_lineage(
|
||||
script,
|
||||
("other-branch",),
|
||||
("head",),
|
||||
)
|
||||
script.get_revision.assert_called_once_with("other-branch")
|
||||
|
||||
|
||||
def test_migration_lineage_wraps_unknown_revision() -> None:
|
||||
"""Alembic 未知 revision 错误应转换为可操作的启动错误。"""
|
||||
script = Mock()
|
||||
script.get_revision.side_effect = CommandError("unknown")
|
||||
|
||||
with pytest.raises(RuntimeError, match="无法识别数据库 revision：future"):
|
||||
db_init._validate_migration_lineage(
|
||||
script,
|
||||
("future",),
|
||||
("head",),
|
||||
)
|
||||
|
||||
|
||||
def test_prepare_database_creates_real_sqlite_restore_point_before_upgrade(
|
||||
tmp_path: Path,
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
"""迁移前备份保留旧版本，活动 SQLite 升级到目标版本。"""
|
||||
script_root = tmp_path / "database"
|
||||
versions = script_root / "versions"
|
||||
versions.mkdir(parents=True)
|
||||
(script_root / "env.py").write_text(
|
||||
"""
|
||||
from alembic import context
|
||||
from sqlalchemy import engine_from_config, pool
|
||||
|
||||
engine = engine_from_config(
|
||||
context.config.get_section(context.config.config_ini_section),
|
||||
prefix="sqlalchemy.",
|
||||
poolclass=pool.NullPool,
|
||||
)
|
||||
with engine.connect() as connection:
|
||||
context.configure(connection=connection)
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
""".strip(),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(versions / "001_base.py").write_text(
|
||||
"""
|
||||
revision = "001"
|
||||
down_revision = None
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
def upgrade():
|
||||
pass
|
||||
|
||||
def downgrade():
|
||||
pass
|
||||
""".strip(),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(versions / "002_head.py").write_text(
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision = "002"
|
||||
down_revision = "001"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
def upgrade():
|
||||
op.add_column("records", sa.Column("migrated", sa.Integer()))
|
||||
|
||||
def downgrade():
|
||||
op.drop_column("records", "migrated")
|
||||
""".strip(),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
engine = create_engine(f"sqlite:///{tmp_path / 'active.db'}")
|
||||
metadata = MetaData()
|
||||
Table("records", metadata, Column("id", Integer, primary_key=True))
|
||||
metadata.create_all(engine)
|
||||
with engine.begin() as connection:
|
||||
connection.execute(
|
||||
text("CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)")
|
||||
)
|
||||
connection.execute(
|
||||
text("INSERT INTO alembic_version (version_num) VALUES ('001')")
|
||||
)
|
||||
|
||||
config = db_init._build_alembic_config(engine)
|
||||
config.set_main_option("script_location", str(script_root))
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ENABLE", True)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_ON_UPGRADE", True)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_PATH", str(tmp_path / "backups"))
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_RETENTION_DAYS", 0)
|
||||
monkeypatch.setattr(db_init.settings, "DB_BACKUP_MAX_COUNT", 0)
|
||||
monkeypatch.setattr(db_init, "get_engine", lambda: engine)
|
||||
monkeypatch.setattr(db_init, "_build_alembic_config", lambda _engine: config)
|
||||
monkeypatch.setattr(db_init, "Base", SimpleNamespace(metadata=metadata))
|
||||
monkeypatch.setattr(db_init, "load_all_models", lambda: None)
|
||||
monkeypatch.setattr(startup_database, "get_engine", lambda: engine)
|
||||
|
||||
db_init.prepare_database()
|
||||
|
||||
artifacts = sorted((tmp_path / "backups").glob("sqlite_*.db"))
|
||||
assert len(artifacts) == 1
|
||||
with create_engine(f"sqlite:///{artifacts[0]}").connect() as connection:
|
||||
backup_revision = connection.execute(
|
||||
text("SELECT version_num FROM alembic_version")
|
||||
).scalar_one()
|
||||
with engine.connect() as connection:
|
||||
active_revision = connection.execute(
|
||||
text("SELECT version_num FROM alembic_version")
|
||||
).scalar_one()
|
||||
active_columns = {
|
||||
column["name"] for column in inspect(connection).get_columns("records")
|
||||
}
|
||||
|
||||
assert backup_revision == "001"
|
||||
assert active_revision == "002"
|
||||
assert active_columns == {"id", "migrated"}
|
||||
|
||||
|
||||
def test_local_setup_returns_failure_when_database_migration_fails(
|
||||
monkeypatch,
|
||||
capsys,
|
||||
|
||||
@@ -405,8 +405,11 @@ def test_application_preserves_stop_requested_before_startup(monkeypatch):
|
||||
lambda *_args: calls.append("signal"),
|
||||
)
|
||||
monkeypatch.setattr(main, "start_tray", lambda: calls.append("tray"))
|
||||
monkeypatch.setattr(main, "init_db", lambda: calls.append("init_db"))
|
||||
monkeypatch.setattr(main, "update_db", lambda: calls.append("update_db"))
|
||||
monkeypatch.setattr(
|
||||
main,
|
||||
"prepare_database",
|
||||
lambda: calls.append("prepare_database"),
|
||||
)
|
||||
monkeypatch.setattr(main.Server, "run", lambda: calls.append("server"))
|
||||
|
||||
main.run_application()
|
||||
@@ -416,8 +419,7 @@ def test_application_preserves_stop_requested_before_startup(monkeypatch):
|
||||
"signal",
|
||||
"signal",
|
||||
"tray",
|
||||
"init_db",
|
||||
"update_db",
|
||||
"prepare_database",
|
||||
"server",
|
||||
]
|
||||
|
||||
@@ -437,8 +439,11 @@ def test_application_does_not_start_server_after_migration_failure(monkeypatch):
|
||||
server_run = MagicMock()
|
||||
monkeypatch.setattr(main.signal, "signal", MagicMock())
|
||||
monkeypatch.setattr(main, "start_tray", MagicMock())
|
||||
monkeypatch.setattr(main, "init_db", MagicMock())
|
||||
monkeypatch.setattr(main, "update_db", MagicMock(side_effect=migration_error))
|
||||
monkeypatch.setattr(
|
||||
main,
|
||||
"prepare_database",
|
||||
MagicMock(side_effect=migration_error),
|
||||
)
|
||||
monkeypatch.setattr(main.Server, "run", server_run)
|
||||
|
||||
with pytest.raises(RuntimeError) as raised:
|
||||
|
||||
@@ -33,6 +33,8 @@ def test_database_backup_policy_rejects_invalid_values(
|
||||
[
|
||||
{"DB_BACKUP_CRON": ""},
|
||||
{"DB_BACKUP_CRON": "0 3 * * *"},
|
||||
{"DB_BACKUP_ON_UPGRADE": True},
|
||||
{"DB_BACKUP_ON_UPGRADE": False},
|
||||
{"DB_BACKUP_PATH": None},
|
||||
{"DB_BACKUP_PATH": ""},
|
||||
{"DB_BACKUP_PATH": "database_backup"},
|
||||
|
||||
Reference in New Issue
Block a user