test(db): 测试改用真实 MySQL，方言差异不再漏到生产

生产是 MySQL 8、测试是 SQLite，两者的方言差异在本地完全测不出来。今天为此
付出了两次代价：

1. `ORDER BY ck_checked_at ASC NULLS FIRST` 在 SQLite 上完全正常，本地 258 条
   全绿，一部署到生产第一次巡检就 1064 语法错误（MySQL 不支持 NULLS FIRST）；
2. P1b 自动重登的结算复用调度器事务读取，在 SQLite 下看不出问题，但在 MySQL 的
   REPEATABLE READ 下读到的是批次开始前的快照 —— 10 条实际成功的重登全部被记成
   失败。SQLite 的读一致性与 MySQL 不同，这类问题跑一万遍也复现不出来。

改动：
- 新增 tests/support/db.py：测试专用引擎/会话工具，连接串从 .env 读取，
  默认指向独立的 douyu_login_test 库；
- 新增 tests/conftest.py：会话级建好表结构，每个用例前后清空所有表做隔离；
- 根 conftest.py 接管连接串设置：它在 pytest 收集阶段最先执行，在这里把
  DATABASE_URL 指向测试库，应用模块首次 import 拿到就是 MySQL，不依赖任何
  import 顺序（原先这里是 sqlite:// 兜底，改造后会变成「先建 SQLite engine
  再打补丁换成 MySQL」，顺序一变就静默跑错库）；
- 21 个测试文件的 setup 样板统一改用共享测试引擎；
- 两个迁移测试改为在临时库 douyu_login_mig_<随机> 上跑完整 alembic upgrade head，
  跑完 drop —— 迁移脚本本身也终于在 MySQL 上被验证；
- dev.sh 与 README 的过时说明一并更新（原先写着「不需要 MySQL / .env 配置」；
  「测试各自建 SQLite 内存库」）。

安全与易用：
- 测试库名与开发库同名时**直接拒绝运行**（clear_tables 会 DELETE 所有表，
  一旦 TEST_MYSQL_DATABASE 被指到 douyu_login 就是数据灾难）；
- 连不上 MySQL 时给出可操作提示（先 ./dev.sh 起库 / 用 TEST_MYSQL_DATABASE 覆盖），
  而不是抛一段 SQLAlchemy 原始堆栈。

验收：
- 全量 301 条通过，连跑两次结果一致（隔离有效），耗时约 12.5 秒（原 SQLite 约 3.6 秒）；
- 单文件独立运行通过（不依赖全量执行顺序）；
- `grep -rn sqlite tests/ conftest.py` 已无实际用法；
- **方言确实生效**：临时把一条查询改回 `NULLS FIRST`，13 条测试立刻变红（含运行期
  失败），还原即恢复；这条由 test_generated_sql_is_mysql_compatible 长期守着。

代价（已在 README 与规划附录 C.2 记明）：跑测试前必须先有 MySQL，pre-push 钩子
同理；CI 需要 service container。

README.md

@@ -26,7 +26,7 @@ live-hub-py/
 │   ├── frontend/              # React + Ant Design + Vite
 │   └── extension/             # 虎牙浏览器兑换助手
 ├── services/yyb-worker/        # 应用宝协议运行时（随主镜像部署）
-├── tests/                      # pytest（默认 SQLite 内存库）
+├── tests/                      # pytest（跑在真实 MySQL 测试库上）
 ├── docs/                       # 协议与联调文档（结论以「虎牙 Cookie 路线状态」为准）
 ├── deploy.sh / dev.sh          # 生产部署 / 本机热更新
 └── docker-compose.yml         # 应用容器 + MySQL 8.4
