From a433f15d6fd20f5fd713268040eba09727db4a2c Mon Sep 17 00:00:00 2001
From: David Spulak <david.spulak@maginteractive.se>
Date: Tue, 17 Oct 2023 23:42:32 +0200
Subject: [PATCH] Fix custom selector when select has a subset of tags of the
 models' tags (#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.
---
 cosmos/dbt/selector.py     |  2 +-
 tests/dbt/test_selector.py | 34 ++++++++++++++++++++++++----------
 2 files changed, 25 insertions(+), 11 deletions(-)

diff --git a/cosmos/dbt/selector.py b/cosmos/dbt/selector.py
index e10f5b9b2..926cc6b1a 100644
--- a/cosmos/dbt/selector.py
+++ b/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}
diff --git a/tests/dbt/test_selector.py b/tests/dbt/test_selector.py
index 8e3fc8c61..9f6071a20 100644
--- a/tests/dbt/test_selector.py
+++ b/tests/dbt/test_selector.py
@@ -45,7 +45,7 @@ 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",
@@ -53,8 +53,8 @@ def test_is_empty_config(selector_config, paths, tags, config, other, expected):
     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",
@@ -62,8 +62,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/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():