Fix TestBehavior.AFTER_ALL is missing project_name information when loading project using manifest file

[tuantran0910]

Description

Problem

When using LoadMode.DBT_MANIFEST with project_name set in ProjectConfig but no dbt_project_path in RenderConfig, the test task created with TestBehavior.AFTER_ALL was incorrectly named _test instead of {project_name}_test.

Example DAG:

import datetime
from pathlib import Path

from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG, Asset, Param
from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig, ExecutionConfig, RenderConfig
from cosmos.constants import TestBehavior, ExecutionMode, LoadMode


# Paths configuration
DAGS_DIR = Path("/opt/airflow/dags")
DBT_PROJECT_PATH = DAGS_DIR / "dbt" / "jaffle_shop"
DBT_PROFILES_PATH = DBT_PROJECT_PATH / "profiles.yml"
DBT_MANIFEST_PATH = DBT_PROJECT_PATH / "target" / "manifest.json"
DBT_CATALOG_PATH = DBT_PROJECT_PATH / "target" / "catalog.json"
DBT_RUN_RESULTS_PATH = DBT_PROJECT_PATH / "target" / "run_results.json"


def dbt_datahub_ingestion(**kwargs) -> None:
    """
    Callback to ingest dbt models's metadata into DataHub.
    """
    from datahub.ingestion.run.pipeline import Pipeline
    from datahub_airflow_plugin.hooks.datahub import DatahubRestHook

    datahub_rest_hook = DatahubRestHook.get_connection(conn_id="datahub_rest_default")
    server, token = datahub_rest_hook.host, datahub_rest_hook.password
    pipeline = Pipeline.create(
        {
            "source": {
                "type": "dbt",
                "config": {
                    "manifest_path": str(DBT_MANIFEST_PATH),
                    "catalog_path": str(DBT_CATALOG_PATH),
                    "run_results_paths": [
                        str(DBT_RUN_RESULTS_PATH),
                    ],
                    "target_platform": "postgres",
                },
            },
            "sink": {
                "type": "datahub-rest",
                "config": {
                    "server": server,
                    "token": token,
                    "retry_max_times": 3,
                },
            },
        }
    )

    pipeline.run()
    pipeline.pretty_print_summary()
    pipeline.raise_from_status()


# Default arguments for the DAG
default_args = {'owner': 'data-team', 'depends_on_past': False, 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': datetime.timedelta(seconds=300), 'priority_weight': 1, 'weight_rule': 'downstream', 'queue': 'default', 'pool': 'default_pool', 'email': ['data-team@example.com']}

# Profile configuration using profiles.yml file
profile_config = ProfileConfig(
    profile_name="jaffle_shop",
    target_name="production",
    profiles_yml_filepath=DBT_PROFILES_PATH,
)

# Execution configuration
execution_config = ExecutionConfig(
    dbt_project_path=DBT_PROJECT_PATH,
    execution_mode=ExecutionMode.LOCAL
)

# Render configuration - select which models to run
render_config = RenderConfig(
    select=['path:models/staging', 'path:models/intermediate', 'path:models/marts'],
    emit_datasets=True,
    test_behavior=TestBehavior.AFTER_ALL,
    load_method=LoadMode.DBT_MANIFEST,
)

# Operator arguments - full_refresh is templated to use DAG params at runtime
operator_args = {
    # Use Jinja templating to read full_refresh from DAG params at runtime
    "full_refresh": "{{ params.full_refresh }}",
}

schedule="@daily"

# Create the DAG with full_refresh as a runtime parameter
dag = DAG(
    dag_id="dbt_jaffle_shop_core",
    description="Run all dbt transformations for jaffle_shop (4-level lineage)",
    default_args=default_args,
    schedule=schedule,
    start_date=datetime.datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['dbt', 'jaffle_shop', 'full-pipeline', 'data-lineage'],
    params={
        "full_refresh": Param(
            default=False,
            type="boolean",
            description="Run dbt with --full-refresh flag to rebuild all incremental models",
        ),
    },
)

outlet_asset = Asset("dbt://dbt_jaffle_shop_core")

with dag:
    dbt_tasks = DbtTaskGroup(
        group_id="dbt_jaffle_shop",
        project_config=ProjectConfig(
            manifest_path=DBT_MANIFEST_PATH,
            project_name="jaffle_shop",
        ),
        profile_config=profile_config,
        execution_config=execution_config,
        render_config=render_config,
        operator_args=operator_args,
    )

    ingest_datahub_metadata = PythonOperator(
        task_id="ingest_datahub_metadata",
        python_callable=dbt_datahub_ingestion,
        trigger_rule="all_done",
    )

    emit_assets = EmptyOperator(
        task_id="emit_assets",
        outlets=[outlet_asset],
        trigger_rule="all_done",
    )

    dbt_tasks >> [ingest_datahub_metadata, emit_assets]

Solution

As we don't specify the parameter dbt_project_path of the class ProjectConfig, we have to specify ProjectConfig.manifest_path and ProjectConfig.project_name. Therefore, we can make use of the ProjectConfig.project_name.

Related Issue(s)

Close #1815

Additional Information

Before:

After:

[netlify]

✅ Deploy Preview for astronomer-cosmos canceled.

Name	Link
🔨 Latest commit	e76b0e5
🔍 Latest deploy log	https://app.netlify.com/projects/astronomer-cosmos/deploys/6959f3b094c2c70008ede892

[Copilot]

Pull request overview

This PR fixes a bug where test tasks created with TestBehavior.AFTER_ALL were incorrectly named _test instead of {project_name}_test when using LoadMode.DBT_MANIFEST with project_name set in ProjectConfig but no dbt_project_path in RenderConfig.

Key Changes:

• Added fallback logic to use project_config.project_name when render_config.project_name is empty
• Added comprehensive test coverage for the manifest loading scenario with explicit project_name

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
cosmos/converter.py	Modified dbt_project_name parameter to fall back to project_config.project_name when render_config.project_name is empty
tests/test_converter.py	Added integration test to verify test task naming uses project_config.project_name when loading from manifest without dbt_project_path in RenderConfig

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 97.99%. Comparing base (d58e0aa) to head (e76b0e5).
⚠️ Report is 8 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2242   +/-   ##
=======================================
  Coverage   97.99%   97.99%           
=======================================
  Files          96       96           
  Lines        6220     6220           
=======================================
  Hits         6095     6095           
  Misses        125      125           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[tatiana]

Thanks a lot for fixing this, @tuantran0910 !