Improve test coverage

tests/operators/test_local.py

@@ -470,6 +470,7 @@ def test_run_operator_dataset_inlets_and_outlets_airflow_210_onwards(caplog):
             project_dir=DBT_PROJ_DIR,
             task_id="seed",
             dag=dag,
+            emit_datasets=False,
             dbt_cmd_flags=["--select", "raw_customers"],
             install_deps=True,
             append_env=True,
@@ -494,7 +495,7 @@ def test_run_operator_dataset_inlets_and_outlets_airflow_210_onwards(caplog):
         )
         seed_operator >> run_operator >> test_operator
 
-    assert seed_operator.outlets == [DatasetAliasModel(name="test_id_1__seed")]
+    assert seed_operator.outlets == []  # because emit_datasets=False,
     assert run_operator.outlets == [DatasetAliasModel(name="test_id_1__run")]
     assert test_operator.outlets == [DatasetAliasModel(name="test_id_1__test")]
 
@@ -510,6 +511,26 @@ def test_run_operator_dataset_inlets_and_outlets_airflow_210_onwards(caplog):
         # assert dataset_model == 1
 
 
+@patch("cosmos.settings.enable_dataset_alias", 0)
+@pytest.mark.skipif(
+    version.parse(airflow_version) < version.parse("2.10"),
+    reason="From Airflow 2.10 onwards, we started using DatasetAlias, which changed this behaviour.",
+)
+@pytest.mark.integration
+def test_run_operator_dataset_inlets_and_outlets_airflow_210_onwards_disabled_via_envvar(caplog):
+    with DAG("test_id_2", start_date=datetime(2022, 1, 1)) as dag:
+        run_operator = DbtRunLocalOperator(
+            profile_config=real_profile_config,
+            project_dir=DBT_PROJ_DIR,
+            task_id="run",
+            dag=dag,
+            dbt_cmd_flags=["--models", "stg_customers"],
+            install_deps=True,
+            append_env=True,
+        )
+    assert run_operator.outlets == []
+
+
 @pytest.mark.skipif(
     version.parse(airflow_version) not in PARTIALLY_SUPPORTED_AIRFLOW_VERSIONS,
     reason="Airflow 2.9.0 and 2.9.1 have a breaking change in Dataset URIs",

tests/test_dataset.py

@@ -0,0 +1,39 @@
+from datetime import datetime
+
+import pytest
+from airflow import DAG
+from airflow.utils.task_group import TaskGroup
+
+from cosmos.dataset import get_dataset_alias_name
+
+START_DATE = datetime(2024, 4, 16)
+example_dag = DAG("dag", start_date=START_DATE)
+
+
+@pytest.mark.parametrize(
+    "dag, task_group, result_identifier",
+    [
+        (example_dag, None, "dag__task_id"),
+        (None, TaskGroup(dag=example_dag, group_id="inner_tg"), "dag__inner_tg__task_id"),
+        (
+            None,
+            TaskGroup(
+                dag=example_dag, group_id="child_tg", parent_group=TaskGroup(dag=example_dag, group_id="parent_tg")
+            ),
+            "dag__parent_tg__child_tg__task_id",
+        ),
+        (
+            None,
+            TaskGroup(
+                dag=example_dag,
+                group_id="child_tg",
+                parent_group=TaskGroup(
+                    dag=example_dag, group_id="mum_tg", parent_group=TaskGroup(dag=example_dag, group_id="nana_tg")
+                ),
+            ),
+            "dag__nana_tg__mum_tg__child_tg__task_id",
+        ),
+    ],
+)
+def test_get_dataset_alias_name(dag, task_group, result_identifier):
+    assert get_dataset_alias_name(dag, task_group, "task_id") == result_identifier