Merge pull request #2 from AidanHarveyNelson/feature/add-complete-integration-tests

AidanHarveyNelson · web-flow · commit 1c2f80e74519 · 2026-03-22T09:58:35.000+11:00
Stashing integration tests
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -1,8 +1,8 @@
 # **what?**
-# Run tests for dbt-external-tables against supported adapters
+# Run tests for dbt-streams-tasks against supported adapters
 
 # **why?**
-# To ensure that dbt-external-tables works as expected with all supported adapters
+# To ensure that dbt-streams-tasks works as expected with all supported adapters
 
 # **when?**
 # On every PR, and every push to main and when manually triggered
@@ -20,34 +20,13 @@ jobs:
   run-tests:
       uses: dbt-labs/dbt-package-testing/.github/workflows/run_tox.yml@v1
       with:
-        # redshift
-        REDSHIFT_HOST: ${{ vars.REDSHIFT_HOST }}
-        REDSHIFT_USER: ${{ vars.REDSHIFT_USER }}
-        REDSHIFT_PORT: ${{ vars.REDSHIFT_PORT }}
-        REDSHIFT_DATABASE: ${{ vars.REDSHIFT_DATABASE }}
-        REDSHIFT_SCHEMA: "integration_tests_redshift_${{ github.run_number }}"
         # snowflake
         SNOWFLAKE_USER: ${{ vars.SNOWFLAKE_USER }}
         SNOWFLAKE_WAREHOUSE: ${{ vars.SNOWFLAKE_WAREHOUSE }}
         SNOWFLAKE_ROLE: ${{ vars.SNOWFLAKE_ROLE }}
         SNOWFLAKE_DATABASE: ${{ vars.SNOWFLAKE_DATABASE }}
         SNOWFLAKE_SCHEMA: "integration_tests_snowflake_${{ github.run_number }}"
-        # bigquery
-        BIGQUERY_PROJECT: ${{ vars.BIGQUERY_PROJECT }}
-        BIGQUERY_SCHEMA: "integration_tests_bigquery_${{ github.run_number }}"
-        # synapse
-        # temporarily removed until we can get the cluster hooked up to the blob correctly
-        # SYNAPSE_DRIVER: ${{ vars.SYNAPSE_DRIVER }}
-        # SYNAPSE_HOST: ${{ vars.SYNAPSE_HOST }}
-        # SYNAPSE_PORT: ${{ vars.SYNAPSE_PORT }}
-        # SYNAPSE_DATABASE: ${{ vars.SYNAPSE_DATABASE }}
-        # SYNAPSE_AUTHENTICATION: ${{ vars.SYNAPSE_AUTHENTICATION }}
-        # SYNAPSE_TENANT_ID: ${{ vars.SYNAPSE_TENANT_ID }}
-        # SYNAPSE_CLIENT_ID: ${{ vars.SYNAPSE_CLIENT_ID }}
 
       secrets:
-        DBT_ENV_SECRET_REDSHIFT_PASS: ${{ secrets.DBT_ENV_SECRET_REDSHIFT_PASS }}
         SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
         DBT_ENV_SECRET_SNOWFLAKE_PASS: ${{ secrets.DBT_ENV_SECRET_SNOWFLAKE_PASS }}
-        BIGQUERY_KEYFILE_JSON: ${{ secrets.BIGQUERY_KEYFILE_JSON }}
-        DBT_ENV_SECRET_SYNAPSE_CLIENT_SECRET: ${{ secrets.DBT_ENV_SECRET_SYNAPSE_CLIENT_SECRET }}
diff --git a/.github/workflows/integration_tests.yml b/.github/workflows/integration_tests.yml
diff --git a/CHANGLOG.md b/CHANGLOG.md
@@ -1 +1,10 @@
 # Changelog
+
+## v0.0.2 (TBA)
+
+- Feature: Expanded integration tests to cover all use cases
+
+
+## v0.0.1 (2025-09-03)
+
+- Feature: Initial implementation logic
diff --git a/dbt_project.yml b/dbt_project.yml
@@ -1,10 +1,5 @@
-name: 'audit_helper'
-version: '0.4.0'
+name: 'dbt_snowflake_streams_tasks'
+version: '0.0.1'
 config-version: 2
-
-require-dbt-version: [">=1.2.0", "<2.0.0"]
-
-target-path: "target"
-clean-targets: ["target", "dbt_packages"]
+require-dbt-version: [">=1.0.0", "<2.0.0"]
 macro-paths: ["macros"]
-log-path: "logs"
diff --git a/integration_tests/macros/prep_structure.sql b/integration_tests/macros/prep_structure.sql
@@ -16,12 +16,27 @@
             create or replace table {{ table }} (
                 id integer,
                 data string
-            );
+            )
+            CHANGE_TRACKING = TRUE
+            ;
             commit;
         {% endset %}
 
         {% do log('Creating testing table ' ~ table, info = true) %}
         {% do run_query(create_testing_table_sql) %}
     {% endfor %}
 
