Implementation:TobikoData Sqlmesh Context Plan Test For Dbt

Knowledge Sources	SQLMesh SQLMesh Docs
Domains	Data_Engineering, Dbt_Migration
Last Updated	2026-02-07 00:00 GMT

Overview

Concrete tool for validating dbt model migrations through SQLMesh's plan generation and testing framework provided by SQLMesh.

Description

The Context class provides plan() and test() methods that validate migrated dbt models work correctly in SQLMesh. The plan() method generates an execution plan that validates model queries parse correctly, dependencies form a valid DAG, and configurations are compatible with the target engine. The test() method executes SQLMesh unit tests against converted models, validating that they produce expected results. Together, these methods provide comprehensive validation of dbt-to-SQLMesh migrations.

For dbt projects, plan() is particularly important because it validates that materialization mappings are correct, incremental strategies work as expected, and macro expansion produces valid SQL. The skip_tests parameter allows generating plans without running tests, useful for quick validation during development.

Usage

Use Context.plan() and Context.test() after loading a dbt project into SQLMesh to validate the migration. Run plan() to validate model structure and dependencies, then run test() to validate model logic and outputs. These methods are essential steps in the dbt migration workflow, catching issues before production deployment.

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/core/context.py

Signature

class Context:
    def plan(
        self,
        environment: t.Optional[str] = None,
        *,
        start: t.Optional[TimeLike] = None,
        end: t.Optional[TimeLike] = None,
        skip_tests: t.Optional[bool] = None,
        restate_models: t.Optional[t.Iterable[str]] = None,
        no_gaps: t.Optional[bool] = None,
        skip_backfill: t.Optional[bool] = None,
        forward_only: t.Optional[bool] = None,
        no_prompts: t.Optional[bool] = None,
        auto_apply: t.Optional[bool] = None,
        select_models: t.Optional[t.Collection[str]] = None,
        ...
    ) -> Plan:
        ...

    def test(
        self,
        match_patterns: t.Optional[t.List[str]] = None,
        tests: t.Optional[t.List[str]] = None,
        verbosity: Verbosity = Verbosity.DEFAULT,
        preserve_fixtures: bool = False,
        stream: t.Optional[t.TextIO] = None,
    ) -> ModelTextTestResult:
        ...

Import

from pathlib import Path
from sqlmesh.core.context import Context

I/O Contract

Inputs

Name	Type	Required	Description
environment	str	No	Target environment name; defaults to configured default environment
start	TimeLike	No	Backfill start date for incremental models
end	TimeLike	No	Backfill end date for incremental models
skip_tests	bool	No	Skip running tests during plan generation; useful for quick validation
select_models	Collection[str]	No	Filter models to include in plan using selection syntax
match_patterns	List[str]	No	Test name patterns to match for test execution
tests	List[str]	No	Specific test names to execute
verbosity	Verbosity	No	Output verbosity level for test results

Outputs

Name	Type	Description
plan	Plan	Execution plan with model changes, backfill requirements, and validation results
test_result	ModelTextTestResult	Test execution results with pass/fail status for each test

Usage Examples

Basic Plan Validation for dbt Project

from pathlib import Path
from sqlmesh.core.context import Context

# Load dbt project
context = Context(paths=Path("/dbt/project"))

# Generate plan for dev environment
plan = context.plan(
    environment="dev",
    skip_tests=False  # Run tests as part of plan
)

# Examine plan results
print(f"Modified models: {len(plan.modified)}")
print(f"New models: {len(plan.new_snapshots)}")
print(f"Validation errors: {len(plan.errors)}")

# Apply plan if validation passed
if not plan.errors:
    context.apply(plan)

Plan with Skip Tests for Quick Validation

from pathlib import Path
from sqlmesh.core.context import Context

context = Context(paths=Path("/dbt/project"))

# Quick validation without running tests
plan = context.plan(
    environment="dev",
    skip_tests=True,  # Skip tests for faster validation
    no_prompts=True   # Non-interactive mode
)

# Check for structural errors
if plan.errors:
    print("Plan validation failed:")
    for error in plan.errors:
        print(f"  - {error}")
else:
    print("Plan structure validated successfully")
    print(f"Models to backfill: {len(plan.missing_intervals)}")

Selective Model Planning

from pathlib import Path
from sqlmesh.core.context import Context

context = Context(paths=Path("/dbt/project"))

# Plan only staging models
plan = context.plan(
    environment="dev",
    select_models=["tag:staging"],  # dbt-style tag selection
    skip_tests=False
)

print(f"Staging models in plan: {len(plan.snapshots)}")

# Or use model name patterns
plan = context.plan(
    environment="dev",
    select_models=["stg_*", "staging.*"],  # glob patterns
    skip_tests=False
)

Running Tests Separately

from pathlib import Path
from sqlmesh.core.context import Context
from sqlmesh.core.test import Verbosity

context = Context(paths=Path("/dbt/project"))

# Run all tests
result = context.test(verbosity=Verbosity.VERBOSE)

print(f"Tests run: {result.tests_run}")
print(f"Failures: {result.failures}")
print(f"Errors: {result.errors}")

# Check if all tests passed
if result.wasSuccessful():
    print("All tests passed!")
else:
    print("Some tests failed:")
    for failure in result.failures:
        print(f"  - {failure[0]}: {failure[1]}")

Pattern-Based Test Execution

from pathlib import Path
from sqlmesh.core.context import Context

context = Context(paths=Path("/dbt/project"))

# Run tests matching specific patterns
result = context.test(match_patterns=["test_staging_*", "test_transform_*"])

print(f"Matched tests: {result.tests_run}")

# Run specific named tests
result = context.test(tests=["test_customer_data", "test_order_totals"])

if result.wasSuccessful():
    print("Selected tests passed!")

Full Validation Workflow

from pathlib import Path
from sqlmesh.core.context import Context

# Complete dbt migration validation workflow
context = Context(paths=Path("/dbt/project"))

print("Step 1: Running unit tests...")
test_result = context.test()
if not test_result.wasSuccessful():
    print(f"Tests failed: {test_result.failures}")
    exit(1)

print("Step 2: Generating plan...")
plan = context.plan(
    environment="dev",
    skip_tests=False,  # Run tests again as part of plan
    no_prompts=True
)

if plan.errors:
    print(f"Plan validation failed: {plan.errors}")
    exit(1)

print("Step 3: Applying plan...")
context.apply(plan)

print("Migration validated and applied successfully!")

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Dbt_Validation_And_Testing

• Environment:TobikoData_Sqlmesh_Dbt_Compatibility