Fix custom selector when select has a subset of tags of the models' t…

…ags (#606)

Before, when using Cosmos custom selector, it expected an exact match of all *tags*.

Example: `model1` with tags `tag1` and `tag2` would only be
selected/excluded if the select/exclude statement in the `RenderConfig`
was set to `select=["tag:tag1,tag:tag2"]`. It would not be selected if the `RenderConfig` had
`select=["tag:tag1"]` or `select=["tag:tag2"]` only.

(cherry picked from commit a433f15)

cosmos/dbt/selector.py

@@ -106,7 +106,7 @@ def should_include_node(node_id: str, node: DbtNode) -> bool:
         visited_nodes.add(node_id)
 
         if config.tags:
-            if not (set(config.tags) == set(node.tags)):
+            if not (set(config.tags) <= set(node.tags)):
                 return False
 
         node_config = {key: value for key, value in node.config.items() if key in SUPPORTED_CONFIG}

tests/dbt/test_selector.py

@@ -45,25 +45,25 @@ def test_is_empty_config(selector_config, paths, tags, config, other, expected):
     depends_on=[],
     file_path=SAMPLE_PROJ_PATH / "gen1/models/grandparent.sql",
     tags=["has_child"],
-    config={"materialized": "view"},
+    config={"materialized": "view", "tags": ["has_child"]},
 )
 parent_node = DbtNode(
     name="parent",
     unique_id="parent",
     resource_type=DbtResourceType.MODEL,
     depends_on=["grandparent"],
     file_path=SAMPLE_PROJ_PATH / "gen2/models/parent.sql",
-    tags=["has_child"],
-    config={"materialized": "view"},
+    tags=["has_child", "is_child"],
+    config={"materialized": "view", "tags": ["has_child", "is_child"]},
 )
 child_node = DbtNode(
     name="child",
     unique_id="child",
     resource_type=DbtResourceType.MODEL,
     depends_on=["parent"],
     file_path=SAMPLE_PROJ_PATH / "gen3/models/child.sql",
-    tags=["nightly"],
-    config={"materialized": "table", "tags": ["is_child"]},
+    tags=["nightly", "is_child"],
+    config={"materialized": "table", "tags": ["nightly", "is_child"]},
 )
 
 grandchild_1_test_node = DbtNode(
@@ -72,8 +72,8 @@ def test_is_empty_config(selector_config, paths, tags, config, other, expected):
     resource_type=DbtResourceType.MODEL,
     depends_on=["parent"],
     file_path=SAMPLE_PROJ_PATH / "gen3/models/grandchild_1.sql",
-    tags=["nightly"],
-    config={"materialized": "table", "tags": ["deprecated", "test"]},
+    tags=["nightly", "deprecated", "test"],
+    config={"materialized": "table", "tags": ["nightly", "deprecated", "test"]},
 )
 
 grandchild_2_test_node = DbtNode(
@@ -82,8 +82,8 @@ def test_is_empty_config(selector_config, paths, tags, config, other, expected):
     resource_type=DbtResourceType.MODEL,
     depends_on=["parent"],
     file_path=SAMPLE_PROJ_PATH / "gen3/models/grandchild_2.sql",
-    tags=["nightly"],
-    config={"materialized": "table", "tags": ["deprecated", "test2"]},
+    tags=["nightly", "deprecated", "test2"],
+    config={"materialized": "table", "tags": ["nightly", "deprecated", "test2"]},
 )
 
 sample_nodes = {
@@ -114,6 +114,7 @@ def test_select_nodes_by_select_config():
 def test_select_nodes_by_select_config_tag():
     selected = select_nodes(project_dir=SAMPLE_PROJ_PATH, nodes=sample_nodes, select=["config.tags:is_child"])
     expected = {
+        parent_node.unique_id: parent_node,
         child_node.unique_id: child_node,
     }
     assert selected == expected
@@ -146,11 +147,24 @@ def test_select_nodes_by_select_union_config_test_tags():
     assert selected == expected
 
 
+def test_select_nodes_by_select_intersection_tag():
+    selected = select_nodes(
+        project_dir=SAMPLE_PROJ_PATH, nodes=sample_nodes, select=["tag:is_child,config.materialized:view"]
+    )
+    expected = {
+        parent_node.unique_id: parent_node,
+    }
+    assert selected == expected
+
+
 def test_select_nodes_by_select_intersection_config_tag():
     selected = select_nodes(
         project_dir=SAMPLE_PROJ_PATH, nodes=sample_nodes, select=["config.tags:is_child,config.materialized:view"]
     )
-    assert selected == {}
+    expected = {
+        parent_node.unique_id: parent_node,
+    }
+    assert selected == expected
 
 
 def test_select_nodes_by_select_path():