Support config: disabled for unit tests. (#9109)

Also, when a model is disabled, disable the corresponding unit tests automatically. (#10540)

core/dbt/artifacts/resources/v1/unit_test_definition.py

@@ -20,6 +20,7 @@ class UnitTestConfig(BaseConfig):
         default_factory=dict,
         metadata=MergeBehavior.Update.meta(),
     )
+    enabled: bool = True
 
 
 class UnitTestFormat(StrEnum):

core/dbt/contracts/graph/manifest.py

@@ -1671,6 +1671,8 @@ def add_disabled(self, source_file: AnySourceFile, node: GraphMemberNode, test_f
                 source_file.semantic_models.append(node.unique_id)
             if isinstance(node, Exposure):
                 source_file.exposures.append(node.unique_id)
+            if isinstance(node, UnitTestDefinition):
+                source_file.unit_tests.append(node.unique_id)
         elif isinstance(source_file, FixtureSourceFile):
             pass
         else:

core/dbt/graph/selector.py

@@ -173,7 +173,8 @@ def _is_graph_member(self, unique_id: UniqueId) -> bool:
             semantic_model = self.manifest.semantic_models[unique_id]
             return semantic_model.config.enabled
         elif unique_id in self.manifest.unit_tests:
-            return True
+            unit_test = self.manifest.unit_tests[unique_id]
+            return unit_test.config.enabled
         elif unique_id in self.manifest.saved_queries:
             saved_query = self.manifest.saved_queries[unique_id]
             return saved_query.config.enabled

core/dbt/parser/unit_tests.py

@@ -52,6 +52,8 @@ def load(self) -> Manifest:
         for unique_id in self.selected:
             if unique_id in self.manifest.unit_tests:
                 unit_test_case: UnitTestDefinition = self.manifest.unit_tests[unique_id]
+                if not unit_test_case.config.enabled:
+                    continue
                 self.parse_unit_test_case(unit_test_case)
         return self.unit_test_manifest
 
@@ -295,7 +297,10 @@ def parse(self) -> ParseResult:
             # for calculating state:modified
             unit_test_definition.build_unit_test_checksum()
             assert isinstance(self.yaml.file, SchemaSourceFile)
-            self.manifest.add_unit_test(self.yaml.file, unit_test_definition)
+            if unit_test_config.enabled:
+                self.manifest.add_unit_test(self.yaml.file, unit_test_definition)
+            else:
+                self.manifest.add_disabled(self.yaml.file, unit_test_definition)
 
         return ParseResult()
 
@@ -505,6 +510,16 @@ def process_models_for_unit_test(
     # The UnitTestDefinition should only have one "depends_on" at this point,
     # the one that's found by the "model" field.
     target_model_id = unit_test_def.depends_on.nodes[0]
+    if target_model_id not in manifest.nodes:
+        if target_model_id in manifest.disabled:
+            # The model is disabled, so we don't need to do anything (#10540)
+            return
+        else:
+            # If we've reached here and the model is not disabled, throw an error
+            raise ParsingError(
+                f"Unit test '{unit_test_def.name}' references a model that does not exist: {target_model_id}"
+            )
+
     target_model = manifest.nodes[target_model_id]
     assert isinstance(target_model, ModelNode)
 

tests/unit/parser/test_unit_tests.py

@@ -135,6 +135,19 @@
         - {"id":  , "col1": "e"}
 """
 
+UNIT_TEST_DISABLED = """
+unit_tests:
+    - name: test_my_model_disabled
+      model: my_model
+      description: "this unit test is disabled"
+      config:
+        enabled: false
+      given: []
+      expect:
+        rows:
+          - {a: 1}
+"""
+
 
 class UnitTestParserTest(SchemaParserTest):
     def setUp(self):
@@ -198,6 +211,22 @@ def test_unit_test_config(self):
         self.assertEqual(sorted(unit_test.config.tags), sorted(["schema_tag", "project_tag"]))
         self.assertEqual(unit_test.config.meta, {"meta_key": "meta_value", "meta_jinja_key": "2"})
 
+    def test_unit_test_disabled(self):
+        block = self.yaml_block_for(UNIT_TEST_DISABLED, "test_my_model.yml")
+        self.root_project_config.unit_tests = {
+            "snowplow": {"my_model": {"+tags": ["project_tag"]}}
+        }
+
+        UnitTestParser(self.parser, block).parse()
+
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=1, unit_tests=0, disabled=1)
+        unit_test_disabled_list = self.parser.manifest.disabled[
+            "unit_test.snowplow.my_model.test_my_model_disabled"
+        ]
+        self.assertEqual(len(unit_test_disabled_list), 1)
+        unit_test_disabled = unit_test_disabled_list[0]
+        self.assertEqual(unit_test_disabled.config.enabled, False)
+
     def test_unit_test_versioned_model(self):
         block = self.yaml_block_for(UNIT_TEST_VERSIONED_MODEL_SOURCE, "test_my_model.yml")
         my_model_versioned_node = MockNode(