Dbt Testing

# packages.yml
packages:
  - package: Snowflake-Labs/dbt_constraints
    version: [">=0.8.0", "<1.0.0"]

  - package: dbt-labs/dbt_utils
    version: [">=1.0.0", "<2.0.0"]

dbt deps

# models/gold/_models.yml
models:
  - name: dim_customers
    columns:
      - name: customer_id
        tests:
          - dbt_constraints.primary_key

models:
  - name: fct_order_lines
    tests:
      - dbt_constraints.primary_key:
          column_names:
            - order_id
            - line_number

columns:
  - name: product_id
    tests:
      - not_null
      - unique

models:
  - name: fct_orders
    columns:
      - name: customer_id
        tests:
          - dbt_constraints.foreign_key:
              pk_table_name: ref('dim_customers')
              pk_column_name: customer_id

models:
  - name: fct_order_lines
    columns:
      - name: order_id
        tests:
          - dbt_constraints.foreign_key:
              pk_table_name: ref('fct_orders')
              pk_column_name: order_id

      - name: product_id
        tests:
          - dbt_constraints.foreign_key:
              pk_table_name: ref('dim_products')
              pk_column_name: product_id

      - name: customer_id
        tests:
          - dbt_constraints.foreign_key:
              pk_table_name: ref('dim_customers')
              pk_column_name: customer_id

columns:
  - name: customer_id
    tests:
      - relationships:
          to: ref('dim_customers')
          field: customer_id

columns:
  - name: customer_email
    tests:
      - dbt_constraints.unique_key

models:
  - name: stg_orders
    tests:
      - dbt_constraints.unique_key:
          column_names:
            - order_number
            - order_source

columns:
  - name: order_status
    tests:
      - not_null
      - accepted_values:
          values: ["pending", "processing", "shipped", "delivered", "cancelled"]

  - name: order_amount
    tests:
      - not_null

columns:
  - name: customer_email
    tests:
      - dbt_utils.not_null_proportion:
          at_least: 0.95 # 95% of rows must have email

  - name: order_amount
    tests:
      - dbt_utils.accepted_range:
          min_value: 0
          max_value: 1000000

  - name: customer_status
    tests:
      - dbt_utils.not_empty_string

-- tests/generic/test_positive_values.sql
{% test positive_values(model, column_name) %}

select count(*)
from {{ model }}
where {{ column_name }} <= 0

{% endtest %}

columns:
  - name: order_total
    tests:
      - positive_values

  - name: quantity
    tests:
      - positive_values

-- tests/generic/test_recent_data.sql
{% test recent_data(model, column_name, days_ago=30) %}

select count(*)
from {{ model }}
where {{ column_name }} < dateadd(day, -{{ days_ago }}, current_date())

{% endtest %}

columns:
  - name: order_date
    tests:
      - recent_data:
          days_ago: 7 # Alert if no orders in last 7 days

-- tests/singular/test_order_dates_sequential.sql
with date_validation as (
    select
        o.order_id,
        o.order_date,
        c.signup_date
    from {{ ref('fct_orders') }} o
    join {{ ref('dim_customers') }} c
        on o.customer_id = c.customer_id
    where o.order_date < c.signup_date  -- Order before signup = invalid
)

select * from date_validation

-- tests/singular/test_revenue_reconciliation.sql
-- Ensure fact table revenue matches source system
with fact_revenue as (
    select sum(order_amount) as total_revenue
    from {{ ref('fct_orders') }}
    where order_date = current_date() - 1
),

source_revenue as (
    select sum(amount) as total_revenue
    from {{ source('erp', 'orders') }}
    where order_date = current_date() - 1
),

comparison as (
    select
        f.total_revenue as fact_revenue,
        s.total_revenue as source_revenue,
        abs(f.total_revenue - s.total_revenue) as difference
    from fact_revenue f
    cross join source_revenue s
)

select * from comparison
where difference > 0.01  -- Tolerance of 1 cent

-- tests/singular/test_orphaned_orders.sql
-- Find orders with invalid customer_id (not in dim_customers)
select
    o.order_id,
    o.customer_id
from {{ ref('fct_orders') }} o
left join {{ ref('dim_customers') }} c
    on o.customer_id = c.customer_id
where c.customer_id is null
  and o.customer_id != -1  -- Exclude ghost key

models:
  - name: stg_tpc_h__customers
    columns:
      - name: customer_id
        tests:
          - dbt_constraints.primary_key

      - name: customer_email
        tests:
          - not_null

models:
  - name: int_customers__with_orders
    columns:
      - name: customer_id
        tests:
          - dbt_constraints.primary_key

      - name: lifetime_orders
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

      - name: lifetime_value
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

models:
  - name: dim_customers
    description: "Customer dimension with full history and metrics"
    columns:
      - name: customer_id
        description: "Unique customer identifier"
        tests:
          - dbt_constraints.primary_key

      - name: customer_tier
        description: "Customer value classification"
        tests:
          - accepted_values:
              values: ["bronze", "silver", "gold", "platinum"]

      - name: customer_email
        tests:
          - dbt_constraints.unique_key

  - name: fct_orders
    description: "Order transactions fact table"
    columns:
      - name: order_id
        tests:
          - dbt_constraints.primary_key

      - name: customer_id
        tests:
          - dbt_constraints.foreign_key:
              pk_table_name: ref('dim_customers')
              pk_column_name: customer_id

      - name: product_id
        tests:
          - dbt_constraints.foreign_key:
              pk_table_name: ref('dim_products')
              pk_column_name: product_id

      - name: order_amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

dbt test --store-failures

# dbt_project.yml
tests:
  +store_failures: true
  +schema: dbt_test_failures

select * from dbt_test_failures.not_null_dim_customers_customer_email

columns:
  - name: customer_email
    tests:
      - dbt_constraints.unique_key:
          config:
            severity: warn # or 'error' (default)

dbt test --select dim_customers

dbt test --select test_type:generic    # All generic tests
dbt test --select test_type:singular   # All singular tests

dbt test --select +dim_customers+  # Test model and all dependencies