@@ -111,7 +111,17 @@ uv run --group dev pytest tests/test_cookie_operations.py
 docker compose --profile test run --rm test
 ```
 
-测试各自建 SQLite 内存库；根目录 `conftest.py` 兜底 `DATABASE_URL` / `APP_ENCRYPTION_KEY`，防止误连开发/生产库。迁移链冒烟见 `tests/test_migrations.py`。兑换、下单、支付属于写操作，默认用 mock 做契约测试，不打真实支付。
+测试跑在**真实 MySQL** 上（不再使用 SQLite）：生产是 MySQL，SQLite 与它有方言差异，
+曾出现「`ORDER BY ... NULLS FIRST` 本地全绿、生产直接 1064」的事故。因此：
+
+- 先起数据库：`./dev.sh`（docker compose，MySQL 映射到 `127.0.0.1:3307`）；
+- 测试库默认 `douyu_login_test`，可用 `TEST_MYSQL_DATABASE` 覆盖；**库名与开发库相同时
+  会被 `tests/support/db.py` 拒绝运行**（每个用例前后会清空所有表，防误清）；
+- `tests/conftest.py` 在导入应用模块前把连接串指向测试库，并把每用例的表清空交给 autouse fixture；
+- 迁移链冒烟见 `tests/test_migrations.py`：它在临时库 `douyu_login_mig_<随机>` 上跑完整
+  `alembic upgrade head`，跑完 drop。
+
+兑换、下单、支付属于写操作，默认用 mock 做契约测试，不打真实支付。
 
 启用提交/推送门禁（改过前端会跑 TypeScript，与 Docker `npm run build` 的 `tsc -b` 一致）：
 
@@ -127,7 +137,7 @@ docker compose --profile test run --rm test
 
 ## 技术栈
 
-- 后端：FastAPI + SQLAlchemy + MySQL 8.4（测试用 SQLite）+ JWT
+- 后端：FastAPI + SQLAlchemy + MySQL 8.4（测试同样跑 MySQL）+ JWT
 - 前端：React + Ant Design + Vite
 - 核心：Python 3.12 + OpenCV + ONNX + pycryptodome + curl-cffi
 - 迁移：Alembic

conftest.py

@@ -1,14 +1,20 @@
-"""pytest 全局引导：在导入任何应用模块前固定测试数据库与加密密钥。
+"""pytest 全局引导：在导入任何应用模块前定好测试库与加密密钥。
 
-web/backend 的模块在 import 阶段就会读取 DATABASE_URL 创建全局引擎、
+这个文件在 pytest 收集阶段**最先执行**，所以「让测试连上测试库」这件事放在这里
-读取 APP_ENCRYPTION_KEY 初始化敏感字段加密，因此这两个环境变量必须在
+最稳 —— 不依赖任何测试文件或 tests/conftest.py 的 import 顺序。
-测试模块加载之前设置好。各测试文件自身使用独立的 SQLite 内存引擎做
+
-数据隔离，这里统一兜底，避免误连开发/生产数据库。
+背景：web/backend 的模块在 import 阶段就会读 DATABASE_URL 建全局 engine、读
+APP_ENCRYPTION_KEY 初始化敏感字段加密。之前这里塞的是 ``sqlite://`` 兜底，
+测试改用 MySQL 后就会变成「先按 SQLite 建 engine、再靠后面打补丁换成 MySQL」，
+顺序一变就静默跑错库。
 """
 
 import os
 
-os.environ.setdefault("DATABASE_URL", "sqlite://")
 os.environ.setdefault(
     "APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="
 )
+
+# 导入即完成设置：tests/support/db.py 会把 DATABASE_URL 指向独立测试库
+# （默认 douyu_login_test；库名与开发库相同时直接拒绝运行，防止清表误伤）。
+from tests.support.db import TEST_DB_URL  # noqa: F401

dev.sh

@@ -74,7 +74,12 @@ if [ "${1:-}" = "logs" ]; then
     exec tail -n "${LOG_TAIL_LINES:-200}" -F "$APP_LOG"
 fi
 
-# 本地回归测试：仅依赖 uv，不需要 MySQL / .env 配置。
+# 本地回归测试：现在跑在**真实 MySQL**上（不再用 SQLite）。
+# 原因：生产是 MySQL，SQLite 与它有方言差异 —— 曾经出现「ORDER BY ... NULLS FIRST」
+# 本地全绿、生产直接 1064 的事故（见 docs/自动化与资料收集规划.md 附录 C.2）。
+# 因此需要：① 先 ./dev.sh 起 docker compose（MySQL 映射到 127.0.0.1:3307）；
+#          ② 测试库默认 douyu_login_test，可用 TEST_MYSQL_DATABASE 覆盖。
+# 测试库与开发库同名时会被 tests/support/db.py 拒绝运行（防误清）。
 # --group dev 确保 pytest 等 dev 依赖已安装。
 if [ "${1:-}" = "test" ]; then
     shift

docs/自动化与资料收集规划.md

@@ -1296,7 +1296,12 @@ OSS_PREFIX=collect
   P1a 踩过一次：`.env` 加了 `AUTOMATION_ENABLED=true`，容器里查不到，调度器照旧不启动；
   早先 §15.1 写入的 `ORDER_SITE_*` 与 `OSS_*` 同样没生效。新增变量时三处都要动：
   `.env`（服务器实值）+ `docker-compose.yml`（透传）+ `.env.example`（留档）；
-- **注意方言差异：生产是 MySQL 8，测试是 SQLite**。SQLite 能吃下的写法 MySQL 未必认，行为测试也发现不了。
+- **测试现已跑在真实 MySQL 上**（2026-10-02 改造完成，不再用 SQLite）：SQLite 与 MySQL 的方言差异
+  曾让 `NULLS FIRST` 这类写法本地全绿、生产 1064，而 P1b 又踩到 SQLite 测不出的
+  REPEATABLE READ 快照问题（见下条）。现在 `tests/conftest.py` 指向独立的 `douyu_login_test`、
+  每用例前后清表，库名与开发库相同时直接拒绝运行（防误清）；`./dev.sh test` 需要 MySQL 已启动。
+  **代价**：跑测试前要先 `./dev.sh` 起库，耗时从约 3.6 秒增至约 13 秒；
+- **注意方言差异：生产是 MySQL 8**。SQLite 能吃下的写法 MySQL 未必认，行为测试也发现不了。
   P1a 踩过一次：`ORDER BY ck_checked_at ASC NULLS FIRST` 在 SQLite 正常，MySQL 直接报
   `1064 ... near 'NULLS FIRST'`。跨方言的排序用可移植写法（如 `col IS NULL DESC, col ASC`），
   并在测试里按 MySQL 方言 `compile()` 一次做回归。**P0 更要注意：它连的是 order_site 的 PostgreSQL**，

tests/conftest.py

@@ -0,0 +1,41 @@
+"""全局测试配置与 MySQL 测试环境初始化。
+
+数据库连接串由根目录 ``conftest.py`` 在导入应用模块之前设好（见那里的说明），
+因此这里可以正常按顺序导入，不需要靠 import 先后保证正确性。
+"""
+
+import pytest
+
+import web.backend.models  # noqa: F401 确保所有 ORM 模型注册到 Base.metadata
+from tests.support.db import (
+    clear_tables,
+    get_test_engine,
+    init_test_schema,
+    make_test_session,
+)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def setup_test_database():
+    """测试会话级别：在测试库中建立干净一致的全新数据库结构。"""
+    engine = get_test_engine()
+    init_test_schema(engine)
+    yield engine
+
+
+@pytest.fixture(autouse=True)
+def clean_tables_per_test():
+    """用例级别：每个测试运行前后清空表数据，保证用例完全隔离且无残余状态。"""
+    clear_tables()
+    yield
+    clear_tables()
+
+
+@pytest.fixture
+def db():
+    """提供一个带自动关闭的测试 Session fixture。"""
+    session = make_test_session()
+    try:
+        yield session
+    finally:
+        session.close()

tests/support/__init__.py

@@ -0,0 +1 @@
+"""测试支持工具模块"""

tests/support/db.py

@@ -0,0 +1,152 @@
+"""测试专用的 MySQL 数据库引擎与会话工具。"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import quote_plus
+
+from sqlalchemy import Engine, create_engine, text
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import Session, sessionmaker
+
+
+def get_mysql_credentials() -> dict[str, str]:
+    """获取连接 MySQL 所需的凭据（优先环境变量，其次项目 .env）。"""
+    root_dir = Path(__file__).resolve().parents[2]
+    env_file = root_dir / ".env"
+    env_vars: dict[str, str] = {}
+    if env_file.exists():
+        with open(env_file, encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if line and not line.startswith("#") and "=" in line:
+                    k, v = line.split("=", 1)
+                    env_vars[k.strip()] = v.strip()
+
+    user = os.getenv("MYSQL_USER") or env_vars.get("MYSQL_USER") or "douyu_login"
+    pwd = os.getenv("MYSQL_PASSWORD") or env_vars.get("MYSQL_PASSWORD") or ""
+    host = (
+        os.getenv("MYSQL_BIND_HOST")
+        or env_vars.get("MYSQL_BIND_HOST")
+        or os.getenv("DEV_DB_HOST")
+        or "127.0.0.1"
+    )
+    port = (
+        os.getenv("MYSQL_HOST_PORT")
+        or env_vars.get("MYSQL_HOST_PORT")
+        or os.getenv("DEV_DB_PORT")
+        or "3307"
+    )
+    db_name = (
+        os.getenv("TEST_MYSQL_DATABASE")
+        or env_vars.get("TEST_MYSQL_DATABASE")
+        or "douyu_login_test"
+    )
+    assert_is_test_database(db_name)
+    return {
+        "user": user,
+        "password": pwd,
+        "host": host,
+        "port": port,
+        "database": db_name,
+    }
+
+
+def assert_is_test_database(db_name: str) -> None:
+    """拒绝把测试指向开发库。
+
+    ``clear_tables()`` 会 DELETE 掉所有业务表；一旦 TEST_MYSQL_DATABASE 被指到
+    ``douyu_login``（开发库）甚至生产库，就是数据灾难。这里做一道硬拦截：
+    库名与 .env 里的 MYSQL_DATABASE 相同就直接拒绝。
+    """
+    root_dir = Path(__file__).resolve().parents[2]
+    dev_db = ""
+    env_file = root_dir / ".env"
+    if env_file.exists():
+        with open(env_file, encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if line.startswith("MYSQL_DATABASE="):
+                    dev_db = line.split("=", 1)[1].strip()
+                    break
+    dev_db = os.getenv("MYSQL_DATABASE") or dev_db
+    if dev_db and db_name == dev_db:
+        raise RuntimeError(
+            f"测试库名 ({db_name}) 与开发库同名，拒绝运行："
+            "测试会清空所有表。请改用独立的测试库（默认 douyu_login_test）。"
+        )
+
+
+def get_test_database_url() -> str:
+    """返回测试数据库的完整 URL。"""
+    creds = get_mysql_credentials()
+    return (
+        f"mysql+pymysql://{quote_plus(creds['user'])}:{quote_plus(creds['password'])}"
+        f"@{creds['host']}:{creds['port']}/{creds['database']}?charset=utf8mb4"
+    )
+
+
+# 1. 确保环境变量始终指向测试库
+TEST_DB_URL = get_test_database_url()
+os.environ["DATABASE_URL"] = TEST_DB_URL
+os.environ["DB_NAME"] = get_mysql_credentials()["database"]
+
+import web.backend.database as database_module
+from web.backend.database import Base
+
+# 2. 覆盖 database_module 中的变量，保证即使被提前 import 过也能指向 MySQL 测试库
+database_module.DATABASE_URL = TEST_DB_URL
+
+_test_engine: Engine | None = None
+
+
+def get_test_engine() -> Engine:
+    """获取测试库共享单例 Engine。"""
+    global _test_engine
+    if _test_engine is None:
+        _test_engine = create_engine(
+            TEST_DB_URL,
+            pool_pre_ping=True,
+            pool_size=10,
+            max_overflow=20,
+            pool_recycle=1800,
+        )
+        database_module.engine = _test_engine
+        database_module.SessionLocal.configure(bind=_test_engine)
+    return _test_engine
+
+
+def make_test_session(engine: Engine | None = None) -> Session:
+    """为测试创建一个独立的 Session。"""
+    eng = engine or get_test_engine()
+    return sessionmaker(bind=eng)()
+
+
+def clear_tables(engine: Engine | None = None) -> None:
+    """快速清空当前 Base.metadata 中的所有表数据（耗时约 0.01s）。"""
+    eng = engine or get_test_engine()
+    tables = [t.name for t in Base.metadata.sorted_tables]
+    with eng.begin() as conn:
+        conn.execute(text("SET FOREIGN_KEY_CHECKS = 0;"))
+        for table in reversed(tables):
+            conn.execute(text(f"DELETE FROM `{table}`;"))
+        conn.execute(text("SET FOREIGN_KEY_CHECKS = 1;"))
+
+
+def init_test_schema(engine: Engine | None = None) -> None:
+    """确保测试库中所有表结构存在并清空所有数据（幂等高效）。"""
+    eng = engine or get_test_engine()
+    try:
+        Base.metadata.create_all(eng)
+        clear_tables(eng)
+    except OperationalError as exc:
+        # 测试不再使用 SQLite，连不上库时必须给出可操作的提示，
+        # 而不是抛一段 SQLAlchemy 原始堆栈。
+        raise RuntimeError(
+            "连接测试用 MySQL 失败，测试无法运行。\n"
+            f"  连接串: {TEST_DB_URL}\n"
+            "  请先启动数据库：./dev.sh（会起 docker compose，MySQL 映射到 127.0.0.1:3307）\n"
+            "  或用 TEST_MYSQL_DATABASE / MYSQL_USER / MYSQL_PASSWORD / MYSQL_HOST_PORT "
+            "指向一个可用的 MySQL 测试库。"
+        ) from exc

tests/test_account_sensitive_fields.py

@@ -1,9 +1,7 @@
 import pytest
 from fastapi import HTTPException
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, User
 from web.backend.routers.accounts import list_accounts, set_account_tag
 from web.backend.schemas import AccountTag
@@ -11,9 +9,8 @@ from web.backend.schemas import AccountTag
 
 class TestAccountSensitiveFields:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.admin = User(username="admin", password_hash="hash", role="super_admin")
         self.session.add(self.admin)
         self.session.add(
@@ -28,8 +25,6 @@ class TestAccountSensitiveFields:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_sensitive_fields_are_hidden_by_default_even_for_admin(self):
         result = list_accounts(

tests/test_audit_logs.py

@@ -4,10 +4,8 @@ import json
 
 import pytest
 from fastapi import HTTPException
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, AuditLog, DouyuTask, User
 from web.backend.routers.audit import list_audit_logs, require_super_admin
 from web.backend.services.audit_service import record_audit
@@ -15,9 +13,8 @@ from web.backend.services.audit_service import record_audit
 
 class TestAuditLog:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.admin = User(username="admin", password_hash="hash", role="super_admin")
         self.operator = User(
             username="operator",
@@ -30,8 +27,6 @@ class TestAuditLog:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_only_super_admin_can_read_audit_logs(self):
         with pytest.raises(HTTPException) as context:

tests/test_automation_ck_health.py

@@ -14,10 +14,8 @@ from datetime import UTC, datetime, timedelta
 
 import pytest
 from fastapi import HTTPException
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.deps import require_permission
 from web.backend.models import Account, AutomationJob, AutomationRun, LoginTask, User
 from web.backend.services.automation import (
@@ -33,9 +31,8 @@ from web.backend.services.automation.jobs import ck_health
 
 class _Base:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.admin = User(
             username="admin", password_hash="h", role="super_admin", is_active=True
         )

tests/test_automation_ck_relogin.py

@@ -413,3 +413,58 @@ class TestLinkageWithHealthJob(_Base):
         relogin = next(j for j in payload["jobs"] if j["key"] == ck_relogin.JOB_KEY)
         assert relogin["batch_size"] == 10
         assert payload["pending_relogin"] == 25  # 不被 limit 10 截断
+
+
+class TestSettlementVisibility(_Base):
+    """回归：结算必须用**新事务**读回 runner 提交的结果。
+
+    MySQL 默认隔离级别是 REPEATABLE READ：在同一个事务里重新查询只会看到事务
+    开始时的快照。第一版实现复用调度器传进来的 session 读结果，而那 10 条重登
+    **实际全部成功**，结算却按失败写回（加了退避、2 条还被标成极验进人工队列）。
+
+    SQLite 的读一致性不同，这条在 SQLite 上复现不出来 —— 属于「测试跑在 SQLite
+    就漏掉」的典型，现在测试跑在 MySQL 上才守得住。
+    """
+
+    def test_settlement_sees_other_session_commit(self):
+        task = self.make_task("acc", attempt_status="pending")
+
+        # 调用方会话先开事务（模拟调度器挑完候选后的状态）。
+        stale = make_test_session(self.engine)
+        assert (
+            stale.query(LoginTask).filter(LoginTask.id == task.id).one().attempt_status
+            == "pending"
+        )
+
+        # runner 在另一个会话里写入成功并提交。
+        runner = make_test_session(self.engine)
+        runner.query(LoginTask).filter(
+            LoginTask.id == task.id
+        ).one().attempt_status = "success"
+        runner.commit()
+        runner.close()
+
+        # 旧实现在 stale 那个事务里读，只会看到 pending —— 于是把成功写成失败。
+        assert (
+            stale.query(LoginTask).filter(LoginTask.id == task.id).one().attempt_status
+            == "pending"
+        ), "REPEATABLE READ 下同一事务读不到别人的提交，这正是当初踩的坑"
+
+        counters = ck_relogin.settle_fresh_results(
+            [task.id], now=datetime.now(UTC), max_attempts=2
+        )
+        self.session.expire_all()
+        assert counters["valid"] == 1 and counters["invalid"] == 0
+        assert self.session.get(LoginTask, task.id).relogin_failure_kind == ""
+        stale.close()
+
+    def test_settlement_marks_real_success_as_valid(self):
+        """结算看到 success 就必须清零退避，不能写成失败。"""
+        task = self.make_task("acc", attempt_status="success", relogin_attempts=1)
+        now = datetime.now(UTC)
+        counters = ck_relogin.settle_results(
+            {task.id: task}, [task.id], now=now, max_attempts=2
+        )
+        assert counters["valid"] == 1 and counters["invalid"] == 0
+        assert task.relogin_failure_kind == ""
+        assert task.relogin_next_at is None

tests/test_cookie_custom_order.py

@@ -1,18 +1,14 @@
 from datetime import UTC, datetime, timedelta
 
-from sqlalchemy import create_engine
+from tests.support.db import get_test_engine, make_test_session
-from sqlalchemy.orm import Session
-
-from web.backend.database import Base
 from web.backend.models import Account, LoginTask, User
 from web.backend.routers.cookies import _order_cookie_tasks
 
 
 class TestCustomCookieOrder:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.db = make_test_session(self.engine)
-        self.db = Session(self.engine)
         self.user = User(username="admin", password_hash="hash", role="super_admin")
         self.db.add(self.user)
         self.db.flush()
@@ -39,8 +35,6 @@ class TestCustomCookieOrder:
 
     def teardown_method(self):
         self.db.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_custom_cookie_order_follows_input_not_finished_time(self):
         selected_names = ["account-c", "account-a", "account-b"]

tests/test_cookie_operations.py

@@ -3,10 +3,8 @@ from typing import cast
 from unittest.mock import patch
 
 import pytest
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, LoginTask, ProxyConfig, User
 from web.backend.routers.cookies import (
     check_cookie_operations,
@@ -26,9 +24,8 @@ from web.backend.services.login_service import (
 
 class TestCookieOperation:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
 
         self.support = User(username="support", password_hash="hash", role="support")
         self.other_support = User(
@@ -81,8 +78,6 @@ class TestCookieOperation:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_support_operation_list_is_scoped_and_never_returns_credentials(self):
         result = list_cookie_operations(

tests/test_douyu_account_name_filter.py

@@ -1,7 +1,4 @@
-from sqlalchemy import create_engine
+from tests.support.db import get_test_engine, make_test_session
-from sqlalchemy.orm import sessionmaker
-
-from web.backend.database import Base
 from web.backend.models import Account, LoginTask, User
 from web.backend.routers.douyu import (
     _parse_account_names,
@@ -12,9 +9,8 @@ from web.backend.routers.douyu import (
 
 class TestDouyuAccountNameFilter:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.user = User(username="admin", password_hash="hash", role="super_admin")
         self.session.add(self.user)
         self.session.commit()
@@ -43,8 +39,6 @@ class TestDouyuAccountNameFilter:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_parse_account_names_trims_and_dedupes(self):
         assert _parse_account_names(

tests/test_douyu_cookie_visibility.py

@@ -8,10 +8,8 @@ Cookie，却让账号从工作台与 CK 列表里凭空消失。
 """
 
 import pytest
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, LoginTask, User
 from web.backend.routers.cookies import list_cookies
 from web.backend.routers.douyu import list_task_account_ids, list_task_accounts
@@ -25,9 +23,8 @@ from web.backend.services.douyu_service import (
 
 class TestDouyuCookieVisibility:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
 
         self.admin = User(username="admin", password_hash="hash", role="super_admin")
         self.support = User(username="kf_7", password_hash="hash", role="support")
@@ -84,8 +81,6 @@ class TestDouyuCookieVisibility:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def _task_account_names(self, current: User) -> set[str]:
         result = list_task_accounts(

tests/test_douyu_elite_open_flow.py

@@ -9,10 +9,8 @@
 from unittest.mock import Mock
 
 import pytest
-from sqlalchemy import create_engine, event
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, DouyuTask, User
 from web.backend.services import douyu_runner_manual
 from web.backend.services.douyu_runner import DouyuBatchRunner
@@ -24,14 +22,8 @@ from web.backend.services.douyu_service import (
 
 class TestEliteHandbookOpenFlow:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-
+        self.session = make_test_session(self.engine)
-        @event.listens_for(self.engine, "connect")
-        def enable_foreign_keys(connection, _):
-            connection.execute("PRAGMA foreign_keys=ON")
-
-        Base.metadata.create_all(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.user = User(username="operator", password_hash="hash", role="super_admin")
         self.account = Account(
             username="dyuser",
@@ -44,8 +36,6 @@ class TestEliteHandbookOpenFlow:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def _runner(self, task_type: str) -> DouyuBatchRunner:
         return DouyuBatchRunner(db=self.session, batch_id="batch1", task_type=task_type)

tests/test_douyu_gold_recharge_channel.py

@@ -4,11 +4,8 @@ import asyncio
 import os
 from unittest.mock import ANY, AsyncMock, Mock, patch
 
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-
 from core.douyu import FishFinRechargeClient, FishFinRechargeConfig
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, AuditLog, DouyuTask, LoginTask, User
 from web.backend.routers.douyu import get_recharge_channel, supplier_recharge_callback
 from web.backend.services.audit_service import record_audit
@@ -18,9 +15,8 @@ from web.backend.services.douyu_service import ensure_douyu_config
 
 class TestDouyuGoldRechargeChannel:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         user = User(username="operator", password_hash="hash", role="super_admin")
         self.session.add(user)
         self.session.commit()
@@ -51,8 +47,6 @@ class TestDouyuGoldRechargeChannel:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     @patch("web.backend.services.douyu_runner_gold.FishFinRechargeClient")
     def test_supplier_channel_creates_order_using_nickname_and_finishes_on_success(
@@ -224,7 +218,7 @@ class TestDouyuGoldRechargeChannel:
         self.runner._execute_one(task_id, {}, 1)
 
         self.runner._execute_create_gold_qr.assert_not_called()
-        verify_session = sessionmaker(bind=self.engine)()
+        verify_session = make_test_session(self.engine)
         try:
             task = verify_session.get(DouyuTask, task_id)
             checked_login = verify_session.get(LoginTask, login_task_id)

tests/test_douyu_workbench_scopes.py

@@ -1,8 +1,6 @@
 import pytest
-from sqlalchemy import create_engine, event
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import (
     Account,
     DouyuTask,
@@ -23,14 +21,8 @@ from web.backend.services.douyu_service import create_douyu_planned_tasks
 
 class TestDouyuWorkbenchScope:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-
+        self.session = make_test_session(self.engine)
-        @event.listens_for(self.engine, "connect")
-        def enable_foreign_keys(connection, _):
-            connection.execute("PRAGMA foreign_keys=ON")
-
-        Base.metadata.create_all(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.user = User(username="operator", password_hash="hash", role="super_admin")
         self.session.add(self.user)
         self.session.commit()
@@ -55,8 +47,6 @@ class TestDouyuWorkbenchScope:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_workbench_accounts_are_per_user_and_scope(self):
         update_workbench_accounts(

tests/test_douyu_write_path_contracts.py

@@ -2,24 +2,15 @@
 
 from unittest.mock import Mock
 
-from sqlalchemy import create_engine, event
+from tests.support.db import get_test_engine, make_test_session
-from sqlalchemy.orm import sessionmaker
-
-from web.backend.database import Base
 from web.backend.models import Account, DouyuGoodsSnapshot, DouyuTask, User
 from web.backend.services.douyu_runner import DouyuBatchRunner
 
 
 class TestDouyuWritePathContracts:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-
+        self.session = make_test_session(self.engine)
-        @event.listens_for(self.engine, "connect")
-        def enable_foreign_keys(connection, _):
-            connection.execute("PRAGMA foreign_keys=ON")
-
-        Base.metadata.create_all(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.user = User(username="operator", password_hash="hash", role="super_admin")
         self.account = Account(
             username="dyuser",
@@ -40,8 +31,6 @@ class TestDouyuWritePathContracts:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def _runner(self, task_type: str) -> DouyuBatchRunner:
         return DouyuBatchRunner(db=self.session, batch_id="batch1", task_type=task_type)

tests/test_huya_app_login.py

@@ -6,8 +6,6 @@ import struct
 from unittest.mock import MagicMock, patch
 
 import pytest
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
 
 from core.huya import (
     HuyaAppLoginError,
@@ -31,7 +29,7 @@ from core.huya.nonce_forge import K1_DEFAULT, gen_nonce
 from core.huya.udb_aes import udb_decrypt, udb_encrypt
 from core.huya.wup_encoder import _make_name, build_password_login_wup
 from core.huya.wup_protocol import WupResponse
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import HuyaAccount, User
 from web.backend.routers.huya import (
     app_password_login_account,
@@ -210,17 +208,14 @@ def test_supplement_anonymous_device_failure_does_not_raise(monkeypatch):
 
 class TestHuyaAppLogin:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.admin = User(username="admin", password_hash="hash", role="super_admin")
         self.session.add(self.admin)
         self.session.commit()
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_udb_aes_roundtrip(self):
         key = b"4VYcPdvKKqjBHZtC"

tests/test_huya_browser_exchange.py

@@ -7,10 +7,8 @@ from pathlib import Path
 import pytest
 from fastapi import HTTPException
 from pydantic import ValidationError
-from sqlalchemy import create_engine, event
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.huya_defaults import HUYA_DEFAULT_ROOM_PID
 from web.backend.models import AuditLog, HuyaAccount, HuyaConfig, User
 from web.backend.routers.huya import create_browser_exchange_kit
@@ -27,14 +25,8 @@ EXTENSION_DIR = REPO_ROOT / "web" / "extension" / "huya-exchange-helper"
 
 class TestHuyaBrowserExchangeKit:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-
+        self.session = make_test_session(self.engine)
-        @event.listens_for(self.engine, "connect")
-        def enable_foreign_keys(connection, _):
-            connection.execute("PRAGMA foreign_keys=ON")
-
-        Base.metadata.create_all(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.user = User(username="operator", password_hash="hash", role="super_admin")
         self.account = HuyaAccount(
             uid="100", yyuid="100", cookie="yyuid=100; udb_uid=100"
@@ -44,8 +36,6 @@ class TestHuyaBrowserExchangeKit:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def _kit(self, **kwargs):
         return create_browser_exchange_kit(

tests/test_huya_cookie_rebuild.py

@@ -1,11 +1,9 @@
 """虎牙网页 Cookie 回写与 raw_cookie 重建（不发起登录或支付）。"""
 
 import pytest
-from sqlalchemy import create_engine, event
-from sqlalchemy.orm import sessionmaker
 
 from core.huya.web_cookie_fields import cookie_fields, missing_web_cookie_fields
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import HuyaAccount, User
 from web.backend.routers.huya.cookies import rebuild_web_cookie
 from web.backend.services.huya_service import (
@@ -20,14 +18,8 @@ RAW_COOKIE = (
 
 class TestHuyaCookieRebuild:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-
+        self.session = make_test_session(self.engine)
-        @event.listens_for(self.engine, "connect")
-        def enable_foreign_keys(connection, _):
-            connection.execute("PRAGMA foreign_keys=ON")
-
-        Base.metadata.create_all(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.user = User(username="operator", password_hash="hash", role="super_admin")
         self.account = HuyaAccount(
             uid="100",
@@ -41,8 +33,6 @@ class TestHuyaCookieRebuild:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_persist_writes_filled_session_cookie(self):
         filled = RAW_COOKIE + "; guid=0123456789abcdef0123456789abcdef"

tests/test_huya_selected_login_queue.py

@@ -2,12 +2,9 @@
 
 from unittest.mock import patch
 
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-
 from core.huya.app_login import HuyaAppQrAuthRequiredError
 from core.huya.login import HuyaCredentialError, HuyaLoginResult
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import HuyaAccount, User
 from web.backend.routers.huya.login import app_password_login_selected_accounts
 from web.backend.schemas import HuyaPasswordLoginSelectedRequest
@@ -35,17 +32,14 @@ def _fail_result(
 
 class TestSelectedAppLoginQueue:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.admin = User(username="admin", password_hash="hash", role="super_admin")
         self.session.add(self.admin)
         self.session.commit()
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def _add_account(self, username: str, password: str = "pw") -> HuyaAccount:
         account = HuyaAccount(

tests/test_huya_workbench_scopes.py

@@ -1,8 +1,6 @@
 import pytest
-from sqlalchemy import create_engine, event
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import HuyaAccount, HuyaWorkbench, HuyaWorkbenchAccount, User
 from web.backend.routers.huya import (
     delete_account,
@@ -17,14 +15,8 @@ from web.backend.services.huya_service import create_planned_tasks
 
 class TestHuyaWorkbenchScope:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-
+        self.session = make_test_session(self.engine)
-        @event.listens_for(self.engine, "connect")
-        def enable_foreign_keys(connection, _):
-            connection.execute("PRAGMA foreign_keys=ON")
-
-        Base.metadata.create_all(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.user = User(username="operator", password_hash="hash", role="super_admin")
         self.account = HuyaAccount(
             uid="100", yyuid="100", cookie="yyuid=100; udb_uid=100"
@@ -34,8 +26,6 @@ class TestHuyaWorkbenchScope:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_accounts_and_tasks_are_scoped(self):
         update_huya_workbench_accounts(

tests/test_login_attempt_isolation.py

@@ -6,15 +6,12 @@
 更新，所以停止、失败、进程重启都不可能破坏账号已有的 CK。
 """
 
-import tempfile
-from pathlib import Path
 from types import SimpleNamespace
 
-from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
 import web.backend.database as database_module
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, LoginTask, User
 from web.backend.routers.login import list_tasks
 from web.backend.services import login_service
@@ -47,13 +44,9 @@ class _FakeLogin:
 
 class TestLoginAttemptIsolation:
     def setup_method(self):
-        self.tmpdir = tempfile.TemporaryDirectory()
+        self.engine = get_test_engine()
-        url = f"sqlite:///{Path(self.tmpdir.name) / 'attempt.db'}"
-        self.engine = create_engine(url, connect_args={"check_same_thread": False})
-        Base.metadata.create_all(self.engine)
-        # 批次执行器在后台线程里用 SessionLocal，测试期间指向同一个文件库。
         database_module.SessionLocal.configure(bind=self.engine)
-        self.session = sessionmaker(bind=self.engine)()
+        self.session = make_test_session(self.engine)
 
         self.user = User(username="admin", password_hash="hash", role="super_admin")
         self.session.add(self.user)
@@ -84,11 +77,7 @@ class TestLoginAttemptIsolation:
         _FakeLogin.calls = 0
 
     def teardown_method(self):
-        database_module.SessionLocal.configure(bind=database_module.engine)
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
-        self.tmpdir.cleanup()
 
     def _run_batch(self, monkeypatch, *, mode="login", account_ids=None, stop=False):
         monkeypatch.setattr(login_service, "DouyuLogin", _FakeLogin)

tests/test_login_attempt_migration.py

@@ -1,4 +1,4 @@
-"""迁移 20261001_0035 的存量数据归一验证。
+"""迁移 20261001_0035 的存量数据归一验证（在真实 MySQL 临时库上执行）。
 
 线上升级时要保证：
 - 持有 CK 的记录 status 归一到 success，历史批次状态搬进 attempt_*；
@@ -6,13 +6,17 @@
 - 任何情况下都不丢 CK。
 """
 
-import tempfile
+import secrets
 from pathlib import Path
+from urllib.parse import quote_plus
 
 from alembic import command
 from alembic.config import Config
 from sqlalchemy import create_engine, text
 
+import web.backend.database as database_module
+from tests.support.db import get_mysql_credentials, get_test_engine
+
 PROJECT_ROOT = Path(__file__).resolve().parents[1]
 
 
@@ -43,12 +47,22 @@ def _insert_task(conn, *, status: str, cookie: str, message: str = "") -> None:
 
 class TestLoginAttemptMigration:
     def setup_method(self):
-        import web.backend.database as database_module
+        creds = get_mysql_credentials()
+        self.mig_db = f"douyu_login_mig_{secrets.token_hex(4)}"
+        self.url = (
+            f"mysql+pymysql://{quote_plus(creds['user'])}:{quote_plus(creds['password'])}"
+            f"@{creds['host']}:{creds['port']}/{self.mig_db}?charset=utf8mb4"
+        )
+        self.admin_engine = get_test_engine()
+        with self.admin_engine.begin() as conn:
+            conn.execute(
+                text(
+                    f"CREATE DATABASE `{self.mig_db}` CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;"
+                )
+            )
 
         self.database_module = database_module
         self.original_url = database_module.DATABASE_URL
-        self.tmpdir = tempfile.TemporaryDirectory()
-        self.url = f"sqlite:///{Path(self.tmpdir.name) / 'upgrade.db'}"
         # env.py 用模块级 DATABASE_URL 覆盖配置里的连接串。
         database_module.DATABASE_URL = self.url
         self.config = _migration_config(self.url)
@@ -91,9 +105,10 @@ class TestLoginAttemptMigration:
             _insert_task(conn, status="failed", cookie="ck-failed", message="登录失败")
 
     def teardown_method(self):
-        self.database_module.DATABASE_URL = self.original_url
         self.engine.dispose()
-        self.tmpdir.cleanup()
+        self.database_module.DATABASE_URL = self.original_url
+        with self.admin_engine.begin() as conn:
+            conn.execute(text(f"DROP DATABASE IF EXISTS `{self.mig_db}`;"))
 
     def _rows(self) -> list[dict]:
         with self.engine.connect() as conn:

tests/test_migrations.py

@@ -1,22 +1,26 @@
-"""Alembic 迁移链冒烟测试（不依赖真实 MySQL）。
+"""Alembic 迁移链冒烟测试（在真实 MySQL 临时库上执行）。
 
 覆盖三件事：
 1. 链完整性：只有一个头版本、头与最新迁移一致、每个修订的
    down_revision 都能回溯到根（无悬挂引用、无分叉链）。
 2. 全部迁移脚本可编译：防止迁移文件被改坏后部署流程才暴露。
-3. 空库整链执行：迁移脚本自带方言防护（MySQL 专属 DDL 有 _is_mysql()
+3. 空库整链执行：迁移脚本在真实 MySQL 临时库上执行 `upgrade head`，
-   之类分支），整条链可在 SQLite 上真实执行 `upgrade head`，随后校验
+   随后校验表/列/索引与 Base.metadata 模型元数据对齐（模型要求的不允许缺失），
-   表/列/索引与 Base.metadata 模型元数据对齐（模型要求的不允许缺失），
    并确认迁移新增的复合查询索引确实落库。
 """
 
-import tempfile
+import secrets
 from pathlib import Path
+from urllib.parse import quote_plus
 
 from alembic import command
 from alembic.config import Config
 from alembic.script import ScriptDirectory
-from sqlalchemy import create_engine, inspect
+from sqlalchemy import create_engine, inspect, text
+
+import web.backend.database as database_module
+from tests.support.db import get_mysql_credentials, get_test_engine
+from web.backend.database import Base
 
 TESTS_DIR = Path(__file__).resolve().parent
 PROJECT_ROOT = TESTS_DIR.parent
@@ -66,52 +70,56 @@ class TestMigrationSmoke:
             source = path.read_text(encoding="utf-8")
             compile(source, str(path), "exec")  # 语法/缩进等编译错误会在此抛出
 
-    def test_full_chain_applies_on_fresh_sqlite_and_matches_models(self):
+    def test_full_chain_applies_on_fresh_mysql_and_matches_models(self):
         """空库执行整条迁移链，校验表/列/索引与模型元数据对齐。"""
-        import web.backend.database as database_module
+        creds = get_mysql_credentials()
-        from web.backend.database import Base
+        mig_db = f"douyu_login_mig_{secrets.token_hex(4)}"
+        mig_url = f"mysql+pymysql://{quote_plus(creds['user'])}:{quote_plus(creds['password'])}@{creds['host']}:{creds['port']}/{mig_db}?charset=utf8mb4"
+
+        admin_engine = get_test_engine()
+        with admin_engine.begin() as conn:
+            conn.execute(
+                text(
+                    f"CREATE DATABASE `{mig_db}` CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;"
+                )
+            )
 
         original_url = database_module.DATABASE_URL
-        with tempfile.TemporaryDirectory() as tmp:
+        try:
-            db_path = Path(tmp) / "migrate.db"
+            database_module.DATABASE_URL = mig_url
-            try:
+            cfg = _migration_config()
-                # env.py 会用模块级 DATABASE_URL 覆盖配置里的 sqlalchemy.url。
+            cfg.set_main_option("sqlalchemy.url", mig_url)
-                database_module.DATABASE_URL = f"sqlite:///{db_path}"
+            command.upgrade(cfg, "head")
-                command.upgrade(_migration_config(), "head")
 
-                inspector = inspect(create_engine(f"sqlite:///{db_path}"))
+            inspector = inspect(create_engine(mig_url))
-                db_tables = set(inspector.get_table_names())
+            db_tables = set(inspector.get_table_names())
-                assert db_tables
+            assert db_tables
 
-                model_tables = Base.metadata.tables
+            model_tables = Base.metadata.tables
-                missing_tables = set(model_tables) - db_tables
+            missing_tables = set(model_tables) - db_tables
-                assert not missing_tables
+            assert not missing_tables
 
-                all_db_indexes = set()
+            all_db_indexes = set()
-                for name, table in model_tables.items():
+            for name, table in model_tables.items():
-                    db_columns = {
+                db_columns = {column["name"] for column in inspector.get_columns(name)}
-                        column["name"] for column in inspector.get_columns(name)
+                model_columns = set(table.columns.keys())
-                    }
+                missing_columns = model_columns - db_columns
-                    model_columns = set(table.columns.keys())
+                assert not missing_columns
-                    missing_columns = model_columns - db_columns
-                    assert not missing_columns
 
-                    db_indexes = set()
+                db_indexes = set()
-                    for index in inspector.get_indexes(name):
+                for index in inspector.get_indexes(name):
-                        index_name = index["name"]
+                    index_name = index["name"]
-                        if index_name and not index_name.startswith("sqlite_autoindex"):
+                    if index_name:
-                            db_indexes.add(index_name)
+                        db_indexes.add(index_name)
-                    all_db_indexes |= db_indexes
+                all_db_indexes |= db_indexes
-                    model_indexes = {
+                model_indexes = {index.name for index in table.indexes if index.name}
-                        index.name for index in table.indexes if index.name
+                missing_indexes = model_indexes - db_indexes
-                    }
+                assert not missing_indexes
-                    missing_indexes = model_indexes - db_indexes
-                    assert not missing_indexes
 
-                # 迁移新增的复合查询索引（不入模型，但应存在）。
+            # 迁移新增的复合查询索引（不入模型，但应存在）。
-                for index_name in EXTRA_INDEXES:
+            for index_name in EXTRA_INDEXES:
-                    assert index_name in all_db_indexes, (
+                assert index_name in all_db_indexes, f"迁移链未创建索引 {index_name}"
-                        f"迁移链未创建索引 {index_name}"
+        finally:
-                    )
+            database_module.DATABASE_URL = original_url
-            finally:
+            with admin_engine.begin() as conn:
-                database_module.DATABASE_URL = original_url
+                conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))

tests/test_user_deletion.py

@@ -1,8 +1,6 @@
 import pytest
-from sqlalchemy import create_engine, event
-from sqlalchemy.orm import sessionmaker
 
-from web.backend.database import Base
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, AuditLog, LoginTask, User
 from web.backend.routers.users import delete_user, list_users, rename_user
 from web.backend.schemas import UserRename
@@ -10,14 +8,8 @@ from web.backend.schemas import UserRename
 
 class TestUserDeletion:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-
+        self.session = make_test_session(self.engine)
-        @event.listens_for(self.engine, "connect")
-        def enable_foreign_keys(connection, _):
-            connection.execute("PRAGMA foreign_keys=ON")
-
-        Base.metadata.create_all(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
         self.admin = User(username="admin", password_hash="hash", role="super_admin")
         self.user = User(username="support", password_hash="hash", role="support")
         self.session.add_all([self.admin, self.user])
@@ -44,8 +36,6 @@ class TestUserDeletion:
 
     def teardown_method(self):
         self.session.close()
-        Base.metadata.drop_all(self.engine)
-        self.engine.dispose()
 
     def test_delete_user_soft_deletes_and_releases_assigned_accounts(self):
         result = delete_user(

tests/test_utc_datetime.py

@@ -14,18 +14,17 @@
 
 from datetime import UTC, datetime, timedelta, timezone
 
-from sqlalchemy import create_engine, text
+from sqlalchemy import text
-from sqlalchemy.orm import sessionmaker
 
+from tests.support.db import get_test_engine, make_test_session
 from web.backend.database import Base, UTCDateTime
-from web.backend.models import AutomationRun, LoginTask
+from web.backend.models import Account, AutomationRun, LoginTask, User
 
 
 class TestUtcDateTime:
     def setup_method(self):
-        self.engine = create_engine("sqlite://")
+        self.engine = get_test_engine()
-        Base.metadata.create_all(self.engine)
+        self.session = make_test_session(self.engine)
-        self.session = sessionmaker(bind=self.engine)()
 
     def teardown_method(self):
         self.session.close()
@@ -80,10 +79,14 @@ class TestUtcDateTime:
 
     def test_login_task_attempt_at_round_trips_as_utc(self):
         """真实模型抽查：写 aware、读 aware、能直接比较。"""
+        user = User(username="u", password_hash="p", role="admin")
+        account = Account(username="a", password="p", email="e", email_password="ep")
+        self.session.add_all([user, account])
+        self.session.commit()
         task = LoginTask(
             batch_id="b",
-            account_id=1,
+            account_id=account.id,
-            created_by=1,
+            created_by=user.id,
             attempt_at=datetime(2026, 10, 2, 12, 0, tzinfo=UTC),
         )
         self.session.add(task)