Dynamic Unit Testing in dbt

profiles.yml

  my_data_warehouse: #project name
  target: dev #default target
  outputs:
    dev:
      ...
    ci:
      type: snowflake
      account: 
      user: 
      password: 
      role: TRANSFORMER
      database: ANALYTICS_DEV_CI
      warehouse: TRANSFORMING
      schema: ANALYTICS_DEV_CI
      threads: 1
      client_session_keep_alive: False
      query_tag: dbt

schema.yml

version: 2
models:

  - name: fact_model
    description: "Your description goes here"
    tests:
      ...
      - my_data_warehouse.equality:
          compare_model: ref('expected_output_data')
          compare_columns: 
            - account_id,
            - acount_registration,
            - account_status
          env: ci

stg_model.sql

with source as (

    {% if target.name == 'ci' %}

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

    {% else %}

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

    {% endif %}
    
),

test_equality.sql

{% macro test_equality(model, compare_columns=None) %} -- macro definition

-- get compare_model input parameter
{% set compare_model = kwargs.get('compare_model') %} 
-- get env input parameter
{% set env = kwargs.get('env') %} 

{%- if target.name == env -%} -- check if env input parameter matches the current environment

    {#-
    If the compare_cols arg is provided, we can run this test without querying the
    information schema
    -#}

    {%- if not compare_columns -%}
        {%- set compare_columns = adapter.get_columns_in_relation(model) | map(attribute='quoted') -%}
    {%- endif -%}

    {% set compare_cols_csv = compare_columns | join(', ') %}

    select *
     from (
            (
                select {{compare_cols_csv}}
                from {{ model }}
                except
                select {{compare_cols_csv}}
                from {{ compare_model }} 
            )  
            union
            (
                select {{compare_cols_csv}}
                from {{ compare_model }}
                except
                select {{compare_cols_csv}}
                from {{ model }} 
            )
          ) tmp

{%- else -%}

    -- if no input or different env return nothing (true)
    select top 0 column1

{%- endif -%}

{% endmacro %}