test: [SNOW-1890085] add integration test for simple dbt workflow

Author: sfc-gh-jwilkowski

tests_integration/conftest.py

@@ -185,6 +185,19 @@ def invoke_with_connection(
     ) -> CommandResult:
         return self.invoke_with_config([*args, "-c", connection], **kwargs)
 
+    def invoke_passthrough_with_connection(
+        self,
+        args,
+        connection: str = "integration",
+        passthrough_args: Optional[list[str]] = None,
+        **kwargs,
+    ) -> CommandResult:
+        if passthrough_args is None:
+            passthrough_args = list()
+        return self.invoke_with_config(
+            [*args, "-c", connection, *passthrough_args], **kwargs
+        )
+
 
 @pytest.fixture
 def runner(test_snowcli_config_provider, default_username, resource_suffix):

tests_integration/test_data/projects/dbt_project/dbt_project.yml

@@ -0,0 +1,20 @@
+name: 'dbt_integration_project'
+version: '1.0.0'
+
+profile: 'dbt_integration_project'
+
+model-paths: ["models"]
+analysis-paths: ["analyses"]
+test-paths: ["tests"]
+seed-paths: ["seeds"]
+macro-paths: ["macros"]
+snapshot-paths: ["snapshots"]
+
+clean-targets:
+  - "target"
+  - "dbt_packages"
+
+models:
+  dbt_integration_project:
+    example:
+      +materialized: view

tests_integration/test_data/projects/dbt_project/models/example/my_first_dbt_model.sql

@@ -0,0 +1,11 @@
+{{ config(materialized='table') }}
+
+with source_data as (
+    select 1 as id
+    union all
+    select null as id
+)
+
+select *
+from source_data
+where id is not null

tests_integration/test_data/projects/dbt_project/models/example/my_second_dbt_model.sql

@@ -0,0 +1,3 @@
+select *
+from {{ ref('my_first_dbt_model') }}
+where id = 1

tests_integration/test_data/projects/dbt_project/models/example/schema.yml

@@ -0,0 +1,21 @@
+
+version: 2
+
+models:
+  - name: my_first_dbt_model
+    description: "A starter dbt model"
+    columns:
+      - name: id
+        description: "The primary key for this table"
+        data_tests:
+          - unique
+          - not_null
+
+  - name: my_second_dbt_model
+    description: "A starter dbt model"
+    columns:
+      - name: id
+        description: "The primary key for this table"
+        data_tests:
+          - unique
+          - not_null

tests_integration/test_dbt.py

@@ -0,0 +1,67 @@
+# Copyright (c) 2025 Snowflake Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import datetime
+
+import pytest
+
+
+@pytest.mark.integration
+@pytest.mark.qa_only
+def test_dbt_deploy(
+    runner,
+    snowflake_session,
+    test_database,
+    project_directory,
+):
+    with project_directory("dbt_project"):
+        # Given a local dbt project
+        ts = int(datetime.datetime.now().timestamp())
+        name = f"dbt_project_{ts}"
+
+        # When it's deployed
+        result = runner.invoke_with_connection_json(["dbt", "deploy", name])
+        assert result.exit_code == 0, result.output
+
+        # Then it can be listed
+        result = runner.invoke_with_connection_json(
+            [
+                "dbt",
+                "list",
+                "--like",
+                name,
+            ]
+        )
+        assert result.exit_code == 0, result.output
+        assert len(result.json) == 1
+        dbt_object = result.json[0]
+        assert dbt_object["name"].lower() == name.lower()
+
+        # And when dbt run gets called on it
+        result = runner.invoke_passthrough_with_connection(
+            args=[
+                "dbt",
+                "execute",
+            ],
+            passthrough_args=[name, "run"],
+        )
+
+        # Then is succeeds and models get populated according to expectations
+        assert result.exit_code == 0, result.output
+        assert "Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2" in result.output
+
+        result = runner.invoke_with_connection_json(
+            ["sql", "-q", "select count(*) as COUNT from my_second_dbt_model;"]
+        )
+        assert len(result.json) == 1, result.json
+        assert result.json[0]["COUNT"] == 1, result.json[0]