diff --git a/app/startup/database_initializer.py b/app/startup/database_initializer.py index 0249578c7..5146c925d 100644 --- a/app/startup/database_initializer.py +++ b/app/startup/database_initializer.py @@ -14,20 +14,8 @@ from app.runtime.config import settings from app.db.base import Base from app.db.engine import get_engine from app.db.models import load_all_models -from app.db.session import SessionFactory, async_session_scope -from app.db.uow import configure_transaction_runners from app.runtime.log import logger from app.startup.database import build_database_governance -from app.startup.transaction import TransactionalWriteRunner - - -def _configure_migration_transaction_runner() -> None: - """在 Alembic 调用旧无会话 Oper 前装配可独立提交的兼容事务。""" - runner = TransactionalWriteRunner( - sync_session=SessionFactory, - async_session=async_session_scope, - ) - configure_transaction_runners(sync=runner.sync, async_=runner.async_) def _build_alembic_config(engine: Engine | None = None) -> Config: @@ -161,8 +149,6 @@ def update_db(alembic_cfg: Config | None = None): 更新数据库 """ try: - # 早期迁移脚本会调用 SystemConfigOper()，此时 modules_initializer 尚未执行。 - _configure_migration_transaction_runner() alembic_cfg = alembic_cfg or _build_alembic_config() upgrade(alembic_cfg, 'head') except Exception as error: diff --git a/database/versions/262735d025da_2_0_1.py b/database/versions/262735d025da_2_0_1.py index 91c35f67a..0ec2b70c0 100644 --- a/database/versions/262735d025da_2_0_1.py +++ b/database/versions/262735d025da_2_0_1.py @@ -6,8 +6,8 @@ Create Date: 2024-09-11 08:07:02.753307 """ -from app.db.oper.systemconfig import SystemConfigOper -from app.schemas.types import SystemConfigKey +from alembic import op +import sqlalchemy as sa # revision identifiers, used by Alembic. revision = '262735d025da' @@ -19,9 +19,18 @@ depends_on = None def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### # 初始化消息通知范围 - _systemconfig = SystemConfigOper() - if not _systemconfig.get(SystemConfigKey.NotificationSwitchs): - _systemconfig.set(SystemConfigKey.NotificationSwitchs, [ + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + key = "NotificationSwitchs" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + if not row or not row[0]: + value = [ { 'type': '资源下载', 'action': 'all', @@ -54,7 +63,15 @@ def upgrade() -> None: 'type': '其它', 'action': 'admin', }, - ]) + ] + if row: + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=value + ) + ) + else: + connection.execute(systemconfig.insert().values(key=key, value=value)) # ### end Alembic commands ### diff --git a/database/versions/294b007932ef_2_0_0.py b/database/versions/294b007932ef_2_0_0.py index 84ec7974d..de97ed496 100644 --- a/database/versions/294b007932ef_2_0_0.py +++ b/database/versions/294b007932ef_2_0_0.py @@ -8,13 +8,11 @@ Create Date: 2024-07-20 08:43:40.741251 import secrets +from alembic import op +import sqlalchemy as sa from app.runtime.config import settings from app.application.security.token import get_password_hash -from app.db import SessionFactory -from app.db.models import * -from app.db.oper.systemconfig import SystemConfigOper from app.runtime.log import logger -from app.schemas.types import SystemConfigKey # revision identifiers, used by Alembic. revision = '294b007932ef' @@ -27,50 +25,88 @@ def upgrade() -> None: """ v2.0.0 数据库初始化 """ - with SessionFactory() as db: - # 初始化超级管理员 - _user = User.get_by_name(db=db, name=settings.SUPERUSER) - if not _user: - if settings.SUPERUSER_PASSWORD: - init_password = settings.SUPERUSER_PASSWORD - else: - # 生成随机密码 - init_password = secrets.token_urlsafe(16) - logger.info( - f"【超级管理员初始密码】{init_password} 请登录系统后在设定中修改。 注：该密码只会显示一次，请注意保存。") - _user = User( + connection = op.get_bind() + user = sa.table( + "user", + sa.column("name", sa.String()), + sa.column("email", sa.String()), + sa.column("hashed_password", sa.String()), + sa.column("is_active", sa.Boolean()), + sa.column("is_superuser", sa.Boolean()), + sa.column("avatar", sa.String()), + sa.column("is_otp", sa.Boolean()), + sa.column("otp_secret", sa.String()), + sa.column("permissions", sa.JSON()), + sa.column("settings", sa.JSON()), + ) + # 初始化超级管理员 + existing_user = connection.execute( + sa.select(user.c.name).where(user.c.name == settings.SUPERUSER) + ).first() + if not existing_user: + if settings.SUPERUSER_PASSWORD: + init_password = settings.SUPERUSER_PASSWORD + else: + # 生成随机密码 + init_password = secrets.token_urlsafe(16) + logger.info( + f"【超级管理员初始密码】{init_password} 请登录系统后在设定中修改。 注：该密码只会显示一次，请注意保存。") + connection.execute( + user.insert().values( name=settings.SUPERUSER, hashed_password=get_password_hash(init_password), email="admin@movie-pilot.org", + is_active=True, is_superuser=True, - avatar="" + avatar="", + is_otp=False, + otp_secret=None, + permissions={}, + settings={}, ) - _user.create(db) - # 初始化本地存储 - _systemconfig = SystemConfigOper() - if not _systemconfig.get(SystemConfigKey.Storages): - _systemconfig.set(SystemConfigKey.Storages, [ - { - "type": "local", - "name": "本地", - "config": {} - }, - { - "type": "alipan", - "name": "阿里云盘", - "config": {} - }, - { - "type": "u115", - "name": "115网盘", - "config": {} - }, - { - "type": "rclone", - "name": "RClone", - "config": {} - } - ]) + ) + + # 初始化本地存储 + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + key = "Storages" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + if not row or not row[0]: + value = [ + { + "type": "local", + "name": "本地", + "config": {} + }, + { + "type": "alipan", + "name": "阿里云盘", + "config": {} + }, + { + "type": "u115", + "name": "115网盘", + "config": {} + }, + { + "type": "rclone", + "name": "RClone", + "config": {} + } + ] + if row: + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=value + ) + ) + else: + connection.execute(systemconfig.insert().values(key=key, value=value)) def downgrade() -> None: diff --git a/database/versions/3891a5e722a1_2_1_7.py b/database/versions/3891a5e722a1_2_1_7.py index 2e1e692db..9c8d85687 100644 --- a/database/versions/3891a5e722a1_2_1_7.py +++ b/database/versions/3891a5e722a1_2_1_7.py @@ -7,10 +7,9 @@ Create Date: 2025-06-28 08:40:14.516836 """ from alembic import op import sqlalchemy as sa + from sqlalchemy.dialects import sqlite -from app.db.oper.systemconfig import SystemConfigOper -from app.schemas.types import SystemConfigKey # revision identifiers, used by Alembic. revision = '3891a5e722a1' @@ -22,14 +21,31 @@ depends_on = None def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### # rename AList存储 - _systemconfig = SystemConfigOper() - _storages = _systemconfig.get(SystemConfigKey.Storages) + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + key = "Storages" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + _storages = row[0] if row else None if _storages: + changed = False for storage in _storages: if storage["type"] == "alist": - storage["name"] = "OpenList" + if storage.get("name") != "OpenList": + storage["name"] = "OpenList" + changed = True break - _systemconfig.set(SystemConfigKey.Storages, _storages) + if changed: + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=_storages + ) + ) # ### end Alembic commands ### diff --git a/database/versions/3df653756eec_2_1_6.py b/database/versions/3df653756eec_2_1_6.py index e4ee5d1d2..0a69bed4f 100644 --- a/database/versions/3df653756eec_2_1_6.py +++ b/database/versions/3df653756eec_2_1_6.py @@ -5,10 +5,8 @@ Revises: 486e56a62dcb Create Date: 2025-06-11 19:52:57.185355 """ -import json - -from app.db import SessionFactory -from app.db.models import User +from alembic import op +import sqlalchemy as sa # revision identifiers, used by Alembic. revision = '3df653756eec' @@ -18,24 +16,30 @@ depends_on = None def upgrade() -> None: - # ### commands auto generated by Alembic - please adjust! ### - with SessionFactory() as db: - # 所有用户 - users = User.list(db) - for user in users: - if user.is_superuser: - continue - if not user.permissions: - permissions = { - "discovery": True, - "search": True, - "subscribe": True, - "manage": False, - } - user.update(db, { - "permissions": permissions, - }) - # ### end Alembic commands ### + connection = op.get_bind() + user = sa.table( + "user", + sa.column("id", sa.Integer()), + sa.column("is_superuser", sa.Boolean()), + sa.column("permissions", sa.JSON()), + ) + users = connection.execute( + sa.select(user.c.id, user.c.is_superuser, user.c.permissions) + ).mappings().all() + permissions = { + "discovery": True, + "search": True, + "subscribe": True, + "manage": False, + } + for item in users: + if item["is_superuser"] or item["permissions"]: + continue + connection.execute( + user.update() + .where(user.c.id == item["id"]) + .values(permissions=permissions) + ) def downgrade() -> None: diff --git a/database/versions/486e56a62dcb_2_1_5.py b/database/versions/486e56a62dcb_2_1_5.py index 5d20b605e..3e2213cff 100644 --- a/database/versions/486e56a62dcb_2_1_5.py +++ b/database/versions/486e56a62dcb_2_1_5.py @@ -7,8 +7,8 @@ Create Date: 2025-05-13 19:49:51.271319 """ import re -from app.db.oper.systemconfig import SystemConfigOper -from app.schemas.types import SystemConfigKey +from alembic import op +import sqlalchemy as sa # revision identifiers, used by Alembic. revision = '486e56a62dcb' @@ -20,16 +20,33 @@ depends_on = None def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### ### 将消息模板中的 `season`(为单数字, 且重命名需要这个字段)替换为 `season_fmt`(Sxx格式字符串) ### - _systemconfig = SystemConfigOper() - templates = _systemconfig.get(SystemConfigKey.NotificationTemplates) + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + key = "NotificationTemplates" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + templates = row[0] if row else None if isinstance(templates, dict): _re = r'(?<={{)(?![^}]*[%|])(\s*)season(\s*)(?=}})|(?<={%)if\s+(?![^%]*[%|])season\s*(?=%)' + changed = False for k, v in templates.items(): # 替换season为season_fmt result = re.sub(_re, r'\1season_fmt\2', v) + if result != v: + changed = True templates[k] = result # 将更新后的模板存回系统配置 - _systemconfig.set(SystemConfigKey.NotificationTemplates, templates) + if changed: + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=templates + ) + ) # ### end Alembic commands ### diff --git a/database/versions/4dadad1d161a_3_0_0.py b/database/versions/4dadad1d161a_3_0_0.py index 899e3806c..53d28dec1 100644 --- a/database/versions/4dadad1d161a_3_0_0.py +++ b/database/versions/4dadad1d161a_3_0_0.py @@ -6,9 +6,9 @@ Revises: e8b1c4d7a2f9 Create Date: 2026-08-10 """ -from app.db.oper.systemconfig import SystemConfigOper +from alembic import op +import sqlalchemy as sa from app.runtime.log import logger -from app.schemas.types import SystemConfigKey # revision identifiers, used by Alembic. revision = "4dadad1d161a" @@ -21,7 +21,17 @@ def upgrade() -> None: # V3 为大版本升级，通知模板直接覆盖用户旧设置，且迁移只执行一次； # 默认模板同时兼容影视与音乐（音乐的下载、入库通知补齐艺术家/专辑/音质信息）。 # 覆盖前先将用户现有模板完整输出到日志，作为备份供用户恢复参考。 - old_value = SystemConfigOper().get(SystemConfigKey.NotificationTemplates) + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + key = "NotificationTemplates" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + old_value = row[0] if row else None if old_value: logger.info(f"即将使用 V3 默认通知模板覆盖用户现有通知模板，现有模板内容备份如下：\n{old_value}") value = { @@ -67,8 +77,15 @@ def upgrade() -> None: '{% if overview %}\\n简介：{{ overview }}{% endif %}' }""" } - SystemConfigOper().set(SystemConfigKey.NotificationTemplates, value) + if row and row[0] != value: + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=value + ) + ) + elif not row: + connection.execute(systemconfig.insert().values(key=key, value=value)) def downgrade() -> None: - pass \ No newline at end of file + pass diff --git a/database/versions/89d24811e894_2_1_4.py b/database/versions/89d24811e894_2_1_4.py index 7ae0f0de6..75b6fd485 100644 --- a/database/versions/89d24811e894_2_1_4.py +++ b/database/versions/89d24811e894_2_1_4.py @@ -6,8 +6,8 @@ Create Date: 2025-05-03 17:29:07.635618 """ -from app.db.oper.systemconfig import SystemConfigOper -from app.schemas.types import SystemConfigKey +from alembic import op +import sqlalchemy as sa # revision identifiers, used by Alembic. revision = '89d24811e894' @@ -57,9 +57,25 @@ def upgrade() -> None: '{% if overview %}\\n简介：{{ overview }}{% endif %}' }""" } - _systemconfig = SystemConfigOper() - if not _systemconfig.get(SystemConfigKey.NotificationTemplates): - _systemconfig.set(SystemConfigKey.NotificationTemplates, value) + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + key = "NotificationTemplates" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + if not row or not row[0]: + if row: + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=value + ) + ) + else: + connection.execute(systemconfig.insert().values(key=key, value=value)) # ### end Alembic commands ### diff --git a/database/versions/a295e41830a6_2_0_6.py b/database/versions/a295e41830a6_2_0_6.py index d293aa319..b4e8da153 100644 --- a/database/versions/a295e41830a6_2_0_6.py +++ b/database/versions/a295e41830a6_2_0_6.py @@ -6,8 +6,8 @@ Create Date: 2024-11-14 12:49:13.838120 """ -from app.db.oper.systemconfig import SystemConfigOper -from app.schemas.types import SystemConfigKey +from alembic import op +import sqlalchemy as sa # revision identifiers, used by Alembic. revision = 'a295e41830a6' @@ -19,16 +19,28 @@ depends_on = None def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### # 初始化AList存储 - _systemconfig = SystemConfigOper() - _storages = _systemconfig.get(SystemConfigKey.Storages) - if _storages: - if "alist" not in [storage["type"] for storage in _storages]: - _storages.append({ + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + key = "Storages" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + _storages = row[0] if row else None + if _storages and "alist" not in [storage["type"] for storage in _storages]: + _storages.append({ "type": "alist", "name": "AList", "config": {} }) - _systemconfig.set(SystemConfigKey.Storages, _storages) + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=_storages + ) + ) # ### end Alembic commands ### diff --git a/database/versions/a73f2dbf5c09_2_0_4.py b/database/versions/a73f2dbf5c09_2_0_4.py index de542ce3a..33377973c 100644 --- a/database/versions/a73f2dbf5c09_2_0_4.py +++ b/database/versions/a73f2dbf5c09_2_0_4.py @@ -6,8 +6,8 @@ Create Date: 2024-10-16 15:05:01.775429 """ -from app.db.oper.systemconfig import SystemConfigOper -from app.schemas.types import SystemConfigKey +from alembic import op +import sqlalchemy as sa # revision identifiers, used by Alembic. revision = 'a73f2dbf5c09' @@ -19,7 +19,25 @@ depends_on = None def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### # 初始化下载优先规则 - SystemConfigOper().set(SystemConfigKey.TorrentsPriority, ["torrent", "upload", "seeder"]) + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + key = "TorrentsPriority" + value = ["torrent", "upload", "seeder"] + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == key) + ).first() + if row and row[0] != value: + connection.execute( + systemconfig.update().where(systemconfig.c.key == key).values( + value=value + ) + ) + elif not row: + connection.execute(systemconfig.insert().values(key=key, value=value)) # ### end Alembic commands ### diff --git a/database/versions/e2dbe1421fa4_2_0_3.py b/database/versions/e2dbe1421fa4_2_0_3.py index bd23d13e5..d9f928919 100644 --- a/database/versions/e2dbe1421fa4_2_0_3.py +++ b/database/versions/e2dbe1421fa4_2_0_3.py @@ -5,14 +5,10 @@ Revises: 0fb94bf69b38 Create Date: 2024-10-09 13:44:13.926529 """ -import contextlib - from alembic import op import sqlalchemy as sa from app.runtime.log import logger -from app.db import SessionFactory -from app.db.models import UserConfig # revision identifiers, used by Alembic. revision = 'e2dbe1421fa4' @@ -72,8 +68,7 @@ def upgrade() -> None: except Exception as e: logger.error(f"Could not alter column {column_name} in table {table}: {e}") - with SessionFactory() as db: - UserConfig.truncate(db) + conn.execute(sa.delete(sa.table("userconfig"))) def downgrade() -> None: diff --git a/database/versions/e8b1c4d7a2f9_2_2_18.py b/database/versions/e8b1c4d7a2f9_2_2_18.py index 82bd3ff32..e6e350815 100644 --- a/database/versions/e8b1c4d7a2f9_2_2_18.py +++ b/database/versions/e8b1c4d7a2f9_2_2_18.py @@ -63,9 +63,6 @@ def upgrade() -> None: ]) # 只升级系统旧默认模板；用户编辑过的模板保持原样。 - from app.db.oper.systemconfig import SystemConfigOper - from app.schemas.types import SystemConfigKey - legacy_organize = """ { 'title': '{{ title_year }}' @@ -123,18 +120,31 @@ def upgrade() -> None: '{% if labels %}\\n标签：{{ labels }}{% endif %}' '{% if description %}\\n描述：{{ description }}{% endif %}' }""" - config_oper = SystemConfigOper() - templates = dict(config_oper.get(SystemConfigKey.NotificationTemplates) or {}) + systemconfig = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + connection = op.get_bind() + config_key = "NotificationTemplates" + row = connection.execute( + sa.select(systemconfig.c.value).where(systemconfig.c.key == config_key) + ).first() + templates = dict(row[0] or {}) if row else {} changed = False - for key, legacy, replacement in ( + for template_key, legacy, replacement in ( ("organizeSuccess", legacy_organize, music_organize), ("downloadAdded", legacy_download, music_download), ): - if str(templates.get(key) or "").strip() == legacy.strip(): - templates[key] = replacement + if str(templates.get(template_key) or "").strip() == legacy.strip(): + templates[template_key] = replacement changed = True if changed: - config_oper.set(SystemConfigKey.NotificationTemplates, templates) + connection.execute( + systemconfig.update().where(systemconfig.c.key == config_key).values( + value=templates + ) + ) def downgrade() -> None: diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index a85878eff..6936fe686 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6404, - "edge_sha256": "850e490b866ed8eeee03515e39a3821c3078f934c622a60d9db5e279bab1bfbb", + "edge_count": 6401, + "edge_sha256": "cfc86c48464cace2831f69ccaf8bb9ad1c40a1a4626cd11b8d2bf10d5508b0b8", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -5983,14 +5983,11 @@ "app.startup.database_initializer -> app.db.base", "app.startup.database_initializer -> app.db.engine", "app.startup.database_initializer -> app.db.models", - "app.startup.database_initializer -> app.db.session", - "app.startup.database_initializer -> app.db.uow", "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.database_initializer -> app.startup.transaction", "app.startup.domain_initializer -> app.adapters", "app.startup.domain_initializer -> app.adapters.system", "app.startup.domain_initializer -> app.adapters.system.rust", diff --git a/tests/test_database_migration_startup.py b/tests/test_database_migration_startup.py index 80bcb4d1f..033c01ef5 100644 --- a/tests/test_database_migration_startup.py +++ b/tests/test_database_migration_startup.py @@ -1,3 +1,4 @@ +import importlib import importlib.util from pathlib import Path import sys @@ -5,10 +6,25 @@ from types import SimpleNamespace from unittest.mock import Mock import uuid +from alembic.migration import MigrationContext +from alembic.operations import Operations import pytest from alembic.util import CommandError from fastapi import FastAPI -from sqlalchemy import Column, Integer, MetaData, Table, create_engine, inspect, text +import sqlalchemy as sa +from sqlalchemy import ( + Boolean, + Column, + Integer, + JSON, + MetaData, + String, + Table, + create_engine, + event, + inspect, + text, +) from sqlalchemy.engine.url import make_url from app.startup import database_initializer as db_init @@ -82,7 +98,12 @@ def test_prepare_database_creates_backup_before_schema_changes(monkeypatch) -> N db_init.prepare_database(before_alembic=lambda: calls.append("before_alembic")) - assert calls == ["backup", "create_all", "before_alembic", "alembic"] + assert calls == [ + "backup", + "create_all", + "before_alembic", + "alembic", + ] assert logged_messages == [ "数据库需要从版本 old 升级到 head，正在创建迁移前备份" ] @@ -434,6 +455,315 @@ def downgrade(): assert active_columns == {"id", "migrated"} +def test_migration_config_write_rolls_back_with_alembic_transaction( + monkeypatch, +) -> None: + """配置 DML 不得脱离 Alembic 事务提前提交。""" + migration = importlib.import_module( + "database.versions.e8b1c4d7a2f9_2_2_18" + ) + engine = create_engine("sqlite://") + + metadata = MetaData() + systemconfig = Table( + "systemconfig", + metadata, + Column("id", Integer, primary_key=True), + Column("key", String), + Column("value", JSON), + ) + for table_name in ("subscribe", "subscribehistory", "transferhistory"): + Table(table_name, metadata, Column("id", Integer, primary_key=True)) + metadata.create_all(engine) + + legacy_organize = """ +{ + 'title': '{{ title_year }}' + '{% if season_episode %} {{ season_episode }}{% endif %} 已入库', + 'text': '{% if vote_average %}评分：{{ vote_average }}，{% endif %}' + '类型：{{ type }}' + '{% if category %}，类别：{{ category }}{% endif %}' + '{% if resource_term %}，质量：{{ resource_term }}{% endif %}，' + '共{{ file_count }}个文件，大小：{{ total_size }}' + '{% if err_msg %}，以下文件处理失败：{{ err_msg }}{% endif %}' +}""" + legacy_download = """ +{ + 'title': '{{ title_year }}' + '{% if download_episodes %} {{ season_fmt }} {{ download_episodes }}{% else %}{{ season_episode }}{% endif %} 开始下载', + 'text': '{% if site_name %}站点：{{ site_name }}{% endif %}' + '{% if resource_term %}\\n质量：{{ resource_term }}{% endif %}' + '{% if size %}\\n大小：{{ size }}{% endif %}' + '{% if torrent_title %}\\n种子：{{ torrent_title }}{% endif %}' + '{% if pubdate %}\\n发布时间：{{ pubdate }}{% endif %}' + '{% if freedate %}\\n免费时间：{{ freedate }}{% endif %}' + '{% if seeders %}\\n做种数：{{ seeders }}{% endif %}' + '{% if volume_factor %}\\n促销：{{ volume_factor }}{% endif %}' + '{% if hit_and_run %}\\nHit&Run：{{ hit_and_run }}{% endif %}' + '{% if labels %}\\n标签：{{ labels }}{% endif %}' + '{% if description %}\\n描述：{{ description }}{% endif %}' +}""" + original_templates = { + "organizeSuccess": legacy_organize, + "downloadAdded": legacy_download, + } + with engine.begin() as connection: + connection.execute( + systemconfig.insert().values( + key="NotificationTemplates", + value=original_templates, + ) + ) + + with engine.connect() as connection: + transaction = connection.begin() + monkeypatch.setattr( + migration, + "op", + Operations(MigrationContext.configure(connection)), + ) + config_write_seen = False + + def fail_after_config_write( + _connection, + _cursor, + statement, + _parameters, + _context, + _executemany, + ) -> None: + nonlocal config_write_seen + if statement.lstrip().upper().startswith("UPDATE SYSTEMCONFIG"): + config_write_seen = True + raise RuntimeError("injected migration failure") + + event.listen(engine, "after_cursor_execute", fail_after_config_write) + try: + with pytest.raises(RuntimeError, match="injected migration failure"): + migration.upgrade() + finally: + event.remove(engine, "after_cursor_execute", fail_after_config_write) + transaction.rollback() + + assert config_write_seen + + with engine.connect() as connection: + columns = { + column["name"] + for column in inspect(connection).get_columns("subscribe") + } + # SQLite 默认驱动不回滚 DDL，但配置写入仍必须服从 Alembic 事务。 + assert "audio_quality" in columns + stored_templates = connection.execute( + systemconfig.select().with_only_columns(systemconfig.c.value).where( + systemconfig.c.key == "NotificationTemplates" + ) + ).scalar_one() + assert stored_templates == original_templates + + +def test_initial_migration_rolls_back_user_and_storages_together( + monkeypatch, +) -> None: + """2.0.0 管理员与存储初始化必须共享 Alembic 事务。""" + migration = importlib.import_module( + "database.versions.294b007932ef_2_0_0" + ) + engine = create_engine("sqlite://") + + metadata = MetaData() + Table( + "user", + metadata, + Column("id", Integer, primary_key=True), + Column("name", String, nullable=False), + Column("email", String), + Column("hashed_password", String), + Column("is_active", Boolean), + Column("is_superuser", Boolean), + Column("avatar", String), + Column("is_otp", Boolean), + Column("otp_secret", String), + Column("permissions", JSON), + Column("settings", JSON), + ) + systemconfig = Table( + "systemconfig", + metadata, + Column("id", Integer, primary_key=True), + Column("key", String), + Column("value", JSON), + ) + metadata.create_all(engine) + + monkeypatch.setattr(migration.settings, "SUPERUSER", "migration-admin") + monkeypatch.setattr( + migration.settings, + "SUPERUSER_PASSWORD", + "migration-password", + ) + monkeypatch.setattr( + migration, + "get_password_hash", + lambda password: f"hashed:{password}", + ) + + with engine.connect() as connection: + transaction = connection.begin() + monkeypatch.setattr( + migration, + "op", + Operations(MigrationContext.configure(connection)), + ) + + def fail_storages_write( + _connection, + _cursor, + statement, + _parameters, + _context, + _executemany, + ) -> None: + if statement.lstrip().upper().startswith("INSERT INTO SYSTEMCONFIG"): + raise RuntimeError("injected storages failure") + + event.listen(engine, "after_cursor_execute", fail_storages_write) + try: + with pytest.raises(RuntimeError, match="injected storages failure"): + migration.upgrade() + finally: + event.remove(engine, "after_cursor_execute", fail_storages_write) + transaction.rollback() + + with engine.connect() as connection: + assert connection.execute(text("SELECT COUNT(*) FROM user")).scalar_one() == 0 + assert connection.execute( + text("SELECT COUNT(*) FROM systemconfig") + ).scalar_one() == 0 + + +def test_userconfig_cleanup_migration_uses_alembic_transaction( + monkeypatch, +) -> None: + """2.0.3 用户配置清理必须随当前 Alembic 事务一起回滚。""" + migration = importlib.import_module( + "database.versions.e2dbe1421fa4_2_0_3" + ) + engine = create_engine("sqlite://") + + metadata = MetaData() + table_columns = { + "downloadhistory": (("note", JSON), ("media_category", String)), + "subscribe": ( + ("note", JSON), + ("custom_words", String), + ("media_category", String), + ("filter_groups", JSON), + ), + "mediaserveritem": (("note", JSON),), + "message": (("note", JSON),), + "plugindata": (("value", JSON),), + "site": (("note", JSON),), + "sitestatistic": (("note", JSON),), + "systemconfig": (("value", JSON),), + "userconfig": (("value", JSON),), + } + tables = { + table_name: Table( + table_name, + metadata, + Column("id", Integer, primary_key=True), + *(Column(column_name, column_type) for column_name, column_type in columns), + ) + for table_name, columns in table_columns.items() + } + metadata.create_all(engine) + with engine.begin() as connection: + connection.execute( + tables["userconfig"].insert().values(value={"retained": True}) + ) + + with engine.connect() as connection: + transaction = connection.begin() + monkeypatch.setattr( + migration, + "op", + Operations(MigrationContext.configure(connection)), + ) + migration.upgrade() + assert connection.execute( + sa.select(sa.func.count()).select_from(tables["userconfig"]) + ).scalar_one() == 0 + transaction.rollback() + + with engine.connect() as connection: + assert connection.execute( + sa.select(sa.func.count()).select_from(tables["userconfig"]) + ).scalar_one() == 1 + + +def test_user_permission_migration_uses_alembic_transaction( + monkeypatch, +) -> None: + """2.1.6 权限初始化必须保留原筛选语义并随迁移事务回滚。""" + migration = importlib.import_module( + "database.versions.3df653756eec_2_1_6" + ) + engine = create_engine("sqlite://") + + metadata = MetaData() + user = Table( + "user", + metadata, + Column("id", Integer, primary_key=True), + Column("is_superuser", Boolean), + Column("permissions", JSON), + ) + metadata.create_all(engine) + existing_permissions = {"manage": True} + with engine.begin() as connection: + connection.execute(user.insert(), [ + {"id": 1, "is_superuser": False, "permissions": None}, + {"id": 2, "is_superuser": False, "permissions": existing_permissions}, + {"id": 3, "is_superuser": True, "permissions": None}, + ]) + + with engine.connect() as connection: + transaction = connection.begin() + monkeypatch.setattr( + migration, + "op", + Operations(MigrationContext.configure(connection)), + ) + migration.upgrade() + migrated = connection.execute( + sa.select(user.c.id, user.c.permissions).order_by(user.c.id) + ).all() + assert migrated == [ + ( + 1, + { + "discovery": True, + "search": True, + "subscribe": True, + "manage": False, + }, + ), + (2, existing_permissions), + (3, None), + ] + transaction.rollback() + + with engine.connect() as connection: + assert connection.execute( + sa.select(user.c.id, user.c.permissions).order_by(user.c.id) + ).all() == [ + (1, None), + (2, existing_permissions), + (3, None), + ] + + def test_local_setup_returns_failure_when_database_migration_fails( monkeypatch, capsys, diff --git a/tests/test_notification_template_render.py b/tests/test_notification_template_render.py index 5c9f17545..75bceec64 100644 --- a/tests/test_notification_template_render.py +++ b/tests/test_notification_template_render.py @@ -16,6 +16,9 @@ from pathlib import Path import importlib.util import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations from app.domain.context import MUSIC_ENTITY_ALBUM, MusicInfo from app.domain.meta.metamusic import MetaMusic @@ -207,9 +210,7 @@ def test_message_without_template_config_stays_unchanged( assert message.text is None -def test_v3_migration_overwrites_templates_once( - notification_templates: SystemConfigOper, -) -> None: +def test_v3_migration_overwrites_templates_once(monkeypatch) -> None: """ V3 大版本迁移应无条件覆盖用户旧通知模板配置，并写入全部 4 类模板。 """ @@ -221,12 +222,32 @@ def test_v3_migration_overwrites_templates_once( migration = importlib.util.module_from_spec(spec) spec.loader.exec_module(migration) - notification_templates.set( - SystemConfigKey.NotificationTemplates, {"organizeSuccess": "用户自定义模板"} + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + systemconfig = sa.Table( + "systemconfig", + metadata, + sa.Column("key", sa.String(), primary_key=True), + sa.Column("value", sa.JSON()), ) - migration.upgrade() + with engine.begin() as connection: + metadata.create_all(connection) + connection.execute( + systemconfig.insert().values( + key=SystemConfigKey.NotificationTemplates.value, + value={"organizeSuccess": "用户自定义模板"}, + ) + ) + context = MigrationContext.configure(connection) + monkeypatch.setattr(migration, "op", Operations(context)) + + migration.upgrade() + templates = connection.execute( + sa.select(systemconfig.c.value).where( + systemconfig.c.key == SystemConfigKey.NotificationTemplates.value + ) + ).scalar_one() - templates = notification_templates.get(SystemConfigKey.NotificationTemplates) assert set(templates.keys()) == { "organizeSuccess", "downloadAdded", "subscribeAdded", "subscribeComplete", } diff --git a/tests/test_transferhistory_media_source_migration.py b/tests/test_transferhistory_media_source_migration.py index d4a0ac338..289341435 100644 --- a/tests/test_transferhistory_media_source_migration.py +++ b/tests/test_transferhistory_media_source_migration.py @@ -11,7 +11,8 @@ from app.domain.context import MUSIC_ENTITY_ALBUM, MusicInfo from app.domain.meta.metabase import MetaBase from app.domain.meta.metamusic import MetaMusic from app.db.oper.transferhistory import TransferHistoryOper -from app.schemas import FileItem, TransferInfo +from app.schemas.file import FileItem +from app.schemas.transfer import TransferInfo def test_transferhistory_migration_backfills_existing_source_ids(monkeypatch) -> None: @@ -156,18 +157,25 @@ def test_music_audio_quality_migration_is_idempotent(monkeypatch) -> None: metadata, sa.Column("id", sa.Integer(), primary_key=True), ) - config_oper = Mock() - config_oper.get.return_value = { + systemconfig = sa.Table( + "systemconfig", + metadata, + sa.Column("key", sa.String(), primary_key=True), + sa.Column("value", sa.JSON()), + ) + custom_templates = { "organizeSuccess": "custom organize template", "downloadAdded": "custom download template", } - monkeypatch.setattr( - "app.db.oper.systemconfig.SystemConfigOper", - lambda: config_oper, - ) with engine.begin() as connection: metadata.create_all(connection) + connection.execute( + systemconfig.insert().values( + key="NotificationTemplates", + value=custom_templates, + ) + ) context = MigrationContext.configure(connection) monkeypatch.setattr(migration, "op", Operations(context)) @@ -186,6 +194,11 @@ def test_music_audio_quality_migration_is_idempotent(monkeypatch) -> None: column["name"] for column in inspector.get_columns("transferhistory") } + stored_templates = connection.execute( + sa.select(systemconfig.c.value).where( + systemconfig.c.key == "NotificationTemplates" + ) + ).scalar_one() assert { "audio_quality", @@ -206,7 +219,7 @@ def test_music_audio_quality_migration_is_idempotent(monkeypatch) -> None: "sample_rate", "bitrate", }.issubset(transfer_columns) - config_oper.set.assert_not_called() + assert stored_templates == custom_templates def test_transfer_history_preserves_album_entity_context() -> None: