Enable use of env vars for --resource-types and --exclude-resource-ty…

…pes (#9779)

.changes/unreleased/Features-20240312-140407.yaml

@@ -1,5 +1,5 @@
 kind: Features
-body: Allow excluding resource types for build, list, and clone commands
+body: Allow excluding resource types for build, list, and clone commands, and provide env vars
 time: 2024-03-12T14:04:07.086017-04:00
 custom:
   Author: gshank

core/dbt/cli/option_types.py

@@ -80,7 +80,10 @@ class ChoiceTuple(Choice):
     name = "CHOICE_TUPLE"
 
     def convert(self, value, param, ctx):
-        for value_item in value:
-            super().convert(value_item, param, ctx)
+        if not isinstance(value, str):
+            for value_item in value:
+                super().convert(value_item, param, ctx)
+        else:
+            super().convert(value, param, ctx)
 
         return value

core/dbt/task/base.py

@@ -500,7 +500,7 @@ def resource_types_from_args(
             arg_resource_types.remove("default")
             arg_resource_types.update(default_resource_values)
         # Convert to a set of NodeTypes now that the non-NodeType strings are gone
-        resource_types = set([NodeType(rt) for rt in arg_resource_types])
+        resource_types = set([NodeType(rt) for rt in list(arg_resource_types)])
 
     if args.exclude_resource_types:
         # Convert from a list of strings to a set of NodeTypes

tests/functional/list/test_list.py

@@ -703,6 +703,34 @@ def expect_resource_type_multiple(self):
             "test.outer",
         }
 
+    def expect_resource_type_env_var(self):
+        """Expect selected resources when --resource-type given multiple times"""
+        os.environ["DBT_RESOURCE_TYPES"] = "test model"
+        results = self.run_dbt_ls()
+        assert set(results) == {
+            "test.ephemeral",
+            "test.incremental",
+            "test.not_null_outer_id",
+            "test.outer",
+            "test.sub.inner",
+            "test.metricflow_time_spine",
+            "test.t",
+            "test.unique_outer_id",
+        }
+        del os.environ["DBT_RESOURCE_TYPES"]
+        os.environ[
+            "DBT_EXCLUDE_RESOURCE_TYPES"
+        ] = "test saved_query metric source semantic_model snapshot seed"
+        results = self.run_dbt_ls()
+        assert set(results) == {
+            "test.ephemeral",
+            "test.incremental",
+            "test.outer",
+            "test.sub.inner",
+            "test.metricflow_time_spine",
+        }
+        del os.environ["DBT_EXCLUDE_RESOURCE_TYPES"]
+
     def expect_selected_keys(self, project):
         """Expect selected fields of the the selected model"""
         expectations = [
@@ -798,6 +826,7 @@ def test_ls(self, project):
         self.expect_test_output()
         self.expect_select()
         self.expect_resource_type_multiple()
+        self.expect_resource_type_env_var()
         self.expect_all_output()
         self.expect_selected_keys(project)
 

tests/functional/unit_testing/test_unit_testing.py

@@ -1,4 +1,5 @@
 import pytest
+import os
 from unittest import mock
 from dbt.tests.util import (
     run_dbt,
@@ -69,6 +70,13 @@ def test_basic(self, project):
         )
         assert len(results) == 1
 
+        # Exclude unit tests with environment variable
+        os.environ["DBT_EXCLUDE_RESOURCE_TYPES"] = "unit_test"
+        results = run_dbt(["build", "--select", "my_model"], expect_pass=True)
+        assert len(results) == 1
+
+        del os.environ["DBT_EXCLUDE_RESOURCE_TYPES"]
+
         # Test select by test name
         results = run_dbt(["test", "--select", "test_name:test_my_model_string_concat"])
         assert len(results) == 1

tests/unit/test_cli_flags.py

@@ -105,6 +105,13 @@ def test_anonymous_usage_state(
         flags = Flags(run_context)
         assert flags.SEND_ANONYMOUS_USAGE_STATS == expected_anonymous_usage_stats
 
+    def test_resource_types(self, monkeypatch):
+        monkeypatch.setenv("DBT_RESOURCE_TYPES", "model")
+        build_context = self.make_dbt_context("build", ["build"])
+        build_context.params["resource_types"] = ("unit_test",)
+        flags = Flags(build_context)
+        assert flags.resource_types == ("unit_test",)
+
     def test_empty_project_flags_uses_default(self, run_context, project_flags):
         flags = Flags(run_context, project_flags)
         assert flags.USE_COLORS == run_context.params["use_colors"]