dbt 单元测试

💡提示：此功能从 dbt v1.8 起提供，也适用于 dbt“v1 Latest”发布轨道.

过去，dbt的测试覆盖主要限于数据测试，用于评估输入数据质量或结果数据集的结构，而且只能在模型构建之后执行。

dbt还提供单元测试。与软件开发中验证小段功能代码的单元测试类似，它使用少量静态输入验证SQL建模逻辑，让你在生产环境完整物化模型之前发现问题。它支持测试驱动开发，有助于提高开发效率与代码可靠性。

前置条件

目前只支持SQL模型的单元测试。
目前只能为当前项目中的模型添加单元测试。
目前不支持使用materialized view物化方式的模型。
目前不支持使用递归SQL的模型。
目前不支持使用内省查询的模型。
如果模型有多个版本，默认会对所有版本运行单元测试。详情参见版本化模型的单元测试。
单元测试必须定义在models/目录中的YML文件里。
测试连接逻辑时，表名必须使用别名。
把所有ref或source模型引用作为输入写入单元测试配置，避免编译时出现“node not found”。

dbt从model-paths中发现单元测试，默认路径为models/，因此应在该路径下用.yml文件与模型一同定义。不要把单元测试YAML放入tests/目录，该目录用于数据测试。

适配器特定限制

BigQuery单元测试中，必须指定STRUCT的全部字段，不能只提供部分字段。
Redshift用户需要注意构建单元测试时的一项限制，并采用相应变通方法。
Redshift的source必须与模型位于同一数据库。

提示

可观看Unit tests点播课程，学习如何添加单元测试。

单元测试格式的更多细节见参考文档。

何时为模型添加单元测试

以下情况适合添加单元测试：

SQL包含复杂逻辑，例如：
- 正则表达式。
- 日期运算。
- 窗口函数。
- 包含许多when分支的case when。
- 截断操作。
编写处理输入数据的自定义逻辑，类似于实现一个函数。
不建议对min()之类函数进行单元测试，因为数据仓库已对这些函数进行了广泛测试。出现异常时，问题更可能在底层数据，而不是函数本身；此时单元测试的固定数据通常无法提供有价值的信息。
曾经被报告过缺陷的逻辑。
希望处理、但实际数据尚未出现的边界情况。
重构转换逻辑之前，尤其是重大重构。
关键性高的模型，例如公开模型、带契约的模型，或直接位于exposure上游的模型。

何时运行单元测试

dbt Labs强烈建议只在开发或CI环境运行单元测试。由于输入是静态的，没有必要在生产环境重复消耗计算资源。开发时用于测试驱动流程，CI中则用于确保修改没有破坏既有逻辑。

使用--exclude-resource-type资源类型参数，或dbt v1.11及以后版本的DBT_ENGINE_EXCLUDE_RESOURCE_TYPES环境变量，从生产构建中排除单元测试，以节省计算资源。

按需只运行单元测试时，可使用test_type选择器；dbt v1与dbt v2引擎均支持：

dbt test --select "test_type:unit"

报告代码问题（适用于 dbt v2.0 及以后版本）

在本地运行单元测试

通过环境变量启用

单个单元测试的本地计算执行目前属于实验功能。使用compute: local之前，必须在dbt运行环境中设置DBT_ENGINE_EXPERIMENTAL_LOCAL_UNIT_TESTS=true。

处理复杂 SQL 时，可以在本地运行单元测试，立即判断逻辑是否正确。默认情况下，每个单元测试都会向数据平台发送查询并等待结果，这可能拖慢测试并消耗数据仓库的计算资源。

由于单元测试使用静态测试数据而非真实数据，因此不一定需要在数据平台中运行。使用 compute: local 配置，可通过 DuckDB 在本地运行测试，更快获得反馈， 无需承担 数据仓库的计算成本。

这样可以获得：

测试迭代更及时：修改SQL、重跑测试并重复，无需等待数据仓库队列。
在编写过程中随时测试，在CI甚至生产环境之前发现错误逻辑。
查询按与远程数据仓库相同的函数和运行时语义执行。
让数据平台的计算资源留给模型构建。

前置条件

将DBT_ENGINE_EXPERIMENTAL_LOCAL_UNIT_TESTS设为true。此配置为实验功能，需要主动启用；否则dbt会报无效配置错误，并指出该变量。
使用Snowflake或BigQuery。目前其他数据平台不支持本地执行，而且本地执行只适用于单元测试。
被测模型的直接上游模型已存在于数据平台。dbt需要读取其schema来转换SQL，否则会报无法获取上游关系schema的错误。
测试的static_analysis不能设为off。本地执行需要静态分析来转换SQL，因此compute: local会把该测试的静态分析提升为strict。若设为off，将以ExecutorFailed（dbt1401）失败。

使用前须知

首次运行时，dbt读取并缓存上游schema；后续运行复用缓存，不再重新获取。
本地执行要求dbt能够编译模型SQL并转换为DuckDB。复杂SQL或平台特定函数可能没有DuckDB等价物，例如Snowflake的AI_CLASSIFY与haversine。调用haversine的模型会报错：failed in db_runner: Internal: Catalog Error: Scalar Function with name haversine does not exist!。
转换失败即测试失败。local不会回退到数据平台执行，dbt会以非零状态退出。

配置方法

首先，在运行 dbt 的环境中启用此配置：

export DBT_ENGINE_EXPERIMENTAL_LOCAL_UNIT_TESTS=true

报告代码问题

接着，可以为单个单元测试进行配置：models/schema.yml

unit_tests:
  - name: test_is_valid_email_address
    model: dim_customers
    config:
      compute: local

报告代码问题

也可以为项目中的所有单元测试配置：dbt_project.yml

unit_tests:
  my_project:
    +compute: local

报告代码问题

compute 接受两个值：

值	作用
`remote`	把测试发送到数据平台，像其他查询一样使用数据仓库计算资源。这是默认值；只有要覆盖项目级`+compute: local`、让某个测试改为远程执行时，才需要显式设置。
`local`	在dbt所在环境使用DuckDB执行测试，不把测试发送到数据平台，因此响应快，也不使用数据仓库计算资源。

错误消息中可能出现sidecar，它与local含义相同。

为模型编写单元测试

本例新建dim_customers模型，使用is_valid_email_address字段判断客户邮箱是否合法：

with customers as (

    select * from {{ ref('stg_customers') }}

),

accepted_email_domains as (

    select * from {{ ref('top_level_email_domains') }}

),
	
check_valid_emails as (

    select
        customers.customer_id,
        customers.first_name,
        customers.last_name,
        customers.email,
	      coalesce (regexp_like(
            customers.email, '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'
        )
        = true
        and accepted_email_domains.tld is not null,
        false) as is_valid_email_address
    from customers
		left join accepted_email_domains
        on customers.email_top_level_domain = lower(accepted_email_domains.tld)

)

select * from check_valid_emails

报告代码问题

这段逻辑不容易直接验证。可以添加单元测试，覆盖已知边界：缺少点号、缺少@、以及域名无效的邮箱。

unit_tests:
  - name: test_is_valid_email_address
    description: "Check my is_valid_email_address logic captures all known edge cases - emails without ., emails without @, and emails from invalid domains."
    model: dim_customers
    given:
      - input: ref('stg_customers')
        format: dict
        rows:
          - {email: cool@example.com,    email_top_level_domain: example.com}
          - {email: cool@unknown.com,    email_top_level_domain: unknown.com}
          - {email: badgmail.com,        email_top_level_domain: gmail.com}
          - {email: missingdot@gmailcom, email_top_level_domain: gmail.com}
      - input: ref('top_level_email_domains')
        format: dict
        rows:
          - {tld: example.com}
          - {tld: gmail.com}
    expect:
      format: dict
      rows:
        - {email: cool@example.com,    is_valid_email_address: true}
        - {email: cool@unknown.com,    is_valid_email_address: false}
        - {email: badgmail.com,        is_valid_email_address: false}
        - {email: missingdot@gmailcom, is_valid_email_address: false}

报告代码问题

上例使用内联dict格式定义模拟数据，也可以使用csv或sql，写在内联配置或独立fixture文件中。fixture应放在任一测试路径的fixtures子目录，例如tests/fixtures/my_unit_test_fixture.sql。

以下示例通过csv和sql定义模拟输入与预期输出。

models/schema.yml

unit_tests:
  - name: test_is_valid_email_address__csv
    model: dim_customers
    given:
      - input: ref('stg_customers')
        format: dict
        rows:
          - {email: cool@example.com,    email_top_level_domain: example.com}
          - {email: cool@unknown.com,    email_top_level_domain: unknown.com}
          - {email: badgmail.com,        email_top_level_domain: gmail.com}
          - {email: missingdot@gmailcom, email_top_level_domain: gmail.com}
      - input: ref('top_level_email_domains')
        format: csv
        rows: |
          tld
          example.com
          gmail.com
    expect:
      format: csv
      fixture: valid_email_address_fixture_output

报告代码问题：models/schema.yml

unit_tests:
  - name: test_is_valid_email_address__sql
    model: dim_customers
    given:
      - input: ref('stg_customers')
        format: dict
        rows:
          - {email: cool@example.com,    email_top_level_domain: example.com}
          - {email: cool@unknown.com,    email_top_level_domain: unknown.com}
          - {email: badgmail.com,        email_top_level_domain: gmail.com}
          - {email: missingdot@gmailcom, email_top_level_domain: gmail.com}
      - input: ref('top_level_email_domains')
        format: sql
        rows: |
          select 'example.com' as tld union all
          select 'gmail.com' as tld
    expect:
      format: sql
      fixture: valid_email_address_fixture_output

报告代码问题

使用dict或csv时，只需为相关列定义模拟数据，从而让测试简洁、聚焦。

注意

执行单元测试之前，待测模型的直接上游模型（本例中的 stg_customers 和 top_level_email_domains）必须已经存在于数据仓库中。

使用 --empty 标志构建模型的空版本，以节省数据仓库费用。


dbt run --select "stg_customers top_level_email_domains" --empty

报告代码问题

或者使用 dbt build ，按血缘依赖顺序执行以下操作：

运行模型的单元测试。
在数据仓库物化模型。
运行模型的数据测试。

现在可以执行单元测试。根据需要限定的范围，有以下命令选择：

dbt test --select dim_customers：运行dim_customers上的全部测试。
dbt test --select "dim_customers,test_type:unit"：运行该模型的全部单元测试。
dbt test --select test_is_valid_email_address：运行指定名称的测试。


dbt test --select test_is_valid_email_address
16:03:49  Running with dbt=1.8.0-a1
16:03:49  Registered adapter: postgres=1.8.0-a1
16:03:50  Found 6 models, 5 seeds, 4 data tests, 0 sources, 0 exposures, 0 metrics, 410 macros, 0 groups, 0 semantic models, 1 unit test
16:03:50  
16:03:50  Concurrency: 5 threads (target='postgres')
16:03:50  
16:03:50  1 of 1 START unit_test dim_customers::test_is_valid_email_address ................... [RUN]
16:03:51  1 of 1 FAIL 1 dim_customers::test_is_valid_email_address ............................ [FAIL 1 in 0.26s]
16:03:51  
16:03:51  Finished running 1 unit_test in 0 hours 0 minutes and 0.67 seconds (0.67s).
16:03:51  
16:03:51  Completed with 1 error and 0 warnings:
16:03:51  
16:03:51  Failure in unit_test test_is_valid_email_address (models/marts/unit_tests.yml)
16:03:51    

actual differs from expected:

@@ ,email           ,is_valid_email_address
→  ,cool@example.com,True→False
   ,cool@unknown.com,False
...,...             ,...


16:03:51  
16:03:51    compiled Code at models/marts/unit_tests.yml
16:03:51  
16:03:51  Done. PASS=0 WARN=0 ERROR=1 SKIP=0 TOTAL=1

报告代码问题

原示例中的正则表达式并没有预期那样正确：模型把cool@example.com错误判断为无效邮箱。

把正则逻辑更新为'^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}，修正转义字符，再运行测试即可解决：，修正转义字符，再运行测试即可解决：


dbt test --select test_is_valid_email_address
16:09:11  Running with dbt=1.8.0-a1
16:09:12  Registered adapter: postgres=1.8.0-a1
16:09:12  Found 6 models, 5 seeds, 4 data tests, 0 sources, 0 exposures, 0 metrics, 410 macros, 0 groups, 0 semantic models, 1 unit test
16:09:12  
16:09:13  Concurrency: 5 threads (target='postgres')
16:09:13  
16:09:13  1 of 1 START unit_test dim_customers::test_is_valid_email_address ................... [RUN]
16:09:13  1 of 1 PASS dim_customers::test_is_valid_email_address .............................. [PASS in 0.26s]
16:09:13  
16:09:13  Finished running 1 unit_test in 0 hours 0 minutes and 0.75 seconds (0.75s).
16:09:13  
16:09:13  Completed successfully
16:09:13  
16:09:13  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

报告代码问题

示例模型现在可以进入生产流程。单元测试在dim_customers物化到数据仓库之前发现了SQL逻辑问题，也有助于今后保持模型可靠。

增量模型的单元测试

单元测试配置可以覆盖宏、vars或环境变量的输出，因此能够分别验证增量模型的全量刷新与增量模式。

注意

在运行单元测试或执行 dbt build之前，增量模型必须已经存在于数据库中。使用 --empty 标志构建模型的空版本，以节省数据仓库费用。还可以选择使用以下标志，仅选择增量模型： --select 标志.

dbt run --select "config.materialized:incremental" --empty

报告代码问题

运行上述命令后，可以对模型执行常规dbt build，再运行单元测试。

增量模型测试的预期输出，是此次物化准备合并或插入的记录，而不是合并或插入后最终表的完整状态。

例如，项目中有以下增量模型：

my_incremental_model.sql


{{
    config(
        materialized='incremental'
    )
}}

select * from {{ ref('events') }}
{% if is_incremental() %}
where event_time > (select max(event_time) from {{ this }})
{% endif %}

报告代码问题

可以为my_incremental_model定义以下单元测试，验证增量逻辑符合预期：


unit_tests:
  - name: my_incremental_model_full_refresh_mode
    model: my_incremental_model
    overrides:
      macros:
        # unit test this model in "full refresh" mode
        is_incremental: false 
    given:
      - input: ref('events')
        rows:
          - {event_id: 1, event_time: 2020-01-01}
    expect:
      rows:
        - {event_id: 1, event_time: 2020-01-01}

  - name: my_incremental_model_incremental_mode
    model: my_incremental_model
    overrides:
      macros:
        # unit test this model in "incremental" mode
        is_incremental: true 
    given:
      - input: ref('events')
        rows:
          - {event_id: 1, event_time: 2020-01-01}
          - {event_id: 2, event_time: 2020-01-02}
          - {event_id: 3, event_time: 2020-01-03}
      - input: this 
        # contents of current my_incremental_model
        rows:
          - {event_id: 1, event_time: 2020-01-01}
    expect:
      # what will be inserted/merged into my_incremental_model
      rows:
        - {event_id: 2, event_time: 2020-01-02}
        - {event_id: 3, event_time: 2020-01-03}

报告代码问题

目前还不能通过单元测试验证dbt框架是否正确地把记录插入或合并到现有模型中；维护者正在研究未来支持。

依赖 ephemeral 模型的单元测试

被测模型依赖ephemeral模型时，该输入必须使用format: sql。

unit_tests:
  - name: my_unit_test
    model: dim_customers
    given:
      - input: ref('ephemeral_model')
        format: sql
        rows: |
          select 1 as id, 'emily' as first_name
    expect:
      rows:
        - {id: 1, first_name: emily}

报告代码问题