数据库迁移测试

概念定义

迁移是在不可替代的数据上恰好运行一次的代码。迁移测试就是把普通测试纪律施加到这份代码上：代表性输入、预期输出、时长与锁窗口实测、可逆性刻意决策。

工作机制

CI 从脱敏生产快照建库、按序施加迁移、断言：模式与期望定义一致、约束在转换后的数据上成立、幂等或失败语义符合设计。时长与锁窗口按产量实测，超过维护预算就不过闸。

使用场景

每个模式与数据迁移，包括看似无害但在某些引擎上重写整表的列变更。配合 expand-contract 分阶段，让应用部署与模式切换解耦。

常见误区

“迁移只跑一次所以没法测”恰好反了：一次性脚本正是测试的靶子，生产上的失败迁移是最贵的部署。空库测试没意义——空库的执行计划、约束顺序、耗时都不同，毫秒级完成掩盖了生产的数小时。给破坏性变更写回滚脚本会丢中间提交的数据，forward-only 加补偿才诚实。快照会老化，统计与分布漂移改变计划与时长；脱敏可能破坏迁移依赖的唯一性；多租户要覆盖租户变体，否则第九百零一个租户上翻车。

代码示例

export function createBuilder(defaults) {
  if (defaults === null || typeof defaults !== "object" || Array.isArray(defaults)) {
    throw new TypeError("defaults must be a plain object");
  }
  const layers = [];
  const build = (overrides = {}) => Object.assign({}, defaults, ...layers, overrides);
  const assertOverrides = (overrides) => {
    if (overrides === null || typeof overrides !== "object" || Array.isArray(overrides)) {
      throw new TypeError("overrides must be a plain object");
    }
  };
  const api = {
    with(overrides) {
      assertOverrides(overrides);
      layers.push(overrides);
      return api;
    },
    reset() {
      layers.length = 0;
      return api;
    },
    build,
    buildList(count, overrides = () => ({})) {
      if (!Number.isInteger(count) || count < 0) {
        throw new RangeError("count must be a non-negative integer");
      }
      if (typeof overrides !== "function") throw new TypeError("overrides must be a function");
      return Array.from({ length: count }, (_, index) => build(overrides(index)));
    },
  };
  return api;
}

const user = createBuilder({ role: "member", active: true });
console.log(user.with({ role: "admin" }).build({ name: "Ada" }));
console.log(user.reset().buildList(2, (index) => ({ id: index + 1 })));

import sqlite3
from typing import Iterator

import pytest


@pytest.fixture
def db() -> Iterator[sqlite3.Connection]:
    connection = sqlite3.connect(":memory:")
    connection.execute(
        "CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT NOT NULL)"
    )
    try:
        yield connection
    finally:
        connection.close()


def test_insert_item(db: sqlite3.Connection) -> None:
    db.execute("INSERT INTO items (name) VALUES (?)", ("widget",))
    count = db.execute("SELECT COUNT(*) FROM items").fetchone()[0]
    assert count == 1