+
+    {% set test_table  = [target.database, target.schema, 'testing_ref']|join('.') %}
+
+    {% set insert_data_sql %}
+        begin;
+        insert into {{test_table}} values (1, 'test data');
+        commit;
+    {% endset %}
+
+    {% do log('Inserting data into testing table ' ~ test_table, info = true) %}
+    {% do run_query(insert_data_sql) %}
+
+
 {% endmacro %}
diff --git a/integration_tests/models/customers/_models.yml b/integration_tests/models/customers/_models.yml
@@ -17,6 +17,24 @@ models:
         name: "transform_from_stream_test"
         description: "Task for the customers table"
         when: "SYSTEM$STREAM_HAS_DATA('stream_from_source_test')"
-        schedule: "5 MINUTES"
+        schedule: "1 MINUTES"
     columns:
       - name: id
+  
+  - name: stream_from_ref
+    config:
+      materialized: stream_task
+      unique_key:
+        - id
+      stream:
+        name: "stream_from_ref_test"
+        description: "Stream for the customers table"
+        tags:
+          - customers
+        copy_grants: True
+        append_only: False
+      task:
+        name: "transform_from_ref_test"
+        description: "Task for the customers table"
+        when: "SYSTEM$STREAM_HAS_DATA('stream_from_ref_test')"
+        schedule: "1 MINUTES"
diff --git a/integration_tests/models/customers/customers.sql b/integration_tests/models/customers/customers.sql
diff --git a/integration_tests/models/customers/stream_from_ref.sql b/integration_tests/models/customers/stream_from_ref.sql
@@ -0,0 +1,2 @@
+select *
+from {{ dbt_snowflake_streams_tasks.stream_ref('view_on_source') }}
diff --git a/integration_tests/models/customers/view_on_source.sql b/integration_tests/models/customers/view_on_source.sql
@@ -0,0 +1,8 @@
+{{
+  config(
+    materialized = 'view',
+    )
+}}
+
+select *
+from {{ source('customers', 'testing_ref') }}
diff --git a/macros/helpers/create_stream.sql b/macros/helpers/create_stream.sql
@@ -1,15 +1,30 @@
 {%- macro create_stream(source_table) -%}
     {%- do log('Source Table is: ' ~ source_table, info = true) -%}
     {%- set name = config.require('stream').name -%}
-    {%- do log('Creating Stream: ' ~ name, info = true) -%}
-    {%- set args_sql -%}
-        APPEND_ONLY = {{ config.get('stream').append_only|upper or 'TRUE' }}
-        COMMENT = '{{ config.get('stream').description or '' }}'
-        {{- "COPY GRANTS" ~ config.get('stream').copy_grants if config.get('stream').copy_grants else '' -}}
-    {%- endset -%}
+
     {%- set run_sql -%}
