Add selected_resources to the Jinja context (#5001)

* Add selected_resources in the Jinja context

* Add tests for the Jinja variable selected_resources

* Add Changie entry for the addition of selected_resources

* Move variable to the ProviderContext

* Move selected_resources from ModelContext to ProviderContext

* Update unit tests for context to cater for the new selected_resources variable

* Move tests to a Class where tests are run after a dbt build

.changes/unreleased/Features-20220404-190439.yaml

@@ -0,0 +1,9 @@
+kind: Features
+body: Add a variable called selected_resources in the Jinja context containing a list
+  of all the resources matching the nodes for  the --select, --exclude and/or --selector
+  parameters.
+time: 2022-04-04T19:04:39.347479+02:00
+custom:
+  Author: b-per
+  Issue: "3471"
+  PR: "5001"

core/dbt/context/providers.py

@@ -62,6 +62,8 @@
 
 from dbt.utils import merge, AttrDict, MultiDict
 
+from dbt import selected_resources
+
 import agate
 
 
@@ -1143,6 +1145,15 @@ def env_var(self, var: str, default: Optional[str] = None) -> str:
             msg = f"Env var required but not provided: '{var}'"
             raise_parsing_error(msg)
 
+    @contextproperty
+    def selected_resources(self) -> List[str]:
+        """The `selected_resources` variable contains a list of the resources
+        selected based on the parameters provided to the dbt command.
+        Currently, is not populated for the command `run-operation` that
+        doesn't support `--select`.
+        """
+        return selected_resources.SELECTED_RESOURCES
+
 
 class MacroContext(ProviderContext):
     """Internally, macros can be executed like nodes, with some restrictions:

core/dbt/graph/selector.py

@@ -17,6 +17,8 @@
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.state import PreviousState
 
+from dbt import selected_resources
+
 
 def get_package_names(nodes):
     return set([node.split(".")[1] for node in nodes])
@@ -269,6 +271,7 @@ def get_graph_queue(self, spec: SelectionSpec) -> GraphQueue:
         dependecies.
         """
         selected_nodes = self.get_selected(spec)
+        selected_resources.set_selected_resources(selected_nodes)
         new_graph = self.full_graph.get_subset_graph(selected_nodes)
         # should we give a way here for consumers to mutate the graph?
         return GraphQueue(new_graph.graph, self.manifest, selected_nodes)

core/dbt/selected_resources.py

@@ -0,0 +1,6 @@
+SELECTED_RESOURCES = []
+
+
+def set_selected_resources(selected_resources):
+    global SELECTED_RESOURCES
+    SELECTED_RESOURCES = list(selected_resources)

test/unit/test_context.py

@@ -221,6 +221,7 @@ def assert_has_keys(
     'sql',
     'sql_now',
     'adapter_macro',
+    'selected_resources'
 }
 REQUIRED_MODEL_KEYS = REQUIRED_MACRO_KEYS | {'this'}
 MAYBE_KEYS = frozenset({'debug'})

tests/functional/selected_resources/fixtures.py

@@ -0,0 +1,35 @@
+on_run_start_macro_assert_selected_models_expected_list = """
+{% macro assert_selected_models_expected_list(expected_list) %}
+
+  {% if execute and (expected_list is not none) %}
+
+    {% set sorted_selected_resources = selected_resources | sort %}
+    {% set sorted_expected_list = expected_list | sort %}
+
+    {% if sorted_selected_resources != sorted_expected_list %}
+      {{ exceptions.raise_compiler_error("FAIL: sorted_selected_resources" ~ sorted_selected_resources ~ " is different from " ~ sorted_expected_list) }}
+    {% endif %}
+
+  {% endif %}
+
+{% endmacro %}
+"""
+
+
+my_model1 = """
+select 1 as id
+"""
+
+my_model2 = """
+select * from {{ ref('model1') }}
+"""
+
+my_snapshot = """
+{% snapshot cc_all_snapshot %}
+    {{ config(
+        check_cols='all', unique_key='id', strategy='check',
+        target_database=database, target_schema=schema
+    ) }}
+    select * from {{ ref('model2') }}
+{% endsnapshot %}
+"""

tests/functional/selected_resources/test_selected_resources.py

@@ -0,0 +1,104 @@
+import pytest
+from dbt.tests.util import run_dbt
+from tests.functional.selected_resources.fixtures import (
+    on_run_start_macro_assert_selected_models_expected_list,
+    my_model1,
+    my_model2,
+    my_snapshot,
+)
+
+
+@pytest.fixture(scope="class")
+def macros():
+    return {
+        "assert_selected_models_expected_list.sql": on_run_start_macro_assert_selected_models_expected_list,
+    }
+
+
+@pytest.fixture(scope="class")
+def models():
+    return {"model1.sql": my_model1, "model2.sql": my_model2}
+
+
+@pytest.fixture(scope="class")
+def snapshots():
+    return {
+        "my_snapshot.sql": my_snapshot,
+    }
+
+
+@pytest.fixture(scope="class")
+def project_config_update():
+    return {
+        "on-run-start": "{{ assert_selected_models_expected_list(var('expected_list',None)) }}",
+    }
+
+
+@pytest.fixture
+def build_all(project):
+    run_dbt(["build"])
+
+
+@pytest.mark.usefixtures("build_all")
+class TestSelectedResources:
+    def test_selected_resources_build_selector(self, project):
+        results = run_dbt(
+            [
+                "build",
+                "--select",
+                "model1+",
+                "--vars",
+                '{"expected_list": ["model.test.model1", "model.test.model2", "snapshot.test.cc_all_snapshot"]}',
+            ]
+        )
+        assert results[0].status == "success"
+
+    def test_selected_resources_build_selector_subgraph(self, project):
+        results = run_dbt(
+            [
+                "build",
+                "--select",
+                "model2+",
+                "--vars",
+                '{"expected_list": ["model.test.model2", "snapshot.test.cc_all_snapshot"]}',
+            ]
+        )
+        assert results[0].status == "success"
+
+    def test_selected_resources_run(self, project):
+        results = run_dbt(
+            [
+                "run",
+                "--select",
+                "model1+",
+                "--vars",
+                '{"expected_list": ["model.test.model2", "model.test.model1"]}',
+            ]
+        )
+        assert results[0].status == "success"
+
+    def test_selected_resources_build_no_selector(self, project):
+        results = run_dbt(
+            [
+                "build",
+                "--vars",
+                '{"expected_list": ["model.test.model1", "model.test.model2", "snapshot.test.cc_all_snapshot"]}',
+            ]
+        )
+        assert results[0].status == "success"
+
+    def test_selected_resources_build_no_model(self, project):
+        results = run_dbt(
+            [
+                "build",
+                "--select",
+                "model_that_does_not_exist",
+                "--vars",
+                '{"expected_list": []}',
+            ]
+        )
+        assert not results
+
+    def test_selected_resources_test_no_model(self, project):
+        results = run_dbt(["test", "--select", "model1+", "--vars", '{"expected_list": []}'])
+        assert not results