yml2213
afa5c815f6
让运营能按订单号查到订单，作为后续资料收集（P2）与皮肤准入（P3/P4）的输入。 order_site 零改动：只读直连、独立极小连接池、可一键降级。 上机核实推翻了规划两处假设（详见 §6.5.1）： - orders 表没有 order_no 列，只有 id 与 platform_order_id → 查询入口用后者； - platform_order_id 非唯一：541 组重复，形态固定为「91kaquan 内部行(已发货) + 真实店铺行(无人管)」。按规划原文的单行判据，这 540 个已发货订单会被误判 成「没人管」而放行重复履约。改为**订单号维度聚合**：任一行已发货即 auto_fulfilled。 真机验证（只读账号 livehub_ro）：三桶判定全对，重复组正确拦下并回显真实 店铺行的中文名（狂沙舞者，而非内部码 suit_sand_dancer），单次查询 3.3ms。 顺带修一个只在真机才暴露的 bug：_get_engine 先置 _INITIALIZED 再读 URL， 一次落空就把模块永久毒化成不可用，之后补上连接串也救不回来。
126 lines
5.3 KiB
Python
126 lines
5.3 KiB
Python
"""Alembic 迁移链冒烟测试（在真实 MySQL 临时库上执行）。
|
||
|
||
覆盖三件事：
|
||
1. 链完整性：只有一个头版本、头与最新迁移一致、每个修订的
|
||
down_revision 都能回溯到根（无悬挂引用、无分叉链）。
|
||
2. 全部迁移脚本可编译：防止迁移文件被改坏后部署流程才暴露。
|
||
3. 空库整链执行：迁移脚本在真实 MySQL 临时库上执行 `upgrade head`，
|
||
随后校验表/列/索引与 Base.metadata 模型元数据对齐（模型要求的不允许缺失），
|
||
并确认迁移新增的复合查询索引确实落库。
|
||
"""
|
||
|
||
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, 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
|
||
VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
|
||
|
||
HEAD_REVISION = "20261002_0039"
|
||
|
||
# 迁移新增、但不声明在模型里的复合查询索引。
|
||
EXTRA_INDEXES = (
|
||
"ix_douyu_tasks_handbook_scope_id",
|
||
"ix_douyu_tasks_handbook_scope_task_type_id",
|
||
"ix_douyu_tasks_status_finished_at_id",
|
||
"ix_login_tasks_status_finished_at_id",
|
||
)
|
||
|
||
|
||
def _migration_config() -> Config:
|
||
config = Config(str(PROJECT_ROOT / "alembic.ini"))
|
||
config.set_main_option(
|
||
"script_location", str(PROJECT_ROOT / "web" / "backend" / "migrations")
|
||
)
|
||
return config
|
||
|
||
|
||
class TestMigrationSmoke:
|
||
def test_chain_is_linear_and_head_matches_latest_file(self):
|
||
directory = ScriptDirectory.from_config(_migration_config())
|
||
|
||
heads = directory.get_heads()
|
||
assert len(heads) == 1, f"迁移链出现多个头版本（存在分叉）: {heads}"
|
||
assert heads[0] == HEAD_REVISION, f"头版本 {heads[0]} 与最新迁移不一致"
|
||
|
||
version_files = sorted(path.name for path in VERSIONS_DIR.glob("*.py"))
|
||
revisions = list(directory.walk_revisions())
|
||
revision_ids = {revision.revision for revision in revisions}
|
||
assert len(revision_ids) == len(version_files), (
|
||
"迁移文件数量与已加载修订数量不一致（存在无法加载的脚本）"
|
||
)
|
||
for revision in revisions:
|
||
if revision.down_revision is not None:
|
||
assert revision.down_revision in revision_ids, (
|
||
f"{revision.revision} 引用了不存在的父版本 {revision.down_revision}"
|
||
)
|
||
|
||
def test_all_migration_scripts_compile(self):
|
||
for path in sorted(VERSIONS_DIR.glob("*.py")):
|
||
source = path.read_text(encoding="utf-8")
|
||
compile(source, str(path), "exec") # 语法/缩进等编译错误会在此抛出
|
||
|
||
def test_full_chain_applies_on_fresh_mysql_and_matches_models(self):
|
||
"""空库执行整条迁移链，校验表/列/索引与模型元数据对齐。"""
|
||
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
|
||
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(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
|
||
|
||
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:
|
||
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
|
||
with admin_engine.begin() as conn:
|
||
conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))
|