数据库测试与 Schema 变更安全网：迁移、数据层与数据管道的验证实践

挑战一：环境真实性问题
  内存 SQLite ≠ 生产 MySQL/Postgres
  → 类型、锁、隔离、函数差异导致"本地过、生产炸"

挑战二：状态性问题
  数据库是"有状态"的：上一次测试留下的数据污染下一次

挑战三：迁移的时序性
  Schema 要按顺序演进：V1 → V2 → V3
  每种数据库状态都可能是"上线点"，都要可验证

挑战四：性能与成本
  真实数据库测试慢、重、CI 开销大
  → 需要策略：什么测真库，什么可以模拟

分层原则：
  1. 查询/Schema 逻辑 → 真实数据库测试（Testcontainers）
  2. 业务逻辑与 ORM 映射 → 事务回滚的集成测试
  3. 纯计算（聚合、格式化）→ 单元测试（可脱离 DB）
  4. 数据管道正确性 → 管道测试 + 数据质量断言

一句话：能脱离数据库的逻辑尽量单元化，与数据库强相关的必须真库测

· 每个测试用真实数据库镜像（MySQL/Postgres/Redis...）
· 隔离：每套测试独立容器，无历史残留
· 兼容 CI：Docker 即可，本地与 CI 行为一致
· 性能：容器启动可复用（JUnit @Testcontainers 的容器级复用）

import org.junit.jupiter.api.Test;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

@Testcontainers
class OrderRepositoryIT {
    @Container
    static PostgreSQLContainer<?> pg =
        new PostgreSQLContainer<>("postgres:16")
            .withDatabaseName("test")
            .withUsername("test").withPassword("test");

    private OrderRepository repo;   // 注入连 pg 的仓库

    @Test
    void saveAndLoadOrder_roundtrip() {
        repo.save(new Order(1L, "PAID", 500));
        Order loaded = repo.get(1L);
        assertThat(loaded.status()).isEqualTo("PAID");
    }
}

# conftest.py：每个 worker 一个 Postgres 容器（并行隔离）
import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def db_url():
    with PostgresContainer("postgres:16") as pg:
        yield pg.get_connection_url()
        # 会话结束自动销毁容器

迁移（Migration）是生产变更，必须在低风险处验证：
  · 空库能否从 V1 一路迁到最新？（fresh install）
  · 已有库能否从 V(N) 安全升级到 V(N+1)？（upgrade）
  · 迁移能否回滚？（down / rollback）
  · 迁移在大表上是否太慢？（耗时/锁）
  · 迁移后数据是否完好？（before/after 对账）

# test_migrations.py
def test_fresh_install_migrations(db_url):
    """空库从 V1 迁到 latest。"""
    flyway = Flyway(url=db_url, locations=["db/migration"])
    flyway.migrate()                     # 空库跑全部迁移
    assert table_exists(db_url, "orders")

def test_upgrade_path(db_url):
    """已有 V2 的库升级到 latest。"""
    flyway_migrate(db_url, target="V2")  # 先迁到 V2
    flyway.migrate(target="latest")      # 升级到最新
    assert column_exists(db_url, "orders", "status")

def test_migration_rollback(db_url):
    """若支持回滚：V3 回退到 V2 数据完好。"""
    flyway_migrate(db_url, target="V3")
    insert_orders(db_url)
    flyway_migrate(db_url, target="V2", undo=True)
    assert count_orders(db_url) == expected

<!-- changelog 中显式定义校验：checksum 防篡改、上下文控制 -->
<changeSet id="1" author="me" runOnChange="false">
  <createTable tableName="orders">
    <column name="id" type="bigint" autoIncrement="true">
      <constraints primaryKey="true"/>
    </column>
    <column name="status" type="varchar(20)">
      <constraints nullable="false"/>
    </column>
  </createTable>
  <rollback>
    <dropTable tableName="orders"/>
  </rollback>
</changeSet>

Liquibase 测试要点：
  · validate：校验 changelog checksum 未被改动
  · updateTestingRollback：apply + rollback + 校验
  · 上下文（contexts）：dev/test/prod 用不同迁移分支
  · 在每个里程碑版本上跑"升级 + 回滚"套件

生产迁移的安全网（配合测试）：
  · 表结构变更用 ONLINE DDL（MySQL INSTANT/INPLACE）
  · 大规模回填用增量脚本 + 幂等重放（可断点续跑）
  · 迁移前做"影子库"演练：在复制库上先跑一遍真实迁移
  · 迁移后做数据对账：count/checksum 与迁移前对比

def test_rollback_on_failure(db_url):
    """业务逻辑失败 → 事务整体回滚。"""
    with session(db_url) as s:
        s.execute(insert_order_sql(id=1, status="PENDING"))
        with pytest.raises(Exception):
            with s.begin():                     # 内层事务
                s.execute(update_status_sql(id=1, status="PAID"))
                raise RuntimeError("fail")       # 触发回滚
        status = s.execute(select_status_sql(id=1)).scalar()
        assert status == "PENDING"              # 回滚未污染

隔离级别测试：
  · 读已提交（默认）：验证脏读不会发生
  · 可重复读：验证同一事务内读一致
  · 锁等待/死锁：构造并发事务，验证超时与重试逻辑
  → 这些行为 SQLite 与真库不一致，必须真库测

查询性能测试（防止"上线才炸"）：
  · 在接近生产的数据量下 EXPLAIN ANALYZE 关键查询
  · 断言走索引：type != ALL / Seq Scan 不应出现在热路径
  · 用执行计划差异对比"改动前后"（防查询回归）

