From c3fd299f3aec2be2fe4b17a351815e0bc803f6df Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 2 Oct 2026 21:22:08 +0800 Subject: [PATCH] =?UTF-8?q?test(db):=20=E6=B5=8B=E8=AF=95=E6=94=B9?= =?UTF-8?q?=E7=94=A8=E7=9C=9F=E5=AE=9E=20MySQL=EF=BC=8C=E6=96=B9=E8=A8=80?= =?UTF-8?q?=E5=B7=AE=E5=BC=82=E4=B8=8D=E5=86=8D=E6=BC=8F=E5=88=B0=E7=94=9F?= =?UTF-8?q?=E4=BA=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 生产是 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 | 16 ++- conftest.py | 18 ++- dev.sh | 7 +- docs/自动化与资料收集规划.md | 7 +- tests/conftest.py | 41 ++++++ tests/support/__init__.py | 1 + tests/support/db.py | 152 ++++++++++++++++++++++ tests/test_account_sensitive_fields.py | 11 +- tests/test_audit_logs.py | 11 +- tests/test_automation_ck_health.py | 9 +- tests/test_automation_ck_relogin.py | 55 ++++++++ tests/test_cookie_custom_order.py | 12 +- tests/test_cookie_operations.py | 11 +- tests/test_douyu_account_name_filter.py | 12 +- tests/test_douyu_cookie_visibility.py | 11 +- tests/test_douyu_elite_open_flow.py | 16 +-- tests/test_douyu_gold_recharge_channel.py | 14 +- tests/test_douyu_workbench_scopes.py | 16 +-- tests/test_douyu_write_path_contracts.py | 17 +-- tests/test_huya_app_login.py | 11 +- tests/test_huya_browser_exchange.py | 16 +-- tests/test_huya_cookie_rebuild.py | 16 +-- tests/test_huya_selected_login_queue.py | 12 +- tests/test_huya_workbench_scopes.py | 16 +-- tests/test_login_attempt_isolation.py | 17 +-- tests/test_login_attempt_migration.py | 29 ++++- tests/test_migrations.py | 102 ++++++++------- tests/test_user_deletion.py | 16 +-- tests/test_utc_datetime.py | 19 +-- 29 files changed, 429 insertions(+), 262 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/support/__init__.py create mode 100644 tests/support/db.py diff --git a/README.md b/README.md index 51739b4..715c19b 100644 --- a/README.md +++ b/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 diff --git a/conftest.py b/conftest.py index c594f20..2698dbc 100644 --- a/conftest.py +++ b/conftest.py @@ -1,14 +1,20 @@ -"""pytest 全局引导：在导入任何应用模块前固定测试数据库与加密密钥。 +"""pytest 全局引导：在导入任何应用模块前定好测试库与加密密钥。 -web/backend 的模块在 import 阶段就会读取 DATABASE_URL 创建全局引擎、 -读取 APP_ENCRYPTION_KEY 初始化敏感字段加密，因此这两个环境变量必须在 -测试模块加载之前设置好。各测试文件自身使用独立的 SQLite 内存引擎做 -数据隔离，这里统一兜底，避免误连开发/生产数据库。 +这个文件在 pytest 收集阶段**最先执行**，所以「让测试连上测试库」这件事放在这里 +最稳 —— 不依赖任何测试文件或 tests/conftest.py 的 import 顺序。 + +背景：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 diff --git a/dev.sh b/dev.sh index 0f8da1e..4ad60e4 100755 --- a/dev.sh +++ b/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 diff --git a/docs/自动化与资料收集规划.md b/docs/自动化与资料收集规划.md index ba3da1a..3998c55 100644 --- a/docs/自动化与资料收集规划.md +++ b/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**， diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..ea583c8 --- /dev/null +++ b/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() diff --git a/tests/support/__init__.py b/tests/support/__init__.py new file mode 100644 index 0000000..15254f6 --- /dev/null +++ b/tests/support/__init__.py @@ -0,0 +1 @@ +"""测试支持工具模块""" diff --git a/tests/support/db.py b/tests/support/db.py new file mode 100644 index 0000000..4c5bb78 --- /dev/null +++ b/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 diff --git a/tests/test_account_sensitive_fields.py b/tests/test_account_sensitive_fields.py index 4907c6d..0f30c68 100644 --- a/tests/test_account_sensitive_fields.py +++ b/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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_audit_logs.py b/tests/test_audit_logs.py index 8dc6ba6..567c372 100644 --- a/tests/test_audit_logs.py +++ b/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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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: diff --git a/tests/test_automation_ck_health.py b/tests/test_automation_ck_health.py index b4d83f9..dc58006 100644 --- a/tests/test_automation_ck_health.py +++ b/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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(self.engine) self.admin = User( username="admin", password_hash="h", role="super_admin", is_active=True ) diff --git a/tests/test_automation_ck_relogin.py b/tests/test_automation_ck_relogin.py index 4339027..f8e7977 100644 --- a/tests/test_automation_ck_relogin.py +++ b/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 diff --git a/tests/test_cookie_custom_order.py b/tests/test_cookie_custom_order.py index a531ffe..83ff36f 100644 --- a/tests/test_cookie_custom_order.py +++ b/tests/test_cookie_custom_order.py @@ -1,18 +1,14 @@ from datetime import UTC, datetime, timedelta -from sqlalchemy import create_engine -from sqlalchemy.orm import Session - -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 _order_cookie_tasks class TestCustomCookieOrder: def setup_method(self): - self.engine = create_engine("sqlite://") - Base.metadata.create_all(self.engine) - self.db = Session(self.engine) + self.engine = get_test_engine() + self.db = make_test_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"] diff --git a/tests/test_cookie_operations.py b/tests/test_cookie_operations.py index 9a443f4..69623db 100644 --- a/tests/test_cookie_operations.py +++ 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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_douyu_account_name_filter.py b/tests/test_douyu_account_name_filter.py index 8af88f0..1ea3d3d 100644 --- a/tests/test_douyu_account_name_filter.py +++ b/tests/test_douyu_account_name_filter.py @@ -1,7 +1,4 @@ -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.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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_douyu_cookie_visibility.py b/tests/test_douyu_cookie_visibility.py index 2022f1e..66918fa 100644 --- a/tests/test_douyu_cookie_visibility.py +++ b/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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_douyu_elite_open_flow.py b/tests/test_douyu_elite_open_flow.py index 291e7c6..3aca3fd 100644 --- a/tests/test_douyu_elite_open_flow.py +++ b/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://") - - @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.engine = get_test_engine() + self.session = make_test_session(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) diff --git a/tests/test_douyu_gold_recharge_channel.py b/tests/test_douyu_gold_recharge_channel.py index 1bb49e2..00785f2 100644 --- a/tests/test_douyu_gold_recharge_channel.py +++ b/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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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) diff --git a/tests/test_douyu_workbench_scopes.py b/tests/test_douyu_workbench_scopes.py index 118d966..02e9a93 100644 --- a/tests/test_douyu_workbench_scopes.py +++ b/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://") - - @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.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_douyu_write_path_contracts.py b/tests/test_douyu_write_path_contracts.py index ce8ed67..f8661e5 100644 --- a/tests/test_douyu_write_path_contracts.py +++ b/tests/test_douyu_write_path_contracts.py @@ -2,24 +2,15 @@ from unittest.mock import Mock -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, DouyuGoodsSnapshot, DouyuTask, User from web.backend.services.douyu_runner import DouyuBatchRunner class TestDouyuWritePathContracts: def setup_method(self): - self.engine = create_engine("sqlite://") - - @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.engine = get_test_engine() + self.session = make_test_session(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) diff --git a/tests/test_huya_app_login.py b/tests/test_huya_app_login.py index b9641e3..489624b 100644 --- a/tests/test_huya_app_login.py +++ b/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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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" diff --git a/tests/test_huya_browser_exchange.py b/tests/test_huya_browser_exchange.py index 62d9e78..754c569 100644 --- a/tests/test_huya_browser_exchange.py +++ b/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://") - - @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.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_huya_cookie_rebuild.py b/tests/test_huya_cookie_rebuild.py index 2ab56cc..60927e7 100644 --- a/tests/test_huya_cookie_rebuild.py +++ b/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://") - - @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.engine = get_test_engine() + self.session = make_test_session(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" diff --git a/tests/test_huya_selected_login_queue.py b/tests/test_huya_selected_login_queue.py index edf668a..12565c2 100644 --- a/tests/test_huya_selected_login_queue.py +++ b/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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_huya_workbench_scopes.py b/tests/test_huya_workbench_scopes.py index e96d77c..dbb129e 100644 --- a/tests/test_huya_workbench_scopes.py +++ b/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://") - - @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.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_login_attempt_isolation.py b/tests/test_login_attempt_isolation.py index 284d272..98f2aaa 100644 --- a/tests/test_login_attempt_isolation.py +++ b/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() - 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，测试期间指向同一个文件库。 + self.engine = get_test_engine() 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) diff --git a/tests/test_login_attempt_migration.py b/tests/test_login_attempt_migration.py index f078dcf..2ac17b4 100644 --- a/tests/test_login_attempt_migration.py +++ b/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: diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 2acd3d6..1f0826c 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -1,22 +1,26 @@ -"""Alembic 迁移链冒烟测试（不依赖真实 MySQL）。 +"""Alembic 迁移链冒烟测试（在真实 MySQL 临时库上执行）。 覆盖三件事： 1. 链完整性：只有一个头版本、头与最新迁移一致、每个修订的 down_revision 都能回溯到根（无悬挂引用、无分叉链）。 2. 全部迁移脚本可编译：防止迁移文件被改坏后部署流程才暴露。 -3. 空库整链执行：迁移脚本自带方言防护（MySQL 专属 DDL 有 _is_mysql() - 之类分支），整条链可在 SQLite 上真实执行 `upgrade head`，随后校验 - 表/列/索引与 Base.metadata 模型元数据对齐（模型要求的不允许缺失）， +3. 空库整链执行：迁移脚本在真实 MySQL 临时库上执行 `upgrade head`， + 随后校验表/列/索引与 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 - from web.backend.database import Base + creds = get_mysql_credentials() + 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: - db_path = Path(tmp) / "migrate.db" - try: - # env.py 会用模块级 DATABASE_URL 覆盖配置里的 sqlalchemy.url。 - database_module.DATABASE_URL = f"sqlite:///{db_path}" - command.upgrade(_migration_config(), "head") + try: + database_module.DATABASE_URL = mig_url + cfg = _migration_config() + cfg.set_main_option("sqlalchemy.url", mig_url) + command.upgrade(cfg, "head") - inspector = inspect(create_engine(f"sqlite:///{db_path}")) - db_tables = set(inspector.get_table_names()) - assert db_tables + inspector = inspect(create_engine(mig_url)) + db_tables = set(inspector.get_table_names()) + assert db_tables - model_tables = Base.metadata.tables - missing_tables = set(model_tables) - db_tables - assert not missing_tables + model_tables = Base.metadata.tables + missing_tables = set(model_tables) - db_tables + assert not missing_tables - all_db_indexes = set() - for name, table in model_tables.items(): - db_columns = { - column["name"] for column in inspector.get_columns(name) - } - model_columns = set(table.columns.keys()) - missing_columns = model_columns - db_columns - assert not missing_columns + all_db_indexes = set() + for name, table in model_tables.items(): + db_columns = {column["name"] for column in inspector.get_columns(name)} + model_columns = set(table.columns.keys()) + missing_columns = model_columns - db_columns + assert not missing_columns - db_indexes = set() - for index in inspector.get_indexes(name): - index_name = index["name"] - if index_name and not index_name.startswith("sqlite_autoindex"): - db_indexes.add(index_name) - all_db_indexes |= db_indexes - model_indexes = { - index.name for index in table.indexes if index.name - } - missing_indexes = model_indexes - db_indexes - assert not missing_indexes + db_indexes = set() + for index in inspector.get_indexes(name): + index_name = index["name"] + if index_name: + db_indexes.add(index_name) + all_db_indexes |= db_indexes + model_indexes = {index.name for index in table.indexes if index.name} + missing_indexes = model_indexes - db_indexes + assert not missing_indexes - # 迁移新增的复合查询索引（不入模型，但应存在）。 - for index_name in EXTRA_INDEXES: - assert index_name in all_db_indexes, ( - f"迁移链未创建索引 {index_name}" - ) - finally: - database_module.DATABASE_URL = original_url + # 迁移新增的复合查询索引（不入模型，但应存在）。 + for index_name in EXTRA_INDEXES: + assert index_name in all_db_indexes, f"迁移链未创建索引 {index_name}" + finally: + database_module.DATABASE_URL = original_url + with admin_engine.begin() as conn: + conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;")) diff --git a/tests/test_user_deletion.py b/tests/test_user_deletion.py index c7a7bd5..79e8431 100644 --- a/tests/test_user_deletion.py +++ b/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://") - - @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.engine = get_test_engine() + self.session = make_test_session(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( diff --git a/tests/test_utc_datetime.py b/tests/test_utc_datetime.py index d804dc1..bb2857a 100644 --- a/tests/test_utc_datetime.py +++ b/tests/test_utc_datetime.py @@ -14,18 +14,17 @@ from datetime import UTC, datetime, timedelta, timezone -from sqlalchemy import create_engine, text -from sqlalchemy.orm import sessionmaker +from sqlalchemy import text +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://") - Base.metadata.create_all(self.engine) - self.session = sessionmaker(bind=self.engine)() + self.engine = get_test_engine() + self.session = make_test_session(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, - created_by=1, + account_id=account.id, + created_by=user.id, attempt_at=datetime(2026, 10, 2, 12, 0, tzinfo=UTC), ) self.session.add(task)