Testing

SQLMesh Testing Guide: Unit Tests, Audits, and CI Validation for SQL Pipelines

12 Oct 2026 — 5 min read

SQLMesh is a SQL transformation framework that competes with dbt but adds first-class testing, change management, and virtual data environments. Its testing model is notably different from dbt's: unit tests run against in-memory DuckDB without a warehouse, audits run post-deployment assertions, and the sqlmesh plan command shows exactly what SQL will change before any model runs.

This guide covers the full SQLMesh testing stack.

SQLMesh Testing Concepts

SQLMesh has three distinct testing mechanisms:

Unit tests — test model SQL logic with fixtures (no warehouse needed)
Audits — post-deployment assertions that run after a model populates
Plan validation — before any change deploys, sqlmesh plan shows what changes and requires approval

The combination means: catch logic bugs in unit tests locally, catch data quality issues in audits in CI, and catch unintended model changes in plan review.

Project Setup

pip install sqlmesh
sqlmesh init my_project
cd my_project

Project structure:

my_project/
├── config.yaml           # project config (connection, state backend)
├── models/
│   ├── staging/
│   │   ├── stg_orders.sql
│   │   └── stg_customers.sql
│   └── marts/
│       └── orders_by_customer.sql
├── tests/
│   ├── test_stg_orders.yaml
│   └── test_orders_by_customer.yaml
└── audits/
    ├── not_null.sql
    └── unique.sql

Writing Unit Tests

SQLMesh unit tests live in tests/ as YAML files. They specify input fixtures and expected output:

# tests/test_stg_orders.yaml
test_stg_orders_basic:
  model: staging.stg_orders
  inputs:
    raw.orders:
      - order_id: 1
        customer_id: 100
        amount: 50.00
        status: 'completed'
        created_at: '2024-01-15 10:00:00'
      - order_id: 2
        customer_id: 101
        amount: 0.00
        status: 'pending'
        created_at: '2024-01-15 11:00:00'
      - order_id: 3
        customer_id: 100
        amount: -5.00     # refund — should be excluded
        status: 'refunded'
        created_at: '2024-01-15 12:00:00'
  outputs:
    rows:
      - order_id: 1
        customer_id: 100
        amount: 50.00
        status: 'completed'
        order_date: '2024-01-15'
      - order_id: 2
        customer_id: 101
        amount: 0.00
        status: 'pending'
        order_date: '2024-01-15'
    # Order 3 excluded because status = 'refunded'

test_stg_orders_zero_amount:
  model: staging.stg_orders
  inputs:
    raw.orders:
      - order_id: 5
        customer_id: 200
        amount: 0.00
        status: 'completed'
        created_at: '2024-02-01 09:00:00'
  outputs:
    rows:
      - order_id: 5
        amount: 0.00
        status: 'completed'

The model being tested:

-- models/staging/stg_orders.sql
MODEL (
  name staging.stg_orders,
  kind VIEW
);

SELECT
  order_id,
  customer_id,
  amount,
  status,
  DATE(created_at) AS order_date
FROM raw.orders
WHERE status != 'refunded'

Run unit tests:

sqlmesh test              # all tests
sqlmesh test tests/test_stg_orders.yaml  # specific file
sqlmesh test -k test_stg_orders_basic    # specific test by name

Tests run against DuckDB in memory. No warehouse connection required.

Testing Aggregation Logic

For models with complex aggregations:

# tests/test_orders_by_customer.yaml
test_orders_by_customer_aggregation:
  model: marts.orders_by_customer
  inputs:
    staging.stg_orders:
      - order_id: 1
        customer_id: 100
        amount: 50.00
        status: 'completed'
        order_date: '2024-01-15'
      - order_id: 2
        customer_id: 100
        amount: 30.00
        status: 'completed'
        order_date: '2024-01-16'
      - order_id: 3
        customer_id: 101
        amount: 20.00
        status: 'pending'
        order_date: '2024-01-15'
  outputs:
    rows:
      - customer_id: 100
        total_orders: 2
        total_amount: 80.00
        first_order_date: '2024-01-15'
        last_order_date: '2024-01-16'
      - customer_id: 101
        total_orders: 1
        total_amount: 20.00
        first_order_date: '2024-01-15'
        last_order_date: '2024-01-15'

test_single_customer_multiple_same_day:
  model: marts.orders_by_customer
  inputs:
    staging.stg_orders:
      - order_id: 10
        customer_id: 200
        amount: 10.00
        status: 'completed'
        order_date: '2024-03-01'
      - order_id: 11
        customer_id: 200
        amount: 15.00
        status: 'completed'
        order_date: '2024-03-01'
  outputs:
    rows:
      - customer_id: 200
        total_orders: 2
        total_amount: 25.00
        first_order_date: '2024-03-01'
        last_order_date: '2024-03-01'

-- models/marts/orders_by_customer.sql
MODEL (
  name marts.orders_by_customer,
  kind FULL
);

SELECT
  customer_id,
  COUNT(*) AS total_orders,
  SUM(amount) AS total_amount,
  MIN(order_date) AS first_order_date,
  MAX(order_date) AS last_order_date
FROM staging.stg_orders
GROUP BY 1

Partial Column Testing

When you only want to assert specific columns:

test_orders_key_columns_only:
  model: staging.stg_orders
  inputs:
    raw.orders:
      - order_id: 1
        customer_id: 100
        amount: 50.00
        status: 'completed'
        created_at: '2024-01-15 10:00:00'
  outputs:
    partial: true    # only assert the columns listed below
    rows:
      - order_id: 1
        status: 'completed'
        # amount, order_date, customer_id not asserted

This is useful for wide models where you only care about specific transformations.

Audits

