fix(graph): forward RenderConfig.exclude to test tasks for all test behaviors (#2850)

## Description

`RenderConfig.exclude` was only applied to the **aggregate test task**
created for `TestBehavior.AFTER_ALL`. For `AFTER_EACH` / `BUILD`, the
per-model and detached (multiple-parents) test tasks ignored it — so an
exclusion such as `exclude=["resource_type:unit_test"]` did **not** stop
unit tests from running, and users had to repeat the exclusion in
`operator_args` (e.g. `dbt_cmd_flags`) as a workaround.

### Fix
- `create_test_task_metadata()` now forwards `render_config.exclude` in
the **node** (per-model / detached) branch too, via a small
`forward_render_exclude_to_test()` helper.
- `generate_task_or_group()` passes `render_config` through to the
`AFTER_EACH` test task (it previously wasn't, which is why the node
branch never saw the exclusion).

Exclusions are purely additive, so forwarding them is safe. `select` /
`selector` are **intentionally** still only forwarded for `AFTER_ALL` —
combining a global selection with the per-model selection would require
resolving dbt's union/intersection set semantics, which Cosmos does not
attempt. This limitation is now documented in `selecting-excluding.rst`.

### Behavior (validated by the new integration test,
`MULTIPLE_PARENTS_TEST_DBT_PROJECT` with
`exclude=["resource_type:unit_test"]`)
| Test task | `--exclude` before | `--exclude` after |
|---|---|---|
| `combined_model.test` (parent of detached test) | `custom_test_…` |
`resource_type:unit_test custom_test_…` |
| `model_a.test` (parent) | `custom_test_…` | `resource_type:unit_test
custom_test_…` |
| `model_b.test` (not a parent) | _(none)_ | `resource_type:unit_test` |
| detached multi-parent test | _(none)_ | `resource_type:unit_test` |

## Related Issue(s)

Closes #1763
Relates to #1865 (already closed)

## Context