示例断言（MySQL）：
  EXPLAIN SELECT * FROM orders WHERE user_id=?
  → 应显示 ref / range（使用 idx_user）而非 ALL（全表）

-- 查询计划回归测试的种子数据
CREATE TABLE orders (id BIGINT PRIMARY KEY, user_id BIGINT, ...);
CREATE INDEX idx_orders_user ON orders(user_id);
-- 插入足够数据（如 10w 行）让优化器真实决策

· 实体映射 ↔ 表结构一致性测试（列名/类型/长度）
  → 用 information_schema 对比自动发现漂移
· 唯一约束/外键/检查约束测试
  → 构造违反约束的写入，断言被拒绝
· 级联/软删/时间戳默认值的真实行为

# dbt tests/generic/relationships 等内置测试
version: 2
models:
  - name: stg_orders
    tests:
      - unique:
          column_name: order_id
      - not_null:
          column_name: order_id
      - accepted_values:
          column_name: status
          values: ['PENDING', 'PAID', 'CANCELLED']
      - relationships:
          column_name: user_id
          to: ref('stg_users')
          field: user_id

# 数据管道 CI：模型构建 + 测试一步到位
dbt build --target ci        # run + test 全链路
dbt test --select stg_orders # 只测转换层

import great_expectations as gx

context = gx.get_context()
validator = context.sources.pandas_default.read_csv(
    "data/orders_raw.csv").build_expectation_suite("orders_suite")

validator.expect_table_row_count_to_be_between(1_000, 100_000)
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_match_regex("amount", r"^\d+\.\d{2}$")
validator.expect_column_proportion_of_unique_values_to_be_between(
    "order_id", 0.99, 1.0)

result = validator.validate()
assert result.success        # 数据质量门禁

数据管道测试的核心性质：
  · 幂等性：管道重跑一遍，结果一致（无重复无漂移）
  · 全量+增量衔接：增量追平后与全量一致
  · 故障恢复：管道中断后重放，位点正确（配合 CDC/offset 测试）

示例断言：
  · 跑两遍 ETL，目标表 count 相同
  · 增量窗口与全量对账一致（chunk checksum）
  · 断点续跑后无重复行（主键去重验证）

def test_etl_idempotent():
    run_etl()                      # 第一遍
    first = snapshot_table("dw.orders")
    run_etl()                      # 第二遍
    second = snapshot_table("dw.orders")
    assert first == second         # 幂等：两遍结果一致

对账 = 迁移或同步后的最终检查：
  · count 对账：源表与目标表行数一致
  · checksum 对账：按 chunk 计算校验值对比
  · 抽样对账：关键字段逐行比对
  → 与"数据库一致性校验"专题的方法衔接（chunk + checksum）

场景：
  · Schema 迁移后：before/after 数据完好性
  · CDC 同步后：源库 ↔ 目标库一致
  · 备份恢复后：恢复库与源库一致

数据库本身的约束就是最好的"免费测试"：
  · NOT NULL / UNIQUE / CHECK / 外键
  · 触发器：复杂业务约束在 DB 层强制
  · 定期 DBCC CHECKCONSTRAINTS / 一致性检查
  这些在测试与生产同时生效，比应用层校验更可靠

CI 数据库测试的组合：
  · 每次提交：迁移套件（空库 + 升级）跑在 Testcontainers 上
  · 每次提交：核心数据层集成测试（事务/查询计划）
  · 定时（每日）：数据管道 + 数据质量套件（耗时较长）
  · 发布前：生产迁移的"影子库演练" + 对账脚本

并行注意：
  · Testcontainers 容器有资源成本 → 控制并发容器数
  · 每 worker 独立 schema，避免互相污染

jobs:
  db-tests:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: test
        options: >-
          --health-cmd "pg_isready" --health-interval 5s
    steps:
      - run: pytest tests/migrations tests/repository -m "not slow"
      - run: pytest tests/pipeline -m slow
      # dbt + 数据质量
      - run: dbt build --target ci

安全阀一：可回滚
  每个迁移配 rollback；上线前验证回滚路径

安全阀二：先影子库
  在复制库/影子库先跑完整迁移 + 对账

安全阀三：限速与分批
  大回填/DDL 分批执行 + 进度监控，异常即停

安全阀四：变更后验证
  迁移完成立即跑：数据对账 + 查询计划 + 关键查询冒烟

# post_migration_verify.py — 迁移后自动检查
def verify_migration():
    assert table_exists("orders", "status")     # 结构
    assert count_orders() == pre_count + added   # 数据完好
    explain_uses_index("orders", "idx_user")     # 索引生效
    smoke_query("SELECT * FROM orders WHERE user_id=?")  # 冒烟

□ 业务逻辑与 DB 强相关处用 Testcontainers 真库测
□ 迁移测试覆盖：空库安装 / 升级路径 / 回滚 / 大表耗时
□ 事务与隔离级别测试（真实数据库行为）
□ 关键查询做 EXPLAIN 断言（防全表扫描回归）
□ 数据管道测幂等 + 增量衔接 + 断点恢复
□ dbt test / Great Expectations 数据质量门禁
□ 迁移/同步后做 chunk 对账
□ CI 每次提交跑迁移套件，定时跑管道套件
□ 生产变更走影子库演练 + 可回滚 + 变更后自动验证
□ 用数据库约束（NOT NULL/UNIQUE/CHECK）做常驻校验

"数据库变更不是'发布个脚本'，而是'一次需要完整安全网的生产变更'。"