Added Explicit failure when project path is not defined, updated test…

…s to reflect

cosmos/config.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import contextlib
-import os
 import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -224,4 +223,3 @@ class ExecutionConfig:
     execution_mode: ExecutionMode = ExecutionMode.LOCAL
     test_indirect_selection: TestIndirectSelection = TestIndirectSelection.EAGER
     dbt_executable_path: str | Path = get_system_dbt()
-    dbt_project_path_root: str | Path = Path(os.environ.get("AIRFLOW_HOME", "/dbt")) / "dags" / "dbt"

cosmos/converter.py

@@ -112,9 +112,11 @@ def __init__(
         test_indirect_selection = execution_config.test_indirect_selection
         load_mode = render_config.load_method
         dbt_executable_path = execution_config.dbt_executable_path
-        dbt_project_path_runtime_root = execution_config.dbt_project_path_root
         node_converters = render_config.node_converters
 
+        if not project_config.dbt_project_path:
+            raise CosmosValueError("A Project Path in ProjectConfig is required for generating a Task Operators.")
+
         profile_args = {}
         if profile_config.profile_mapping:
             profile_args = profile_config.profile_mapping.profile_args
@@ -146,9 +148,7 @@ def __init__(
         task_args = {
             **operator_args,
             # the following args may be only needed for local / venv:
-            "project_dir": project_config.dbt_project_path
-            if project_config.dbt_project_path
-            else dbt_project_path_runtime_root / project_config.project_name,
+            "project_dir": project_config.dbt_project_path,
             "profile_config": profile_config,
             "emit_datasets": emit_datasets,
         }

dev/dags/cosmos_manifest_example.py

@@ -25,7 +25,9 @@
 cosmos_manifest_example = DbtDag(
     # dbt/cosmos-specific parameters
     project_config=ProjectConfig(
-        manifest_path=DBT_ROOT_PATH / "jaffle_shop" / "target" / "manifest.json", project_name="jaffle_shop"
+        dbt_project_path=DBT_ROOT_PATH / "jaffle_shop",
+        manifest_path=DBT_ROOT_PATH / "jaffle_shop" / "target" / "manifest.json",
+        project_name="jaffle_shop",
     ),
     profile_config=profile_config,
     render_config=RenderConfig(load_method=LoadMode.DBT_MANIFEST, select=["path:models/customers.sql"]),

tests/test_converter.py

@@ -113,7 +113,7 @@ def test_converter_creates_dag_with_project_path_str(mock_load_dbt_graph, execut
 )
 @patch("cosmos.converter.DbtGraph.filtered_nodes", nodes)
 @patch("cosmos.converter.DbtGraph.load")
-def test_converter_creates_dag_with_no_project_dir(mock_load_dbt_graph, execution_mode, operator_args):
+def test_converter_fails_no_project_dir(mock_load_dbt_graph, execution_mode, operator_args):
     """
     This test validates that a project, given a manifest path and project name, with seeds
     is able to successfully generate a converter
@@ -126,12 +126,13 @@ def test_converter_creates_dag_with_no_project_dir(mock_load_dbt_graph, executio
         target_name="my_target_name",
         profiles_yml_filepath=SAMPLE_PROFILE_YML,
     )
-    converter = DbtToAirflowConverter(
-        nodes=nodes,
-        project_config=project_config,
-        profile_config=profile_config,
-        execution_config=execution_config,
-        render_config=render_config,
-        operator_args=operator_args,
-    )
-    assert converter
+    with pytest.raises(CosmosValueError) as err_info:
+        DbtToAirflowConverter(
+            nodes=nodes,
+            project_config=project_config,
+            profile_config=profile_config,
+            execution_config=execution_config,
+            render_config=render_config,
+            operator_args=operator_args,
+        )
+        assert err_info.value.args[0] == "A Project Path in ProjectConfig is required for generating a Task Operators."