Unit test selection method (#10102)

.changes/unreleased/Features-20240507-162717.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: 'Add unit_test: selection method'
+time: 2024-05-07T16:27:17.047585-04:00
+custom:
+  Author: michelleark
+  Issue: "10053"

core/dbt/graph/selector_methods.py

@@ -63,6 +63,7 @@ class MethodName(StrEnum):
     Version = "version"
     SemanticModel = "semantic_model"
     SavedQuery = "saved_query"
+    UnitTest = "unit_test"
 
 
 def is_selected_node(fqn: List[str], node_selector: str, is_versioned: bool) -> bool:
@@ -425,6 +426,31 @@ def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[Uniqu
             yield unique_id
 
 
+class UnitTestSelectorMethod(SelectorMethod):
+    def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[UniqueId]:
+        parts = selector.split(".")
+        target_package = SELECTOR_GLOB
+        if len(parts) == 1:
+            target_name = parts[0]
+        elif len(parts) == 2:
+            target_package, target_name = parts
+        else:
+            msg = (
+                'Invalid unit test selector value "{}". Saved queries must be of '
+                "the form ${{unit_test_name}} or "
+                "${{unit_test_package_name.unit_test_name}}"
+            ).format(selector)
+            raise DbtRuntimeError(msg)
+
+        for unique_id, node in self.unit_tests(included_nodes):
+            if not fnmatch(node.package_name, target_package):
+                continue
+            if not fnmatch(node.name, target_name):
+                continue
+
+            yield unique_id
+
+
 class PathSelectorMethod(SelectorMethod):
     def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[UniqueId]:
         """Yields nodes from included that match the given path."""
@@ -883,6 +909,7 @@ class MethodManager:
         MethodName.Version: VersionSelectorMethod,
         MethodName.SemanticModel: SemanticModelSelectorMethod,
         MethodName.SavedQuery: SavedQuerySelectorMethod,
+        MethodName.UnitTest: UnitTestSelectorMethod,
     }
 
     def __init__(

tests/unit/test_graph_selector_methods.py → tests/unit/graph/test_selector_methods.py

@@ -27,8 +27,10 @@
     TagSelectorMethod,
     TestNameSelectorMethod,
     TestTypeSelectorMethod,
+    UnitTestSelectorMethod,
     VersionSelectorMethod,
 )
+from tests.unit.utils import replace_config
 from tests.unit.utils.manifest import (
     make_exposure,
     make_group,
@@ -38,10 +40,9 @@
     make_saved_query,
     make_seed,
     make_semantic_model,
+    make_unit_test,
 )
 
-from .utils import replace_config
-
 
 def search_manifest_using_method(manifest, method, selection):
     selected = method.search(
@@ -586,6 +587,24 @@ def test_select_saved_query_by_tag(manifest: Manifest) -> None:
     search_manifest_using_method(manifest, method, "any_tag")
 
 
+def test_select_unit_test(manifest: Manifest) -> None:
+    test_model = make_model("test", "my_model", "select 1 as id")
+    unit_test = make_unit_test("test", "my_unit_test", test_model)
+    manifest.unit_tests[unit_test.unique_id] = unit_test
+    methods = MethodManager(manifest, None)
+    method = methods.get_method("unit_test", [])
+
+    assert isinstance(method, UnitTestSelectorMethod)
+    assert not search_manifest_using_method(manifest, method, "not_test_unit_test")
+    assert search_manifest_using_method(manifest, method, "*nit_test") == {unit_test.search_name}
+    assert search_manifest_using_method(manifest, method, "test.my_unit_test") == {
+        unit_test.search_name
+    }
+    assert search_manifest_using_method(manifest, method, "my_unit_test") == {
+        unit_test.search_name
+    }
+
+
 @pytest.fixture
 def previous_state(manifest):
     writable = copy.deepcopy(manifest).writable_manifest()