-        CREATE OR REPLACE STREAM {{ name }} ON TABLE {{ source_table }}
-        {{ args_sql }}
+        CREATE OR REPLACE STREAM {{ name }}
+        {{ "COPY GRANTS" if config.get('stream').copy_grants else '' }}
+        {# To-Do Add Tag Config#}
     {%- endset -%}
+
+    {% if source_table.is_view %}
+        {%- set run_sql -%}
+            {{ run_sql ~ " ON VIEW " ~ source_table }}
+            COMMENT = '{{ config.get('stream').description or '' }}'
+            APPEND_ONLY = {{ config.get('stream').append_only|upper or 'TRUE' }}
+            SHOW_INITIAL_ROWS = {{ config.get('stream').show_initial_rows|upper or 'TRUE' }}
+        {%- endset -%}
+    {% elif source_table.is_table %}
+        {%- set run_sql -%}
+            {{ run_sql ~ " ON TABLE " ~ source_table }}
+            COMMENT = '{{ config.get('stream').description or '' }}'
+            APPEND_ONLY = {{ config.get('stream').append_only|upper or 'TRUE' }}
+            SHOW_INITIAL_ROWS = {{ config.get('stream').show_initial_rows|upper or 'TRUE' }}
+        {%- endset -%}
+    {%- else -%}
+        {%- do exceptions.raise_compiler_error("Invalid materialization type provided to stream macro. Use 'view' or 'table'. Object passed in is " ~ source_table ~ " with type: " ~ source_table.type) -%}
+    {%- endif -%}
+    {%- do log('Creating Stream: ' ~ name, info = true) -%}
     {{ run_sql }}
 {%- endmacro -%}
diff --git a/macros/helpers/create_table.sql b/macros/helpers/create_table.sql
@@ -4,11 +4,11 @@
         {%- if not is_new -%}
             {% set missing_columns = adapter.get_missing_columns(target_table, tmp_relation) %}
             {% do log('Missing columns are ' ~ missing_columns, info=true)%}
-            {% if missing_columns | length > 0 %}
+            {% if missing_columns %}
                 alter table {{ target_table }}
-                {%- for col in adapter.get_missing_columns(target_table, tmp_relation) -%}
+                {% for col in missing_columns %}
                     add column "{{- col.name -}}" {{- col.data_type -}}{{- ";" if loop.last else "\n" -}}
-                {%- endfor -%}
+                {% endfor %}
             {% endif%}
         {%- else -%}
             create table {{ target_table }} as {{ sql }};
diff --git a/macros/materializations/stream_task.sql b/macros/materializations/stream_task.sql
@@ -32,20 +32,26 @@
 -- Run Pre-Hooks
 {{- run_hooks(pre_hooks) -}}
 
-{%- do log('Build SQL: ' ~ build_sql, info = true) -%}
-
 {%- call statement('main') -%}
     {%- do log('Creating Objects for Relation: ' ~ target_relation ~ '\n' ~ sql, info = true) -%}
     {%- set stream_target_config = config.get('src_table') -%}
     {%- do log('Logging Stream Target Config: ' ~ stream_target_config, info = true) -%}
-    {%- set stream_target = source(stream_target_config.source_name, stream_target_config.model_name) -%}
-    {%- do log('Logging Stream Target: ' ~ stream_target, info = true) -%}
+
+    {%- if stream_target_config.model_type == 'source' -%}
+        {%- set stream_source = load_relation(source(stream_target_config.source_name, stream_target_config.model_name)) -%}
+    {%- else -%}
+        {%- set stream_source = load_relation(ref(stream_target_config.model_name)) -%}
+    {%- endif -%}
+
+    {%- set stream_name = config.require('stream').name -%}
+
+    {%- do log('Logging Stream Source: ' ~ stream_source, info = true) -%}
     {%- call statement('create_tmp_relation') -%}
         {{- snowflake__create_view_as_with_temp_flag(tmp_relation, sql, True) -}}
     {%- endcall -%}
-    {%- set stream_query = dbt_snowflake_streams_tasks.create_stream(stream_target) -%}
+    {%- set stream_query = dbt_snowflake_streams_tasks.create_stream(stream_source) -%}
     {%- do log('Logging Stream Query: ' ~ stream_query, info = true) -%}
-    {%- set task_query = dbt_snowflake_streams_tasks.create_task(stream_target, target_relation, sql, tmp_relation, unique_key) -%}
+    {%- set task_query = dbt_snowflake_streams_tasks.create_task(stream_name, target_relation, sql, tmp_relation, unique_key) -%}
     {%- do log('Logging Task Query: ' ~ task_query, info = true) -%}
     {%- set table_query = dbt_snowflake_streams_tasks.create_table(target_relation, sql, tmp_relation, is_new) -%}
     {%- do log('Logging Table Query: ' ~ table_query, info = true) -%}
diff --git a/packages.yml b/packages.yml
diff --git a/pyproject.toml b/pyproject.toml
@@ -5,10 +5,7 @@ description = "A DBT Plugin that allows creating models using streams and tasks"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    "dbt-bigquery>=1.9.0",
     "dbt-core>=1.9.1",
-    "dbt-redshift>=1.9.0",
     "dbt-snowflake>=1.9.0",
-    "dbt-synapse>=1.8.2",
     "tox>=4.23.2",
 ]
diff --git a/run_test.sh b/run_test.sh
@@ -1,38 +1,5 @@
 #!/bin/bash
 
-# echo "Setting up virtual environment for dbt-$1"
-# VENV="venv/bin/activate"
-
-# if [[ ! -f $VENV ]]; then
-#     python3.8 -m venv venv
-#     . $VENV
-#     pip install --upgrade pip setuptools
-#     if [ $1 == 'databricks' ]
-#     then
-#         echo "Installing dbt-spark"
-#         pip install dbt-spark[ODBC] --upgrade --pre
-#     elif [ $1 == 'azuresql' ]
-#     then
-#         echo "Installing dbt-sqlserver"
-#         pip install dbt-sqlserver --upgrade --pre
-#     else
-#         echo "Installing dbt-$1"
-#         pip install dbt-$1 --upgrade --pre
-#         # remove the protobuf installation when all the dbt-provider packaged are updated with dbt core 1.7.9
-#         pip install protobuf==4.25.3
-#     fi
-# fi
-
-# . $VENV
-# echo "Changing working directory: integration_tests"
-# cd integration_tests
-
-# if [[ ! -e ~/.dbt/profiles.yml ]]; then
-#     echo "Copying sample profile"
-#     mkdir -p ~/.dbt
-#     cp ci/sample.profiles.yml ~/.dbt/profiles.yml
-# fi
-
 cd integration_tests
 
 echo "Starting integration tests"
@@ -45,4 +12,5 @@ echo "Running dbt run-operation prep_structure"
 uv run dbt run-operation prep_structure
 uv run dbt run
 uv run dbt run-operation insert_data
+sleep 120
 uv run dbt test

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+select *`
	`2`	`+from {{ dbt_snowflake_streams_tasks.stream_ref('view_on_source') }}`