Audits are SQL assertions that run after a model populates in the warehouse. They're defined inline in model files or in the audits/ directory:

Inline Audits

-- models/staging/stg_orders.sql
MODEL (
  name staging.stg_orders,
  kind VIEW,
  audits (
    not_null(columns := [order_id, customer_id]),
    unique_values(columns := [order_id]),
    accepted_values(column := status, values := ['completed', 'pending', 'cancelled']),
    number_of_rows(threshold := 1)
  )
);

SELECT ...

Built-in audit types:

not_null(columns := [...]) — no NULLs in specified columns
unique_values(columns := [...]) — no duplicates
accepted_values(column := col, values := [...]) — column only contains expected values
number_of_rows(threshold := N) — model has at least N rows
forall(criteria := 'amount >= 0') — all rows satisfy condition

Custom Audits

-- audits/no_negative_amounts.sql
AUDIT (
  name no_negative_amounts,
  dialect spark
);

SELECT *
FROM @this   -- @this refers to the audited model
WHERE amount < 0
-- Returns failing rows — audit passes if result is empty

Reference in model:

MODEL (
  name staging.stg_orders,
  audits (no_negative_amounts())
);

Running Audits

sqlmesh audit             # run all audits
sqlmesh audit --model staging.stg_orders  # specific model
sqlmesh run --audit       # run models + audits together

Plan Validation: Change Review Before Deployment

SQLMesh's sqlmesh plan command is itself a testing mechanism — it shows exactly what will change:

sqlmesh plan dev

Output:

New environment `dev` will be created from `prod`

Added Models:
└── staging.stg_orders_v2

Directly Modified:
└── marts.orders_by_customer [Breaking]
    Change: SELECT now includes order_month column

Indirectly Modified:
└── reporting.monthly_revenue [Non-breaking]

Backfill:
└── marts.orders_by_customer  2024-01-01 - 2024-12-31

Apply - Backfill Tables [y/n]?

"Breaking" vs "Non-breaking" changes:

Breaking: SQL logic changed in a way that produces different results for existing data → requires backfill
Non-breaking: added column, changed metadata → no backfill needed

In CI, use --no-prompts to auto-apply in test environments:

sqlmesh plan dev --no-prompts
sqlmesh run dev
sqlmesh audit dev

Diff Testing: Comparing Environments

SQLMesh can compare model outputs between environments:

# Compare dev vs prod
sqlmesh table_diff prod:dev staging.stg_orders

# Output:
# Column Summary:
#   order_id: 0 differences
#   amount: 1,234 differences (0.3%)
#   status: 0 differences
#
# Row counts: prod=412,832  dev=412,845  (+13 rows)

This catches data drift between environments before promoting changes.

CI Integration

# .github/workflows/sqlmesh.yml
name: SQLMesh Tests

on:
  pull_request:
    paths:
      - 'models/**'
      - 'tests/**'
      - 'audits/**'

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install SQLMesh
        run: pip install sqlmesh[duckdb]

      - name: Run unit tests
        run: sqlmesh test

  plan-check:
    runs-on: ubuntu-latest
    needs: unit-tests
    steps:
      - uses: actions/checkout@v4

      - name: Install SQLMesh
        run: pip install sqlmesh[snowflake]

      - name: Create CI environment
        run: sqlmesh plan ci --no-prompts --auto-apply
        env:
          SQLMESH_SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SQLMESH_SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
          SQLMESH_SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

      - name: Run models and audits
        run: |
          sqlmesh run ci
          sqlmesh audit ci

      - name: Table diff vs prod
        run: sqlmesh table_diff prod:ci --model-filters marts.*
        continue-on-error: true  # informational — don't block PR

Python Tests for Complex Logic

For transformations too complex for YAML fixtures, use pytest:

# tests/test_complex_logic.py
import pytest
from sqlmesh.testing import ModelTestFixture

def test_customer_lifetime_value():
    """Test CLV calculation model."""
    fixture = ModelTestFixture(
        model_name="marts.customer_ltv",
        inputs={
            "marts.orders_by_customer": [
                {"customer_id": 1, "total_orders": 5, "total_amount": 250.0,
                 "first_order_date": "2022-01-01", "last_order_date": "2024-01-01"},
                {"customer_id": 2, "total_orders": 1, "total_amount": 30.0,
                 "first_order_date": "2024-01-01", "last_order_date": "2024-01-01"},
            ]
        }
    )

    results = fixture.run()

    customer_1 = next(r for r in results if r["customer_id"] == 1)
    assert customer_1["ltv_segment"] == "high"
    assert customer_1["years_as_customer"] == 2

    customer_2 = next(r for r in results if r["customer_id"] == 2)
    assert customer_2["ltv_segment"] == "new"

SQLMesh vs dbt Testing

Feature	SQLMesh	dbt
Unit tests	YAML fixtures, DuckDB (no warehouse)	dbt 1.8+ unit tests (warehouse required by default)
Schema tests	Built-in audits	generic tests (not_null, unique, etc.)
Custom tests	SQL returning failing rows	SQL macros
Change preview	`sqlmesh plan` shows SQL diff	`dbt state:modified`
Env isolation	Virtual environments (no data copies)	Schemas/datasets per environment
Backfill	Automatic based on change type	Manual `dbt build --full-refresh`

The key advantage: SQLMesh unit tests run in milliseconds against DuckDB. No warehouse connection, no warehouse cost, fast CI.

Summary

SQLMesh testing is more complete than most SQL transformation frameworks: unit tests validate logic without a warehouse, audits validate data quality post-deployment, and sqlmesh plan makes every infrastructure change reviewable. Start with unit tests for every model that has business logic (joins, filters, calculations), add not_null and unique audits for key columns, and use sqlmesh plan in CI to catch unintended downstream impacts before they reach production.