This is a smaller **re-implementation of #2006** (by @anyapriya, which
went stale) against the **current, refactored** graph code. The two
pieces #2006 carried — the `detached_from_parent: dict[str,
list[DbtNode]]` type-hint fix and avoiding render-config mutation — are
already on `main` (the latter via the `_convert_list_to_str` helper), so
this PR only needs the exclude-forwarding. Credit to @anyapriya for the
original fix and tests, which this preserves.

## Breaking Change?

No.

## Checklist

- [x] I have made corresponding changes to the documentation
- [x] I have added tests that prove my fix is effective (unit tests in
`tests/airflow/test_graph.py`; integration test in
`tests/test_converter.py`)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: tatiana

cosmos/airflow/graph.py

@@ -128,6 +128,15 @@ def calculate_leaves(tasks_ids: list[str], nodes: dict[str, DbtNode]) -> list[st
     return leaves
 
 
+def _split_exclude(value: list[str] | str | None) -> list[str]:
+    """Normalize an ``exclude`` value (space-separated str, list, or None) to a list of items."""
+    if isinstance(value, str):
+        return value.split()
+    if isinstance(value, list):
+        return list(value)
+    return []
+
+
 def exclude_detached_tests_if_needed(
     node: DbtNode,
     task_args: dict[str, str],
@@ -140,21 +149,32 @@ def exclude_detached_tests_if_needed(
     """
     if detached_from_parent is None:
         detached_from_parent = {}
-    current_exclude = task_args.get("exclude")
-    # Handle both list[str] (legacy) and str formats for backward compatibility
-    if isinstance(current_exclude, list):
-        exclude_items = current_exclude
-    elif isinstance(current_exclude, str):
-        exclude_items = current_exclude.split() if current_exclude else []
-    else:
-        exclude_items = []
+    exclude_items = _split_exclude(task_args.get("exclude"))
     tests_detached_from_this_node: list[DbtNode] = detached_from_parent.get(node.unique_id, [])
     for test_node in tests_detached_from_this_node:
         exclude_items.append(test_node.resource_name.split(".")[0])
     if exclude_items:
         task_args["exclude"] = _convert_list_to_str(exclude_items) or ""
 
 
+def forward_render_exclude_to_test(task_args: dict[str, Any], render_config: RenderConfig | None) -> None:
+    """
+    Forward ``RenderConfig.exclude`` to a node command that runs tests, in-place — a per-model
+    (AFTER_EACH / detached) test task, or the inline ``dbt build`` command under TestBehavior.BUILD.
+
+    Exclusions are purely additive: the render-level excludes are unioned with any exclude already
+    present in ``task_args`` (e.g. supplied via ``operator_args``), preserving both. ``select`` /
+    ``selector`` are intentionally not forwarded here.
+    See https://github.com/astronomer/astronomer-cosmos/issues/1763.
+    """
+    if render_config is None or not render_config.exclude:
+        return
+    existing_items = _split_exclude(task_args.get("exclude"))
+    render_items = _split_exclude(render_config.exclude)
+    merged = existing_items + [item for item in render_items if item not in existing_items]
+    task_args["exclude"] = _convert_list_to_str(merged)
+
+
 def _override_profile_if_needed(task_kwargs: dict[str, Any], profile_kwargs_override: dict[str, Any]) -> None:
     """
     Changes in-place the profile configuration if it needs to be overridden.
@@ -189,8 +209,10 @@ def create_test_task_metadata(
     :param execution_mode: The Cosmos execution mode we're aiming to run the dbt task at (e.g. local)
     :param task_args: Arguments to be used to instantiate an Airflow Task
     :param on_warning_callback: A callback function called on warnings with additional Context variables “test_names”
-    and “test_results” of type List.
+        and “test_results” of type List.
     :param node: If the test relates to a specific node, the node reference
+    :param render_config: The RenderConfig for the dbt project. Its ``exclude`` is forwarded to the test task (for
+        every test behavior); its ``select`` / ``selector`` are applied only to the TestBehavior.AFTER_ALL task.
     :param detached_from_parent: Dictionary that maps node ids and their children tests that should be run detached
     :returns: The metadata necessary to instantiate the source dbt node as an Airflow task.
     """
@@ -212,6 +234,8 @@ def create_test_task_metadata(
         else:  # tested with node.resource_type == DbtResourceType.SEED or DbtResourceType.SNAPSHOT
             task_args["select"] = node.resource_name
 
+        forward_render_exclude_to_test(task_args, render_config)
+
         extra_context = {"dbt_node_config": node.context_dict}
         task_owner = node.owner
     elif render_config is not None:  # TestBehavior.AFTER_ALL
@@ -430,6 +454,10 @@ def create_task_metadata(  # noqa: C901
             if test_indirect_selection != TestIndirectSelection.EAGER:
                 args["indirect_selection"] = test_indirect_selection.value
             args["on_warning_callback"] = on_warning_callback
+            # Under BUILD, tests run inline with ``dbt build`` (there is no separate per-model test task),
+            # so forward the render-level exclude here too — otherwise an exclusion such as
+            # exclude=["resource_type:unit_test"] would not be honored for BUILD. See #1763.
+            forward_render_exclude_to_test(args, render_config)
             exclude_detached_tests_if_needed(node, args, detached_from_parent)
             task_id, args = _get_task_id_and_args(
                 node=node,
@@ -678,6 +706,7 @@ def generate_task_or_group(
                     test_indirect_selection,
                     task_args=task_args,
                     node=node,
+                    render_config=render_config,
                     on_warning_callback=on_warning_callback,
                     detached_from_parent=detached_from_parent,
                     enable_owner_inheritance=render_config.enable_owner_inheritance,

docs/guides/translate_dbt_to_airflow/selecting-excluding.rst

@@ -145,6 +145,34 @@ Examples:
         )
     )
 
+How ``exclude`` and ``select`` interact with test tasks
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+``RenderConfig.exclude`` is passed through to the generated **test tasks for every test
+behavior** (``AFTER_EACH``, ``BUILD`` and ``AFTER_ALL``). This means an exclusion such as
+``exclude=["resource_type:unit_test"]`` is applied to the ``dbt test`` commands too, so you can
+exclude a specific test or a whole resource type without also having to repeat it in
+``operator_args``. Exclusions are purely additive, which makes this safe to forward.
+
+.. code-block:: python
+
+    from cosmos.airflow.dag import DbtDag
+    from cosmos.config import RenderConfig
+
+    jaffle_shop = DbtDag(
+        render_config=RenderConfig(
+            exclude=[
+                "resource_type:unit_test"
+            ],  # excluded from both the run and the test tasks
+        )
+    )
+
+``RenderConfig.select`` / ``RenderConfig.selector``, on the other hand, are forwarded to the test
+task **only** for ``TestBehavior.AFTER_ALL``. They are intentionally not forwarded to the per-model
+``AFTER_EACH`` / ``BUILD`` test tasks, because combining a global selection with the per-model
+selection would require resolving dbt's union / intersection set semantics, which Cosmos does not
+attempt. Those test tasks are already scoped to their own model's selection.
+
 Using ``selector``
 ~~~~~~~~~~~~~~~~~~
 .. note::

tests/airflow/test_graph.py

@@ -827,6 +827,30 @@ def test_create_task_metadata_ephemeral_model_disabled_renders_dbt_build_in_buil
     assert metadata.operator_class == "cosmos.operators.local.DbtBuildLocalOperator"
 
 
+def test_create_task_metadata_build_mode_forwards_render_config_exclude():
+    """Under TestBehavior.BUILD, tests run inline with `dbt build`, so RenderConfig.exclude must be
+    forwarded to the build command — otherwise e.g. exclude=["resource_type:unit_test"] would not
+    take effect for BUILD. See https://github.com/astronomer/astronomer-cosmos/issues/1763."""
+    model_node = DbtNode(
+        unique_id=f"{DbtResourceType.MODEL.value}.my_project.model_a",
+        resource_type=DbtResourceType.MODEL,
+        depends_on=[],
+        path_base=Path("base_path"),
+        original_file_path=Path("models/model_a.sql"),
+        fqn=["my_project", "model_a"],
+    )
+    metadata = create_task_metadata(
+        model_node,
+        execution_mode=ExecutionMode.LOCAL,
+        args={},
+        dbt_dag_task_group_identifier="",
+        render_config=RenderConfig(test_behavior=TestBehavior.BUILD, exclude=["resource_type:unit_test"]),
+    )
+    assert metadata.operator_class == "cosmos.operators.local.DbtBuildLocalOperator"
+    assert metadata.arguments["select"] == "fqn:my_project.model_a"
+    assert metadata.arguments["exclude"] == "resource_type:unit_test"
+
+
 def test_create_task_metadata_ephemeral_empty_operator_inherits_owner():
     """The ephemeral EmptyOperator inherits the dbt model owner when owner inheritance is enabled (default)."""
     metadata = create_task_metadata(
@@ -1313,6 +1337,106 @@ def test_create_test_task_metadata(node_type, node_unique_id, test_indirect_sele
     }
 
 
+@pytest.mark.parametrize(
+    "exclude,expected_exclude",
+    [
+        (["resource_type:unit_test"], "resource_type:unit_test"),
+        (["resource_type:unit_test", "tag:skip_in_tests"], "resource_type:unit_test tag:skip_in_tests"),
+    ],
+)
+def test_create_test_task_metadata_forwards_render_config_exclude_to_node_test(exclude, expected_exclude):
+    """RenderConfig.exclude must reach per-model (AFTER_EACH) / detached test tasks, not only AFTER_ALL.
+
+    See https://github.com/astronomer/astronomer-cosmos/issues/1763.
+    """
+    sample_node = DbtNode(
+        unique_id=f"{DbtResourceType.MODEL.value}.my_folder.node_name",
+        resource_type=DbtResourceType.MODEL,
+        depends_on=[],
+        path_base=Path("."),
+        original_file_path=Path("."),
+        tags=[],
+        config={},
+    )
+    metadata = create_test_task_metadata(
+        test_task_name="test",
+        execution_mode=ExecutionMode.LOCAL,
+        test_indirect_selection=TestIndirectSelection.EAGER,
+        task_args={"task_arg": "value"},
+        node=sample_node,
+        render_config=RenderConfig(exclude=exclude),
+    )
+    assert metadata.arguments["select"] == "node_name"
+    assert metadata.arguments["exclude"] == expected_exclude
+
+
+def test_create_test_task_metadata_without_render_config_exclude_preserves_existing_exclude():
+    """An empty RenderConfig.exclude must not clobber an exclude supplied through operator/task args."""
+    sample_node = DbtNode(
+        unique_id=f"{DbtResourceType.MODEL.value}.my_folder.node_name",
+        resource_type=DbtResourceType.MODEL,
+        depends_on=[],
+        path_base=Path("."),
+        original_file_path=Path("."),
+        tags=[],
+        config={},
+    )
+    metadata = create_test_task_metadata(
+        test_task_name="test",
+        execution_mode=ExecutionMode.LOCAL,
+        test_indirect_selection=TestIndirectSelection.EAGER,
+        task_args={"exclude": "tag:my_custom_exclude"},
+        node=sample_node,
+        render_config=RenderConfig(exclude=[]),
+    )
+    assert metadata.arguments["exclude"] == "tag:my_custom_exclude"
+
+
+def test_create_test_task_metadata_unions_render_config_exclude_with_existing_exclude():
+    """A render-level exclude must be unioned with (not overwrite) an operator/task-args exclude."""
+    sample_node = DbtNode(
+        unique_id=f"{DbtResourceType.MODEL.value}.my_folder.node_name",
+        resource_type=DbtResourceType.MODEL,
+        depends_on=[],
+        path_base=Path("."),
+        original_file_path=Path("."),
+        tags=[],
+        config={},
+    )
+    metadata = create_test_task_metadata(
+        test_task_name="test",
+        execution_mode=ExecutionMode.LOCAL,
+        test_indirect_selection=TestIndirectSelection.EAGER,
+        task_args={"exclude": "tag:foo"},
+        node=sample_node,
+        render_config=RenderConfig(exclude=["resource_type:unit_test"]),
+    )
+    # Both the operator-supplied and render-level exclusions are preserved (additive).
+    assert metadata.arguments["exclude"] == "tag:foo resource_type:unit_test"
+
+
+def test_create_test_task_metadata_does_not_duplicate_overlapping_excludes():
+    """An exclude present in both task args and RenderConfig must appear only once."""
+    sample_node = DbtNode(
+        unique_id=f"{DbtResourceType.MODEL.value}.my_folder.node_name",
+        resource_type=DbtResourceType.MODEL,
+        depends_on=[],
+        path_base=Path("."),
+        original_file_path=Path("."),
+        tags=[],
+        config={},
+    )
+    metadata = create_test_task_metadata(
+        test_task_name="test",
+        execution_mode=ExecutionMode.LOCAL,
+        test_indirect_selection=TestIndirectSelection.EAGER,
+        task_args={"exclude": "resource_type:unit_test"},
+        node=sample_node,
+        render_config=RenderConfig(exclude=["resource_type:unit_test", "tag:foo"]),
+    )
+    assert metadata.arguments["exclude"] == "resource_type:unit_test tag:foo"
+
+
 @pytest.mark.parametrize(
     "input,expected", [("snake_case", "SnakeCase"), ("snake_case_with_underscores", "SnakeCaseWithUnderscores")]
 )

tests/test_converter.py

@@ -377,6 +377,75 @@ def test_converter_creates_dag_with_test_with_multiple_parents():
     )
 
 
+@pytest.mark.integration
+def test_converter_passes_render_config_exclude_to_test_tasks():
+    """
+    RenderConfig.exclude must be forwarded to the generated test tasks for every test behavior,
+    not only TestBehavior.AFTER_ALL — i.e. exclude=["resource_type:unit_test"] reaches the
+    ``dbt test`` command of each per-model / detached test task.
+
+    See https://github.com/astronomer/astronomer-cosmos/issues/1763
+    and https://github.com/astronomer/astronomer-cosmos/issues/1865.
+    """
+    project_config = ProjectConfig(dbt_project_path=MULTIPLE_PARENTS_TEST_DBT_PROJECT)
+    execution_config = ExecutionConfig(execution_mode=ExecutionMode.LOCAL)
+    render_config = RenderConfig(should_detach_multiple_parents_tests=True, exclude=["resource_type:unit_test"])
+    profile_config = ProfileConfig(
+        profile_name="default",
+        target_name="dev",
+        profile_mapping=PostgresUserPasswordProfileMapping(
+            conn_id="example_conn",
+            profile_args={"schema": "public"},
+            disable_event_tracking=True,
+        ),
+    )
+    with DAG("sample_dag", start_date=datetime(2024, 4, 16)) as dag:
+        converter = DbtToAirflowConverter(
+            dag=dag,
+            project_config=project_config,
+            profile_config=profile_config,
+            execution_config=execution_config,
+            render_config=render_config,
+        )
+    tasks = converter.tasks_map
+
+    assert len(converter.tasks_map) == 4
+
+    # The render-level exclusion is prepended to each test command; for parents of the detached
+    # test the detached test name is then appended (exclusions are additive).
+    args = tasks["model.my_dbt_project.combined_model"].children["combined_model.test"].build_cmd({})[0]
+    assert args[1:] == [
+        "test",
+        "--select",
+        "combined_model",
+        "--exclude",
+        "resource_type:unit_test custom_test_combined_model_combined_model_",
+    ]
+
+    args = tasks["model.my_dbt_project.model_a"].children["model_a.test"].build_cmd({})[0]
+    assert args[1:] == [
+        "test",
+        "--select",
+        "model_a",
+        "--exclude",
+        "resource_type:unit_test custom_test_combined_model_combined_model_",
+    ]
+
+    # model_b is not a parent of the detached test, so it only carries the render-level exclusion
+    args = tasks["model.my_dbt_project.model_b"].children["model_b.test"].build_cmd({})[0]
+    assert args[1:] == ["test", "--select", "model_b", "--exclude", "resource_type:unit_test"]
+
+    # The detached (multiple-parents) test task also honors the render-level exclusion
+    args = tasks["test.my_dbt_project.custom_test_combined_model_combined_model_.c6e4587380"].build_cmd({})[0]
+    assert args[1:] == [
+        "test",
+        "--select",
+        "custom_test_combined_model_combined_model_",
+        "--exclude",
+        "resource_type:unit_test",
+    ]
+
+
 @pytest.mark.integration
 def test_converter_creates_dag_with_test_with_multiple_parents_with_should_detach_multiple_parents_tests_false():
     """